24025892439

Committed 06 Apr 2026 09:04AM UTC coverage: 80.536% (+0.02%) from 80.515%

Build # 24025892439

Build Type

push

github

Committed by

web-flow

Commit Message

fix: report errors in orphan and test entity expansion (#965)

Coverage Stats

3145 of 4079 branches covered (77.1%)

Branch coverage included in aggregate %.

135 of 188 new or added lines in 5 files covered. (71.81%)

3 existing lines in 1 file now uncovered.

30055 of 37145 relevant lines covered (80.91%)

311.64 hits per line

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

95.95

/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.event === 'data' && 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,	7✔
40	filePath: string	7✔
41	): Promise<void> {	7✔
42	await mkdir(path.dirname(filePath), { recursive: true })	7✔
43		7✔
44	const localFile = fs.createWriteStream(filePath)	7✔
45	let isFirstItem = true	7✔
46	let receivedAnyData = false	7✔
47	let streamedError: Error \| undefined	7✔
48	let deleteLimitReached = false	7✔
49	let inputStreamError: Error \| undefined	7✔
50		7✔
51	const jsonArrayStream = (async function* () {	7✔
52	yield '[\n'	7✔
53		7✔
54	try {	7✔
55	for await (const data of stream as AsyncIterable<OrphanStreamEvent>) {	7✔
56	if (data.event === 'ping') {	6!
NEW 57	console.log('Received ping event, ignoring')	×
NEW 58	continue	×
NEW 59	}	×
60		6✔
61	if (data.event === 'error') {	6✔
62	streamedError = new Error(formatOrphanStreamError(data.error))	1✔
63	console.error('Server error:', formatOrphanStreamError(data.error))	1✔
64	continue	1✔
65	}	1✔
66		5✔
67	if (data.event === 'data' && Array.isArray(data.value)) {	6✔
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✔
82		1✔
83	console.warn(	1✔
84	'Received data with invalid format:',	1✔
85	JSON.stringify(data).substring(0, 100) + '...'	1✔
86	)	1✔
87	}	1✔
88	} catch (err) {	7✔
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		7✔
98	yield '\n]'	7✔
99	})()	7✔
100		7✔
101	await pipeline(jsonArrayStream, localFile)	7✔
102		6✔
103	if (inputStreamError) {	7✔
104	throw inputStreamError	2✔
105	}	2✔
106		4✔
107	if (streamedError) {	7✔
108	throw streamedError	1✔
109	}	1✔
110		3✔
111	if (!receivedAnyData) {	7✔
112	console.warn(`No data was received! File might be empty: ${filePath}`)	1✔
113	return	1✔
114	}	1✔
115		2✔
116	if (deleteLimitReached) {	7✔
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 / 24025892439

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