23952851253

Committed 03 Apr 2026 04:07PM UTC coverage: 80.56% (+0.05%) from 80.508%

Build # 23952851253

Build Type

Pull #965

github

Committed by

web-flow

Commit Message

Merge d2bf1979a into 55b13d856

Pull Request Pull Request #965: fix: report errors in orphan and test entity expansion

Coverage Stats

3146 of 4081 branches covered (77.09%)

Branch coverage included in aggregate %.

123 of 152 new or added lines in 5 files covered. (80.92%)

4 existing lines in 1 file now uncovered.

30047 of 37122 relevant lines covered (80.94%)

311.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.36

/src/scripts/orphan-client-stream.ts

import fs from 'fs'
import { mkdir } from 'fs/promises'
import path from 'path'
import { pipeline } from 'stream/promises'

export interface OrphanObject {
  event: 'data'
  type: 'dbOrphans' | 's3Orphans'
  value: {
    name: string
    version: string
    size: number
  }[]
}

export interface PingObject {
  event: 'ping'
}

export interface StreamedErrorPayload {
  statusCode: string
  code: string
  error: string
  message: string
}

export interface ErrorObject {
  event: 'error'
  error: StreamedErrorPayload
}

export type OrphanStreamEvent = OrphanObject | PingObject | ErrorObject

export function formatOrphanStreamError(error: StreamedErrorPayload) {
  return `[${error.code}] ${error.message}`
}

export async function writeStreamToJsonArray(
  stream: NodeJS.ReadableStream,
  filePath: string
): Promise<void> {
  await mkdir(path.dirname(filePath), { recursive: true })

  const localFile = fs.createWriteStream(filePath)
  let isFirstItem = true
  let receivedAnyData = false
  let streamedError: Error | undefined
  let deleteLimitReached = false
  let inputStreamError: Error | undefined

  const jsonArrayStream = (async function* () {
    yield '[\n'

    try {
      for await (const data of stream as AsyncIterable<OrphanStreamEvent>) {
        if (data.event === 'ping') {
          console.log('Received ping event, ignoring')
          continue
        }

        if (data.event === 'error') {
          streamedError = new Error(formatOrphanStreamError(data.error))
          console.error('Server error:', formatOrphanStreamError(data.error))
          continue
        }

        if (data.value && Array.isArray(data.value)) {
          receivedAnyData = true
          console.log(`Processing ${data.value.length} objects`)

          for (const item of data.value) {
            if (!isFirstItem) {
              yield ',\n'
            } else {
              isFirstItem = false
            }

            yield JSON.stringify({ ...item, orphanType: data.type }, null, 2)
          }
          continue
        }

        console.warn(
          'Received data with invalid format:',
          JSON.stringify(data).substring(0, 100) + '...'
        )
      }
    } catch (err) {
      if (err instanceof Error && err.message === 'DELETE_LIMIT_REACHED') {
        deleteLimitReached = true
      } else {
        inputStreamError =
          err instanceof Error ? err : new Error('Unexpected stream failure', { cause: err })
        console.error('Stream error:', inputStreamError)
      }
    }

    yield '\n]'
  })()

  await pipeline(jsonArrayStream, localFile)

  if (inputStreamError) {
    throw inputStreamError
  }

  if (streamedError) {
    throw streamedError
  }

  if (!receivedAnyData) {
    console.warn(`No data was received! File might be empty: ${filePath}`)
    return
  }

  if (deleteLimitReached) {
    console.log(`Finished writing data to ${filePath}. Delete limit reached, data saved.`)
    return
  }

  console.log(`Finished writing data to ${filePath}. Data was received and saved.`)
}

1	import fs from 'fs'	1✔
2	import { mkdir } from 'fs/promises'	1✔
3	import path from 'path'	1✔
4	import { pipeline } from 'stream/promises'	1✔
5		1✔
6	export interface OrphanObject {	1✔
7	event: 'data'	1✔
8	type: 'dbOrphans' \| 's3Orphans'	1✔
9	value: {	1✔
10	name: string	1✔
11	version: string	1✔
12	size: number	1✔
13	}[]	1✔
14	}	1✔
15		1✔
16	export interface PingObject {	1✔
17	event: 'ping'	1✔
18	}	1✔
19		1✔
20	export interface StreamedErrorPayload {	1✔
21	statusCode: string	1✔
22	code: string	1✔
23	error: string	1✔
24	message: string	1✔
25	}	1✔
26		1✔
27	export interface ErrorObject {	1✔
28	event: 'error'	1✔
29	error: StreamedErrorPayload	1✔
30	}	1✔
31		1✔
32	export type OrphanStreamEvent = OrphanObject \| PingObject \| ErrorObject	1✔
33		1✔
34	export function formatOrphanStreamError(error: StreamedErrorPayload) {	2✔
35	return `[${error.code}] ${error.message}`	2✔
36	}	2✔
37		1✔
38	export async function writeStreamToJsonArray(	1✔
39	stream: NodeJS.ReadableStream,	6✔
40	filePath: string	6✔
41	): Promise<void> {	6✔
42	await mkdir(path.dirname(filePath), { recursive: true })	6✔
43		6✔
44	const localFile = fs.createWriteStream(filePath)	6✔
45	let isFirstItem = true	6✔
46	let receivedAnyData = false	6✔
47	let streamedError: Error \| undefined	6✔
48	let deleteLimitReached = false	6✔
49	let inputStreamError: Error \| undefined	6✔
50		6✔
51	const jsonArrayStream = (async function* () {	6✔
52	yield '[\n'	6✔
53		6✔
54	try {	6✔
55	for await (const data of stream as AsyncIterable<OrphanStreamEvent>) {	6✔
56	if (data.event === 'ping') {	5!
NEW 57	console.log('Received ping event, ignoring')	×
NEW 58	continue	×
NEW 59	}	×
60		5✔
61	if (data.event === 'error') {	5✔
62	streamedError = new Error(formatOrphanStreamError(data.error))	1✔
63	console.error('Server error:', formatOrphanStreamError(data.error))	1✔
64	continue	1✔
65	}	1✔
66		4✔
67	if (data.value && Array.isArray(data.value)) {	5✔
68	receivedAnyData = true	4✔
69	console.log(`Processing ${data.value.length} objects`)	4✔
70		4✔
71	for (const item of data.value) {	4✔
72	if (!isFirstItem) {	4!
NEW 73	yield ',\n'	×
74	} else {	4✔
75	isFirstItem = false	4✔
76	}	4✔
77		4✔
78	yield JSON.stringify({ ...item, orphanType: data.type }, null, 2)	4✔
79	}	4✔
80	continue	4✔
81	}	4✔
NEW 82		×
NEW 83	console.warn(	×
NEW 84	'Received data with invalid format:',	×
NEW 85	JSON.stringify(data).substring(0, 100) + '...'	×
NEW 86	)	×
NEW 87	}	×
88	} catch (err) {	3✔
89	if (err instanceof Error && err.message === 'DELETE_LIMIT_REACHED') {	3✔
90	deleteLimitReached = true	1✔
91	} else {	3✔
92	inputStreamError =	2✔
93	err instanceof Error ? err : new Error('Unexpected stream failure', { cause: err })	2✔
94	console.error('Stream error:', inputStreamError)	2✔
95	}	2✔
96	}	3✔
97		6✔
98	yield '\n]'	6✔
99	})()	6✔
100		6✔
101	await pipeline(jsonArrayStream, localFile)	6✔
102		5✔
103	if (inputStreamError) {	6✔
104	throw inputStreamError	2✔
105	}	2✔
106		3✔
107	if (streamedError) {	6✔
108	throw streamedError	1✔
109	}	1✔
110		2✔
111	if (!receivedAnyData) {	6!
NEW 112	console.warn(`No data was received! File might be empty: ${filePath}`)	×
NEW 113	return	×
NEW 114	}	×
115		2✔
116	if (deleteLimitReached) {	6✔
117	console.log(`Finished writing data to ${filePath}. Delete limit reached, data saved.`)	1✔
118	return	1✔
119	}	1✔
120		1✔
121	console.log(`Finished writing data to ${filePath}. Data was received and saved.`)	1✔
122	}	1✔

supabase / storage / 23952851253

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous