23952459316

Committed 03 Apr 2026 03:54PM UTC coverage: 80.534% (+0.03%) from 80.508%

Build # 23952459316

Build Type

Pull #965

github

Committed by

web-flow

Commit Message

Merge 5cfcf5c2d into 55b13d856

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

Coverage Stats

3139 of 4077 branches covered (76.99%)

Branch coverage included in aggregate %.

124 of 153 new or added lines in 5 files covered. (81.05%)

7 existing lines in 1 file now uncovered.

30041 of 37123 relevant lines covered (80.92%)

311.62 hits per line

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

88.11

/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 inputStreamFailed = false
  let inputStreamError: unknown

  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 {
        inputStreamFailed = true
        inputStreamError = err
        console.error('Stream error:', err)
      }
    }

    yield '\n]'
  })()

  await pipeline(jsonArrayStream, localFile)

  if (inputStreamFailed) {
    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,	5✔
40	filePath: string	5✔
41	): Promise<void> {	5✔
42	await mkdir(path.dirname(filePath), { recursive: true })	5✔
43		5✔
44	const localFile = fs.createWriteStream(filePath)	5✔
45	let isFirstItem = true	5✔
46	let receivedAnyData = false	5✔
47	let streamedError: Error \| undefined	5✔
48	let deleteLimitReached = false	5✔
49	let inputStreamFailed = false	5✔
50	let inputStreamError: unknown	5✔
51		5✔
52	const jsonArrayStream = (async function* () {	5✔
53	yield '[\n'	5✔
54		5✔
55	try {	5✔
56	for await (const data of stream as AsyncIterable<OrphanStreamEvent>) {	5✔
57	if (data.event === 'ping') {	5!
NEW 58	console.log('Received ping event, ignoring')	×
NEW 59	continue	×
NEW 60	}	×
61		5✔
62	if (data.event === 'error') {	5✔
63	streamedError = new Error(formatOrphanStreamError(data.error))	1✔
64	console.error('Server error:', formatOrphanStreamError(data.error))	1✔
65	continue	1✔
66	}	1✔
67		4✔
68	if (data.value && Array.isArray(data.value)) {	5✔
69	receivedAnyData = true	4✔
70	console.log(`Processing ${data.value.length} objects`)	4✔
71		4✔
72	for (const item of data.value) {	4✔
73	if (!isFirstItem) {	4!
NEW 74	yield ',\n'	×
75	} else {	4✔
76	isFirstItem = false	4✔
77	}	4✔
78		4✔
79	yield JSON.stringify({ ...item, orphanType: data.type }, null, 2)	4✔
80	}	4✔
81	continue	4✔
82	}	4✔
NEW 83		×
NEW 84	console.warn(	×
NEW 85	'Received data with invalid format:',	×
NEW 86	JSON.stringify(data).substring(0, 100) + '...'	×
NEW 87	)	×
NEW 88	}	×
89	} catch (err) {	5✔
90	if (err instanceof Error && err.message === 'DELETE_LIMIT_REACHED') {	2✔
91	deleteLimitReached = true	1✔
92	} else {	1✔
93	inputStreamFailed = true	1✔
94	inputStreamError = err	1✔
95	console.error('Stream error:', err)	1✔
96	}	1✔
97	}	2✔
98		5✔
99	yield '\n]'	5✔
100	})()	5✔
101		5✔
102	await pipeline(jsonArrayStream, localFile)	5✔
103		4✔
104	if (inputStreamFailed) {	5✔
105	throw inputStreamError	1✔
106	}	1✔
107		3✔
108	if (streamedError) {	5✔
109	throw streamedError	1✔
110	}	1✔
111		2✔
112	if (!receivedAnyData) {	5!
NEW 113	console.warn(`No data was received! File might be empty: ${filePath}`)	×
NEW 114	return	×
NEW 115	}	×
116		2✔
117	if (deleteLimitReached) {	5✔
118	console.log(`Finished writing data to ${filePath}. Delete limit reached, data saved.`)	1✔
119	return	1✔
120	}	1✔
121		1✔
122	console.log(`Finished writing data to ${filePath}. Data was received and saved.`)	1✔
123	}	1✔

supabase / storage / 23952459316

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