@@ -33,7 +33,7 @@ interface TableCleanupResult {
3333async function cleanupTable (
3434 tableDef : PgTable ,
3535 workspaceIdCol : PgColumn ,
36- createdAtCol : PgColumn ,
36+ timestampCol : PgColumn ,
3737 workspaceIds : string [ ] ,
3838 retentionDate : Date ,
3939 tableName : string
@@ -46,20 +46,30 @@ async function cleanupTable(
4646
4747 while ( hasMore && batchesProcessed < MAX_BATCHES_PER_TABLE ) {
4848 try {
49+ // SELECT with LIMIT first to avoid unbounded DELETE
50+ const batch = await db
51+ . select ( { id : sql < string > `id` } )
52+ . from ( tableDef )
53+ . where ( and ( inArray ( workspaceIdCol , workspaceIds ) , lt ( timestampCol , retentionDate ) ) )
54+ . limit ( BATCH_SIZE )
55+
56+ if ( batch . length === 0 ) {
57+ logger . info ( `[${ tableName } ] No expired rows found` )
58+ hasMore = false
59+ break
60+ }
61+
62+ const ids = batch . map ( ( r ) => r . id )
4963 const deleted = await db
5064 . delete ( tableDef )
51- . where ( and ( inArray ( workspaceIdCol , workspaceIds ) , lt ( createdAtCol , retentionDate ) ) )
65+ . where ( inArray ( sql `id` , ids ) )
5266 . returning ( { id : sql `id` } )
5367
5468 result . deleted += deleted . length
55- hasMore = deleted . length === BATCH_SIZE
69+ hasMore = batch . length === BATCH_SIZE
5670 batchesProcessed ++
5771
58- if ( deleted . length > 0 ) {
59- logger . info ( `[${ tableName } ] Batch ${ batchesProcessed } : deleted ${ deleted . length } rows` )
60- } else {
61- logger . info ( `[${ tableName } ] No expired rows found` )
62- }
72+ logger . info ( `[${ tableName } ] Batch ${ batchesProcessed } : deleted ${ deleted . length } rows` )
6373 } catch ( error ) {
6474 result . failed ++
6575 logger . error ( `[${ tableName } ] Batch delete failed:` , { error } )
0 commit comments