1- import { opendir , rmdir } from 'node:fs/promises'
2- import { dirname , join } from 'node:path'
31import { db } from '@sim/db'
42import { project , uploadSession } from '@sim/db/schema'
53import { describeError } from '@sim/utils/errors'
@@ -18,22 +16,18 @@ import {
1816 PROJECT_FILE_PREFIX_CLEANUP_EVENT ,
1917 PROJECT_STORAGE_RECONCILE_EVENT ,
2018} from '@/lib/projects/files/outbox-events'
21- import {
22- getStorageConfig ,
23- type StorageContext ,
24- USE_BLOB_STORAGE ,
25- USE_GCS_STORAGE ,
26- USE_S3_STORAGE ,
27- } from '@/lib/uploads/config'
2819import { isObjectNotFoundError } from '@/lib/uploads/core/errors'
29- import { UPLOAD_DIR_SERVER } from '@/lib/uploads/core/setup.server'
30- import { deleteFile } from '@/lib/uploads/core/storage-service'
31- import { abortProviderUpload , UPLOAD_URL_TTL_MS } from '@/lib/uploads/upload-session/provider'
20+ import {
21+ deleteStorageKeysUnderPrefix ,
22+ listStorageKeyPage ,
23+ outboxDeadlineNear ,
24+ storageCleanupAvailableAt ,
25+ } from '@/lib/uploads/core/prefix-sweep'
26+ import { abortProviderUpload } from '@/lib/uploads/upload-session/provider'
3227import type { UploadSessionRecord } from '@/lib/uploads/upload-session/service'
3328import { PROJECT_FILE_UPLOAD_BINDING_KEY } from '@/lib/uploads/upload-session/types'
3429
3530const PAGE_SIZE = 500
36- const CLOCK_SKEW_MS = 60_000
3731
3832type UploadCleanup = Pick <
3933 UploadSessionRecord ,
@@ -45,16 +39,6 @@ interface CleanupPayload {
4539 uploadCursor ?: number
4640}
4741
48- /** Expired signatures stop new transfers, but a transfer already in progress may finish later. */
49- export function projectUploadCleanupAvailableAt ( now = new Date ( ) ) : Date {
50- return storageCleanupAvailableAt ( now )
51- }
52-
53- /** Earliest moment no upload URL signed before `now` can still write an object. */
54- export function storageCleanupAvailableAt ( now = new Date ( ) ) : Date {
55- return new Date ( now . getTime ( ) + UPLOAD_URL_TTL_MS + CLOCK_SKEW_MS )
56- }
57-
5842function prefixFor ( projectId : string ) : string {
5943 if ( ! / ^ [ a - z A - Z 0 - 9 _ - ] + $ / . test ( projectId ) ) throw new Error ( 'Invalid Project cleanup identity' )
6044 return `project/${ projectId } /`
@@ -108,7 +92,7 @@ async function enqueueSweeps(
10892) {
10993 await enqueueOutboxEvent ( executor , PROJECT_FILE_PREFIX_CLEANUP_EVENT , payload )
11094 await enqueueOutboxEvent ( executor , PROJECT_FILE_PREFIX_CLEANUP_EVENT , payload , {
111- availableAt : projectUploadCleanupAvailableAt ( retiredAt ) ,
95+ availableAt : storageCleanupAvailableAt ( retiredAt ) ,
11296 } )
11397}
11498
@@ -217,171 +201,20 @@ export async function queueRetiredProjectUploadCleanup(
217201 } )
218202}
219203
220- async function removeEmptyDirectory ( path : string ) : Promise < void > {
221- try {
222- await rmdir ( path )
223- } catch ( error ) {
224- if ( ! [ 'ENOENT' , 'ENOTEMPTY' , 'EEXIST' ] . includes ( describeError ( error ) . code ?? '' ) ) throw error
225- }
226- }
227-
228- async function localKeys ( prefix : string , signal : AbortSignal ) : Promise < string [ ] > {
229- const keys : string [ ] = [ ]
230- async function visit ( key : string ) : Promise < void > {
231- signal . throwIfAborted ( )
232- let directory
233- try {
234- directory = await opendir ( join ( UPLOAD_DIR_SERVER , key ) )
235- } catch ( error ) {
236- if ( describeError ( error ) . code === 'ENOENT' ) return
237- throw error
238- }
239- for await ( const entry of directory ) {
240- signal . throwIfAborted ( )
241- const child = `${ key } ${ entry . name } `
242- if ( entry . isDirectory ( ) ) await visit ( `${ child } /` )
243- else keys . push ( child )
244- if ( keys . length >= PAGE_SIZE ) return
245- }
246- await removeEmptyDirectory ( join ( UPLOAD_DIR_SERVER , key ) )
247- }
248- await visit ( prefix )
249- return keys
250- }
251-
252- /** One provider page of object keys; `done` once the listing has nothing after this page. */
253- interface StorageKeyPage {
254- keys : string [ ]
255- done : boolean
256- /** Provider cursor for the following page, valid only within the current run. */
257- nextPageToken ?: string
258- }
259-
260- /**
261- * Lists one page of object keys under `prefix` in the bucket serving `context`. A local listing
262- * has no cursor: it returns the first keys still on disk, so callers advance by deleting them.
263- */
264- export async function listStorageKeyPage (
265- prefix : string ,
266- context : StorageContext ,
267- signal : AbortSignal ,
268- pageToken ?: string
269- ) : Promise < StorageKeyPage > {
270- signal . throwIfAborted ( )
271- if ( USE_S3_STORAGE ) {
272- const { ListObjectsV2Command } = await import ( '@aws-sdk/client-s3' )
273- const { getS3Client } = await import ( '@/lib/uploads/providers/s3/client' )
274- const config = getStorageConfig ( context )
275- const page = await getS3Client ( ) . send (
276- new ListObjectsV2Command ( {
277- Bucket : config . bucket ,
278- Prefix : prefix ,
279- MaxKeys : PAGE_SIZE ,
280- ...( pageToken ? { ContinuationToken : pageToken } : { } ) ,
281- } ) ,
282- { abortSignal : signal }
283- )
284- const nextPageToken = page . IsTruncated ? page . NextContinuationToken : undefined
285- return {
286- keys : ( page . Contents ?? [ ] ) . flatMap ( ( object ) => ( object . Key ? [ object . Key ] : [ ] ) ) ,
287- done : ! nextPageToken ,
288- ...( nextPageToken ? { nextPageToken } : { } ) ,
289- }
290- }
291- if ( USE_BLOB_STORAGE ) {
292- const { getBlobServiceClient } = await import ( '@/lib/uploads/providers/blob/client' )
293- const config = getStorageConfig ( context )
294- if ( ! config . containerName ) throw new Error ( `Blob container missing for ${ context } storage` )
295- const pages = ( await getBlobServiceClient ( ) )
296- . getContainerClient ( config . containerName )
297- . listBlobsFlat ( { prefix, abortSignal : signal } )
298- . byPage ( { maxPageSize : PAGE_SIZE , ...( pageToken ? { continuationToken : pageToken } : { } ) } )
299- const page = await pages . next ( )
300- if ( page . done ) return { keys : [ ] , done : true }
301- const nextPageToken = page . value . continuationToken || undefined
302- return {
303- keys : page . value . segment . blobItems . map ( ( object ) => object . name ) ,
304- done : ! nextPageToken ,
305- ...( nextPageToken ? { nextPageToken } : { } ) ,
306- }
307- }
308- if ( USE_GCS_STORAGE ) {
309- const { getGcsClient } = await import ( '@/lib/uploads/providers/google-cloud-storage/client' )
310- const config = getStorageConfig ( context )
311- if ( ! config . bucket ) throw new Error ( `Bucket missing for ${ context } storage` )
312- const [ files , nextQuery ] = await ( await getGcsClient ( ) ) . bucket ( config . bucket ) . getFiles ( {
313- prefix,
314- maxResults : PAGE_SIZE ,
315- autoPaginate : false ,
316- ...( pageToken ? { pageToken } : { } ) ,
317- } )
318- signal . throwIfAborted ( )
319- const nextPageToken =
320- isRecordLike ( nextQuery ) && typeof nextQuery . pageToken === 'string' && nextQuery . pageToken
321- ? nextQuery . pageToken
322- : undefined
323- return {
324- keys : files . map ( ( file ) => file . name ) ,
325- done : ! nextPageToken ,
326- ...( nextPageToken ? { nextPageToken } : { } ) ,
327- }
328- }
329- const keys = await localKeys ( prefix , signal )
330- return { keys, done : keys . length < PAGE_SIZE }
331- }
332-
333- /**
334- * Deletes listed keys, each of which must sit under `prefix`, and prunes local directories the
335- * deletions empty, never above the prefix itself. Stops before another deletion once the handler
336- * deadline is near; `complete` is false when it did. `deleted` counts only objects actually
337- * removed: a key already gone (or, locally, one whose on-disk name no storage key resolves to) is
338- * no progress, so a caller re-listing the same page can tell that it is stuck.
339- */
340- export async function deleteStorageKeysUnderPrefix (
341- prefix : string ,
342- context : StorageContext ,
343- keys : readonly string [ ] ,
344- outbox : Pick < OutboxEventContext , 'signal' | 'deadlineAt' >
345- ) : Promise < { deleted : number ; complete : boolean } > {
346- if ( prefix . length < 2 || ! prefix . endsWith ( '/' ) ) throw new Error ( 'Unbounded storage prefix' )
347- let attempted = 0
348- let deleted = 0
349- for ( const key of keys ) {
350- if ( ! key . startsWith ( prefix ) )
351- throw new Error ( 'Storage provider returned a key outside the prefix' )
352- outbox . signal . throwIfAborted ( )
353- if ( attempted > 0 && outbox . deadlineAt && Date . now ( ) + 5000 >= outbox . deadlineAt )
354- return { deleted, complete : false }
355- attempted ++
356- let removed = true
357- try {
358- await deleteFile ( { key, context, signal : outbox . signal } )
359- } catch ( error ) {
360- if ( describeError ( error ) . code !== 'ENOENT' ) throw error
361- removed = false
362- }
363- if ( ! USE_S3_STORAGE && ! USE_BLOB_STORAGE && ! USE_GCS_STORAGE ) {
364- const root = join ( UPLOAD_DIR_SERVER , prefix . slice ( 0 , - 1 ) )
365- let directory = dirname ( join ( UPLOAD_DIR_SERVER , key ) )
366- while ( directory === root || directory . startsWith ( `${ root } /` ) ) {
367- await removeEmptyDirectory ( directory )
368- directory = dirname ( directory )
369- }
370- }
371- if ( removed ) deleted ++
372- }
373- return { deleted, complete : true }
374- }
375-
376204/** `stalled` when a page that must be re-listed removed nothing, so continuing cannot progress. */
377205async function sweepPrefix (
378206 projectId : string ,
379207 context : OutboxEventContext
380208) : Promise < 'done' | 'continue' | 'stalled' > {
381209 const prefix = prefixFor ( projectId )
382- const { keys } = await listStorageKeyPage ( prefix , 'project' , context . signal )
383- const { deleted, complete } = await deleteStorageKeysUnderPrefix ( prefix , 'project' , keys , context )
384- if ( complete && keys . length < PAGE_SIZE ) return 'done'
210+ const page = await listStorageKeyPage ( prefix , 'project' , context . signal )
211+ const { deleted, complete } = await deleteStorageKeysUnderPrefix (
212+ prefix ,
213+ 'project' ,
214+ page . keys ,
215+ context
216+ )
217+ if ( complete && page . done ) return 'done'
385218 return deleted > 0 ? 'continue' : 'stalled'
386219}
387220
@@ -417,7 +250,7 @@ const cleanupPrefix: OutboxHandler<unknown> = async (raw, context) => {
417250 }
418251 if ( ( index + 1 ) % 20 === 0 || index === uploads . length - 1 )
419252 await context . checkpointPayload ( { uploadCursor : index + 1 } )
420- if ( context . deadlineAt && Date . now ( ) + 5000 >= context . deadlineAt ) {
253+ if ( outboxDeadlineNear ( context ) ) {
421254 await context . checkpointPayload ( { uploadCursor : index + 1 } )
422255 return continueOutboxHandler ( 'Continuing retired Project upload cleanup' )
423256 }
0 commit comments