24690350305

Committed 20 Apr 2026 09:05PM UTC coverage: 69.744% (-0.2%) from 69.897%

Build # 24690350305

Build Type

Pull #1038

github

Committed by

web-flow

Commit Message

Merge 7d01e34c6 into f17973e9e

Pull Request Pull Request #1038: chore: drop axios from orphan script

Coverage Stats

3394 of 5399 branches covered (62.86%)

Branch coverage included in aggregate %.

55 of 83 new or added lines in 1 file covered. (66.27%)

2 existing lines in 1 file now uncovered.

7113 of 9666 relevant lines covered (73.59%)

375.85 hits per line

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

59.48

/src/scripts/orphan-client.ts

import { NdJsonTransform } from '@internal/streams/ndjson'
import path from 'path'
import { Readable } from 'stream'
import type { ReadableStream as NodeReadableStream } from 'stream/web'
import { OrphanStreamEvent, writeStreamToJsonArray } from './orphan-client-stream'

const DEFAULT_DELETE_LIMIT = 1000000

type OrphanAction = 'list' | 'delete'

interface OrphanClientConfig {
  adminUrl: string
  adminApiKey: string
  tenantId: string

  // bucket id to search, can handle multiple comma delimited buckets (aaa,bbb,ccc)
  bucketId: string

  // limits the number of delete operations to avoid overwhelming our queue
  deleteLimit: number

  // optional cutoff override for orphan list/delete requests
  before?: string
}

interface FetchOrphanStreamOptions {
  action: OrphanAction
  adminApiKey?: string
  adminUrl: string
  before?: string
  bucketId: string
  tenantId: string
}

interface WriteDeleteOrphanStreamOptions {
  requestStream: Readable
  cancel: () => void
  deleteLimit: number
  filePath: string
}

const FILE_PATH = (operation: string, tenantId: string, bucketId: string) =>
  `../../dist/${operation}-${tenantId}-${bucketId}-${Date.now()}-orphan-objects.json`

export function parseConfig(env: NodeJS.ProcessEnv): OrphanClientConfig | string {
  const { ADMIN_URL, ADMIN_API_KEY, TENANT_ID, BUCKET_ID, DELETE_LIMIT, ORPHAN_BEFORE } = env

  if (!ADMIN_URL) return 'Please provide an admin URL'
  if (!ADMIN_API_KEY) return 'Please provide an admin API key'
  if (!TENANT_ID) return 'Please provide a tenant ID'
  if (!BUCKET_ID) return 'Please provide a bucket ID'

  return {
    adminUrl: ADMIN_URL,
    adminApiKey: ADMIN_API_KEY,
    tenantId: TENANT_ID,
    bucketId: BUCKET_ID,
    deleteLimit: parseInt(DELETE_LIMIT || String(DEFAULT_DELETE_LIMIT), 10),
    before: ORPHAN_BEFORE,
  }
}

export function resolveAdminUrl(
  baseUrl: string,
  requestPath: string,
  query?: Record<string, string | undefined>
): URL {
  const url = new URL(`${baseUrl.replace(/\/+$/, '')}/${requestPath.replace(/^\/+/, '')}`)

  for (const [key, value] of Object.entries(query ?? {})) {
    if (value !== undefined) {
      url.searchParams.set(key, value)
    }
  }

  return url
}

async function assertStreamResponse(response: Response, context: string) {
  if (!response.ok) {
    const body = await response.text()
    const details = body ? `: ${body}` : ''

    throw new Error(`${context} failed with ${response.status} ${response.statusText}${details}`)
  }

  if (!response.body) {
    throw new Error(`${context} returned an empty response body`)
  }
}

export async function fetchOrphanStream(options: FetchOrphanStreamOptions) {
  const requestPath = `/tenants/${options.tenantId}/buckets/${options.bucketId}/orphan-objects`
  const url = resolveAdminUrl(
    options.adminUrl,
    requestPath,
    options.action === 'list' ? { before: options.before } : undefined
  )
  const headers = new Headers()

  if (options.adminApiKey) {
    headers.set('ApiKey', options.adminApiKey)
  }

  const requestBody =
    options.action === 'delete'
      ? JSON.stringify({
          deleteS3Keys: true,
          before: options.before,
        })
      : undefined

  if (requestBody) {
    headers.set('Content-Type', 'application/json')
  }

  const controller = new AbortController()
  const response = await fetch(url, {
    method: options.action === 'list' ? 'GET' : 'DELETE',
    headers,
    body: requestBody,
    signal: controller.signal,
  })

  await assertStreamResponse(response, `${options.action.toUpperCase()} ${url}`)

  const stream = Readable.fromWeb(response.body as unknown as NodeReadableStream<Uint8Array>)
  let completed = false

  stream.once('end', () => {
    completed = true
  })

  stream.once('close', () => {
    if (!completed && !controller.signal.aborted) {
      controller.abort()
    }
  })

  return {
    stream,
    cancel: () => {
      if (!stream.destroyed) {
        stream.destroy()
      }

      if (!controller.signal.aborted) {
        controller.abort()
      }
    },
  }
}

export async function writeDeleteOrphanStream(options: WriteDeleteOrphanStreamOptions) {
  const transformStream = new NdJsonTransform()
  let deleteLimitReached = false

  options.requestStream.on('error', (err: Error) => {
    if (deleteLimitReached) {
      return
    }

    transformStream.emit('error', err)
  })

  const jsonStream = options.requestStream.pipe(transformStream)

  let itemCount = 0
  const limitedStream = Readable.from(
    (async function* () {
      for await (const chunk of jsonStream as AsyncIterable<OrphanStreamEvent>) {
        yield chunk

        if (chunk.event === 'data' && chunk.value && Array.isArray(chunk.value)) {
          itemCount += chunk.value.length

          if (itemCount >= options.deleteLimit) {
            deleteLimitReached = true
            console.log(
              `Delete limit of ${options.deleteLimit} reached. Stopping after this batch. Ensure these operations complete before queuing additional jobs.`
            )
            options.cancel()
            return
          }
        }
      }
    })(),
    { objectMode: true }
  )

  await writeStreamToJsonArray(limitedStream, options.filePath)
}

export async function main(env: NodeJS.ProcessEnv = process.env, argv: string[] = process.argv) {
  const action = argv[2]

  if (action !== 'list' && action !== 'delete') {
    console.error('Please provide an action: list or delete')
    return
  }

  const config = parseConfig(env)
  if (typeof config === 'string') {
    console.error(config)
    return
  }

  const buckets = config.bucketId
    .split(',')
    .map((bucketId) => bucketId.trim())
    .filter(Boolean)

  for (const bucket of buckets) {
    console.log(' ')
    console.log(`${action} items in bucket ${bucket}...`)
    if (action === 'list') {
      await listOrphans(config, bucket)
    } else {
      await deleteS3Orphans(config, bucket)
    }
  }
}

/**
 * List Orphan objects in a bucket
 * @param tenantId
 * @param bucketId
 */
async function listOrphans(config: OrphanClientConfig, bucketId: string) {
  const { stream: requestStream } = await fetchOrphanStream({
    action: 'list',
    adminApiKey: config.adminApiKey,
    adminUrl: config.adminUrl,
    before: config.before,
    bucketId,
    tenantId: config.tenantId,
  })

  const transformStream = new NdJsonTransform()
  requestStream.on('error', (err: Error) => {
    transformStream.emit('error', err)
  })

  const jsonStream = requestStream.pipe(transformStream)

  await writeStreamToJsonArray(
    jsonStream,
    path.resolve(__dirname, FILE_PATH('list', config.tenantId, bucketId))
  )
}

/**
 * Deletes S3 orphan objects in a bucket
 * @param tenantId
 * @param bucketId
 */
async function deleteS3Orphans(config: OrphanClientConfig, bucketId: string) {
  const { stream: requestStream, cancel } = await fetchOrphanStream({
    action: 'delete',
    adminApiKey: config.adminApiKey,
    adminUrl: config.adminUrl,
    before: config.before,
    bucketId,
    tenantId: config.tenantId,
  })

  await writeDeleteOrphanStream({
    requestStream,
    cancel,
    deleteLimit: config.deleteLimit,
    filePath: path.resolve(__dirname, FILE_PATH('delete', config.tenantId, bucketId)),
  })
}

if (typeof require !== 'undefined' && typeof module !== 'undefined' && require.main === module) {
  void main()
    .then(() => {
      console.log('Done')
    })
    .catch((e) => {
      process.exitCode = 1
      console.error('Error:', e)
    })
}

1	import { NdJsonTransform } from '@internal/streams/ndjson'
2	import path from 'path'
3	import { Readable } from 'stream'
4	import type { ReadableStream as NodeReadableStream } from 'stream/web'
5	import { OrphanStreamEvent, writeStreamToJsonArray } from './orphan-client-stream'
6
7	const DEFAULT_DELETE_LIMIT = 1000000	1✔
8
9	type OrphanAction = 'list' \| 'delete'
10
11	interface OrphanClientConfig {
12	adminUrl: string
13	adminApiKey: string
14	tenantId: string
15
16	// bucket id to search, can handle multiple comma delimited buckets (aaa,bbb,ccc)
17	bucketId: string
18
19	// limits the number of delete operations to avoid overwhelming our queue
20	deleteLimit: number
21
22	// optional cutoff override for orphan list/delete requests
23	before?: string
24	}
25
26	interface FetchOrphanStreamOptions {
27	action: OrphanAction
28	adminApiKey?: string
29	adminUrl: string
30	before?: string
31	bucketId: string
32	tenantId: string
33	}
34
35	interface WriteDeleteOrphanStreamOptions {
36	requestStream: Readable
37	cancel: () => void
38	deleteLimit: number
39	filePath: string
40	}
41
42	const FILE_PATH = (operation: string, tenantId: string, bucketId: string) =>	1✔
NEW 43	`../../dist/${operation}-${tenantId}-${bucketId}-${Date.now()}-orphan-objects.json`	×
44
45	export function parseConfig(env: NodeJS.ProcessEnv): OrphanClientConfig \| string {
46	const { ADMIN_URL, ADMIN_API_KEY, TENANT_ID, BUCKET_ID, DELETE_LIMIT, ORPHAN_BEFORE } = env	1✔
47
48	if (!ADMIN_URL) return 'Please provide an admin URL'	1!
49	if (!ADMIN_API_KEY) return 'Please provide an admin API key'	1!
50	if (!TENANT_ID) return 'Please provide a tenant ID'	1!
51	if (!BUCKET_ID) return 'Please provide a bucket ID'	1!
52
53	return {	1✔
54	adminUrl: ADMIN_URL,
55	adminApiKey: ADMIN_API_KEY,
56	tenantId: TENANT_ID,
57	bucketId: BUCKET_ID,
58	deleteLimit: parseInt(DELETE_LIMIT \|\| String(DEFAULT_DELETE_LIMIT), 10),	2✔
59	before: ORPHAN_BEFORE,
60	}
61	}
62
63	export function resolveAdminUrl(
64	baseUrl: string,
65	requestPath: string,
66	query?: Record<string, string \| undefined>
67	): URL {
68	const url = new URL(`${baseUrl.replace(/\/+$/, '')}/${requestPath.replace(/^\/+/, '')}`)	5✔
69
70	for (const [key, value] of Object.entries(query ?? {})) {	5✔
71	if (value !== undefined) {	3✔
72	url.searchParams.set(key, value)	2✔
73	}
74	}
75
76	return url	5✔
77	}
78
79	async function assertStreamResponse(response: Response, context: string) {
80	if (!response.ok) {	3✔
81	const body = await response.text()	1✔
82	const details = body ? `: ${body}` : ''	1!
83
84	throw new Error(`${context} failed with ${response.status} ${response.statusText}${details}`)	1✔
85	}
86
87	if (!response.body) {	2!
NEW 88	throw new Error(`${context} returned an empty response body`)	×
89	}
90	}
91
92	export async function fetchOrphanStream(options: FetchOrphanStreamOptions) {
93	const requestPath = `/tenants/${options.tenantId}/buckets/${options.bucketId}/orphan-objects`	3✔
94	const url = resolveAdminUrl(	3✔
95	options.adminUrl,
96	requestPath,
97	options.action === 'list' ? { before: options.before } : undefined	3✔
98	)
99	const headers = new Headers()	3✔
100
101	if (options.adminApiKey) {	3✔
102	headers.set('ApiKey', options.adminApiKey)	2✔
103	}
104
105	const requestBody =
106	options.action === 'delete'	3✔
107	? JSON.stringify({
108	deleteS3Keys: true,
109	before: options.before,
110	})
111	: undefined
112
113	if (requestBody) {	3✔
114	headers.set('Content-Type', 'application/json')	1✔
115	}
116
117	const controller = new AbortController()	3✔
118	const response = await fetch(url, {	3✔
119	method: options.action === 'list' ? 'GET' : 'DELETE',	3✔
120	headers,
121	body: requestBody,
122	signal: controller.signal,
123	})
124
125	await assertStreamResponse(response, `${options.action.toUpperCase()} ${url}`)	3✔
126
127	const stream = Readable.fromWeb(response.body as unknown as NodeReadableStream<Uint8Array>)	2✔
128	let completed = false	2✔
129
130	stream.once('end', () => {	2✔
131	completed = true	2✔
132	})
133
134	stream.once('close', () => {	2✔
135	if (!completed && !controller.signal.aborted) {	2!
NEW 136	controller.abort()	×
137	}
138	})
139
140	return {	2✔
141	stream,
142	cancel: () => {
NEW 143	if (!stream.destroyed) {	×
NEW 144	stream.destroy()	×
145	}
146
NEW 147	if (!controller.signal.aborted) {	×
NEW 148	controller.abort()	×
149	}
150	},
151	}
152	}
153
154	export async function writeDeleteOrphanStream(options: WriteDeleteOrphanStreamOptions) {
155	const transformStream = new NdJsonTransform()	1✔
156	let deleteLimitReached = false	1✔
157
158	options.requestStream.on('error', (err: Error) => {	1✔
159	if (deleteLimitReached) {	1!
160	return	1✔
161	}
162
NEW 163	transformStream.emit('error', err)	×
164	})
165
166	const jsonStream = options.requestStream.pipe(transformStream)	1✔
167
168	let itemCount = 0	1✔
169	const limitedStream = Readable.from(	1✔
170	(async function* () {
171	for await (const chunk of jsonStream as AsyncIterable<OrphanStreamEvent>) {	1✔
172	yield chunk	1✔
173
174	if (chunk.event === 'data' && chunk.value && Array.isArray(chunk.value)) {	1!
175	itemCount += chunk.value.length	1✔
176
177	if (itemCount >= options.deleteLimit) {	1!
178	deleteLimitReached = true	1✔
179	console.log(	1✔
180	`Delete limit of ${options.deleteLimit} reached. Stopping after this batch. Ensure these operations complete before queuing additional jobs.`
181	)
182	options.cancel()	1✔
183	return	1✔
184	}
185	}
186	}
187	})(),
188	{ objectMode: true }
189	)
190
191	await writeStreamToJsonArray(limitedStream, options.filePath)	1✔
192	}
193
194	export async function main(env: NodeJS.ProcessEnv = process.env, argv: string[] = process.argv) {	×
NEW 195	const action = argv[2]	×
196
NEW 197	if (action !== 'list' && action !== 'delete') {	×
NEW 198	console.error('Please provide an action: list or delete')	×
199	return	×
200	}
201
NEW 202	const config = parseConfig(env)	×
NEW 203	if (typeof config === 'string') {	×
NEW 204	console.error(config)	×
205	return	×
206	}
207
NEW 208	const buckets = config.bucketId	×
209	.split(',')
NEW 210	.map((bucketId) => bucketId.trim())	×
211	.filter(Boolean)
212
213	for (const bucket of buckets) {	×
214	console.log(' ')	×
215	console.log(`${action} items in bucket ${bucket}...`)	×
216	if (action === 'list') {	×
NEW 217	await listOrphans(config, bucket)	×
218	} else {
NEW 219	await deleteS3Orphans(config, bucket)	×
220	}
221	}
222	}
223
224	/**
225	* List Orphan objects in a bucket
226	* @param tenantId
227	* @param bucketId
228	*/
229	async function listOrphans(config: OrphanClientConfig, bucketId: string) {
NEW 230	const { stream: requestStream } = await fetchOrphanStream({	×
231	action: 'list',
232	adminApiKey: config.adminApiKey,
233	adminUrl: config.adminUrl,
234	before: config.before,
235	bucketId,
236	tenantId: config.tenantId,
237	})
238
239	const transformStream = new NdJsonTransform()	×
NEW 240	requestStream.on('error', (err: Error) => {	×
241	transformStream.emit('error', err)	×
242	})
243
NEW 244	const jsonStream = requestStream.pipe(transformStream)	×
245
NEW 246	await writeStreamToJsonArray(	×
247	jsonStream,
248	path.resolve(__dirname, FILE_PATH('list', config.tenantId, bucketId))
249	)
250	}
251
252	/**
253	* Deletes S3 orphan objects in a bucket
254	* @param tenantId
255	* @param bucketId
256	*/
257	async function deleteS3Orphans(config: OrphanClientConfig, bucketId: string) {
NEW 258	const { stream: requestStream, cancel } = await fetchOrphanStream({	×
259	action: 'delete',
260	adminApiKey: config.adminApiKey,
261	adminUrl: config.adminUrl,
262	before: config.before,
263	bucketId,
264	tenantId: config.tenantId,
265	})
266
NEW 267	await writeDeleteOrphanStream({	×
268	requestStream,
269	cancel,
270	deleteLimit: config.deleteLimit,
271	filePath: path.resolve(__dirname, FILE_PATH('delete', config.tenantId, bucketId)),
272	})
273	}
274
275	if (typeof require !== 'undefined' && typeof module !== 'undefined' && require.main === module) {	1!
NEW 276	void main()	×
277	.then(() => {
NEW 278	console.log('Done')	×
279	})
280	.catch((e) => {
NEW 281	process.exitCode = 1	×
NEW 282	console.error('Error:', e)	×
283	})
284	}

supabase / storage / 24690350305

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