fix(files): release purged file history atomically with restore

Author: waleedlatif1

apps/docs/openapi-v2-files-audit.json

@@ -4549,7 +4549,7 @@
             "type": "integer",
             "minimum": 1,
             "maximum": 2147483647,
-            "description": "Version number, increasing by one per recorded version. Numbers are never reused, so a gap means retention removed that version.",
+            "description": "Version number, increasing by one per recorded version. Numbers are never reused, so a gap means an older version was removed by retention or deleted.",
             "examples": [3]
           },
           "isCurrent": {

apps/sim/background/cleanup-file-versions.ts

@@ -8,7 +8,7 @@ import type { CleanupJobPayload } from '@/lib/billing/cleanup-dispatcher'
 import type { PlanCategory } from '@/lib/billing/plan-helpers'
 import { DEFAULT_DELETE_CHUNK_SIZE } from '@/lib/cleanup/batch-delete'
 import { retentionCleanupQueue } from '@/lib/cleanup/queue'
-import { deleteWorkspaceStorageObjects } from '@/lib/cleanup/storage-delete'
+import { StorageService } from '@/lib/uploads'
 import {
   FILE_VERSION_RETENTION_KEEP_LATEST,
   MAX_SUPERSEDED_FILE_VERSIONS,
@@ -117,10 +117,17 @@ function selectExpiredVersions(fileIds: string[], cutoff: Date, maxSuperseded: n
  * delete leaves its row — and the next run retries it — instead of orphaning the object.
  */
 async function deleteVersions(rows: Array<{ id: string; key: string }>, label: string) {
-  const failedKeys = await deleteWorkspaceStorageObjects(
-    rows.map((row) => row.key),
-    label
-  )
+  const failedKeys = new Set<string>()
+  for (const batch of chunkArray(rows, DEFAULT_DELETE_CHUNK_SIZE)) {
+    const deletion = await StorageService.deleteFiles(
+      batch.map((row) => row.key),
+      'workspace'
+    )
+    for (const { key, error } of deletion.failed) {
+      failedKeys.add(key)
+      logger.error(`[${label}] Failed to delete file version object ${key}`, { error })
+    }
+  }
   const removable = rows.filter((row) => !failedKeys.has(row.key))
   let deleted = 0
   for (const batch of chunkArray(removable, DEFAULT_DELETE_CHUNK_SIZE)) {

apps/sim/background/cleanup-soft-deletes.test.ts

@@ -82,6 +82,13 @@ vi.mock('@/lib/uploads', () => ({
 
 vi.mock('@/lib/uploads/server/metadata', () => ({ deleteFileMetadata: mockDeleteFileMetadata }))
 
+const { mockReleaseExpiredWorkspaceFileVersions } = vi.hoisted(() => ({
+  mockReleaseExpiredWorkspaceFileVersions: vi.fn(),
+}))
+vi.mock('@/lib/uploads/contexts/workspace/workspace-file-versions', () => ({
+  releaseExpiredWorkspaceFileVersions: mockReleaseExpiredWorkspaceFileVersions,
+}))
+
 vi.mock('@/lib/workflows/utils', () => ({
   deduplicateWorkflowName: mockDeduplicateWorkflowName,
 }))
@@ -122,6 +129,39 @@ describe('cleanup soft deletes', () => {
     })
   })
 
+  it('releases the version history of expired workspace files before purging their objects', async () => {
+    mockSelectRowsByIdChunks
+      .mockResolvedValueOnce([])
+      .mockResolvedValueOnce([])
+      .mockResolvedValueOnce([
+        {
+          id: 'file-1',
+          key: 'workspace/ws-1/file-1',
+          workspaceId: 'ws-1',
+          context: 'workspace',
+          sizeBytes: 5,
+        },
+        {
+          id: 'chat-file',
+          key: 'mothership/chat-file',
+          workspaceId: 'ws-1',
+          context: 'mothership',
+          sizeBytes: 3,
+        },
+      ])
+
+    await runCleanupSoftDeletes(basePayload)
+
+    expect(mockReleaseExpiredWorkspaceFileVersions).toHaveBeenCalledWith(
+      expect.anything(),
+      ['file-1'],
+      expect.any(Date)
+    )
+    expect(mockReleaseExpiredWorkspaceFileVersions.mock.invocationCallOrder[0]).toBeLessThan(
+      mockDeleteFiles.mock.invocationCallOrder[0]
+    )
+  })
+
   it('keeps metadata rows whose object deletion failed', async () => {
     mockSelectRowsByIdChunks
       .mockResolvedValueOnce([])

apps/sim/background/cleanup-soft-deletes.ts

@@ -11,7 +11,6 @@ import {
   workflowMcpServer,
   workspaceFile,
   workspaceFiles,
-  workspaceFileVersion,
 } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { chunkArray } from '@sim/utils/helpers'
@@ -40,12 +39,12 @@ import {
   cleanupOwnerCondition,
   resolveCleanupOwnerScope,
 } from '@/lib/cleanup/resource-scope'
-import { deleteWorkspaceStorageObjects } from '@/lib/cleanup/storage-delete'
 import { deduplicateFolderName } from '@/lib/folders/naming'
 import { hardDeleteDocuments } from '@/lib/knowledge/documents/service'
 import type { StorageContext } from '@/lib/uploads'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
 import { allocateUniqueWorkspaceFileName } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
+import { releaseExpiredWorkspaceFileVersions } from '@/lib/uploads/contexts/workspace/workspace-file-versions'
 import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
 import { getWorkspaceFileSize } from '@/lib/uploads/shared/types'
 import { deduplicateWorkflowName } from '@/lib/workflows/utils'
@@ -224,64 +223,6 @@ async function cleanupWorkspaceFileStorage(
   return result
 }
 
-/** Files whose version rows are read in one query while their objects are collected. */
-const VERSION_OBJECT_FILE_CHUNK_SIZE = 100
-
-/**
- * Deletes the stored objects of every earlier version of the workspace files about to be purged,
- * before their version rows cascade away with the file. A file whose version objects could not all
- * be deleted is withheld from this purge, so its rows survive for the next run instead of leaving
- * those objects unreferenced. Each chunk re-reads which files are still deleted past the cutoff
- * right before deleting, so a file restored since selection keeps its history; the file-row delete
- * re-checks the same condition, so it skips that file too.
- */
-async function cleanupWorkspaceFileVersionStorage(
-  rows: WorkspaceFileScope['multiContextRows'],
-  retentionDate: Date,
-  label: string
-): Promise<{ rows: WorkspaceFileScope['multiContextRows']; failed: number }> {
-  const workspaceRows = rows.filter((row) => row.context === 'workspace')
-  if (workspaceRows.length === 0) return { rows, failed: 0 }
-
-  const currentKeyByFileId = new Map(workspaceRows.map((row) => [row.id, row.key]))
-  const withheldFileIds = new Set<string>()
-  for (const selectedFileIds of chunkArray(
-    [...currentKeyByFileId.keys()],
-    VERSION_OBJECT_FILE_CHUNK_SIZE
-  )) {
-    const stillExpired = await cleanupDb
-      .select({ id: workspaceFiles.id })
-      .from(workspaceFiles)
-      .where(
-        and(
-          inArray(workspaceFiles.id, selectedFileIds),
-          isNotNull(workspaceFiles.deletedAt),
-          lt(workspaceFiles.deletedAt, retentionDate)
-        )
-      )
-    const fileIds = stillExpired.map((file) => file.id)
-    if (fileIds.length === 0) continue
-    const versions = await cleanupDb
-      .select({ fileId: workspaceFileVersion.fileId, key: workspaceFileVersion.key })
-      .from(workspaceFileVersion)
-      .where(inArray(workspaceFileVersion.fileId, fileIds))
-    const fileIdByKey = new Map(
-      versions
-        .filter((version) => version.key !== currentKeyByFileId.get(version.fileId))
-        .map((version) => [version.key, version.fileId])
-    )
-    for (const key of await deleteWorkspaceStorageObjects([...fileIdByKey.keys()], label)) {
-      const fileId = fileIdByKey.get(key)
-      if (fileId) withheldFileIds.add(fileId)
-    }
-  }
-
-  return {
-    rows: rows.filter((row) => !withheldFileIds.has(row.id)),
-    failed: withheldFileIds.size,
-  }
-}
-
 async function deleteExpiredLegacyWorkspaceFileRows(
   rows: WorkspaceFileScope['legacyRows'],
   retentionDate: Date,
@@ -959,16 +900,12 @@ export async function runCleanupSoftDeletes(
     chatCleanup = await prepareChatCleanup([...doomedChatIds], label)
   }
 
-  const versionCleanup = await cleanupWorkspaceFileVersionStorage(
-    fileScope.multiContextRows,
-    retentionDate,
-    label
+  await releaseExpiredWorkspaceFileVersions(
+    cleanupDb,
+    fileScope.multiContextRows.filter((row) => row.context === 'workspace').map((row) => row.id),
+    retentionDate
   )
-  if (budgets && versionCleanup.failed) throw new Error('File version storage cleanup failed')
-  const fileCleanup = await cleanupWorkspaceFileStorage({
-    ...fileScope,
-    multiContextRows: versionCleanup.rows,
-  })
+  const fileCleanup = await cleanupWorkspaceFileStorage(fileScope)
   if (budgets && fileCleanup.filesFailed) throw new Error('File storage cleanup failed')
 
   let totalDeleted = 0

apps/sim/lib/api/contracts/v2/file-versions.ts

@@ -50,7 +50,7 @@ export const v2FileVersionSchema = z
     fileId: z.string().describe('File this version belongs to.'),
     version: versionNumberSchema
       .describe(
-        'Version number, increasing by one per recorded version. Numbers are never reused, so a gap means retention removed that version.'
+        'Version number, increasing by one per recorded version. Numbers are never reused, so a gap means an older version was removed by retention or deleted.'
       )
       .meta({ examples: [3] }),
     isCurrent: z.boolean().describe('Whether this version holds the current content of the file.'),

apps/sim/lib/cleanup/storage-delete.ts

@@ -6,9 +6,11 @@ import path from 'node:path'
 import { db, dbFor } from '@sim/db'
 import {
   organization,
+  outboxEvent,
   user,
   workspace,
   workspaceFileSecretProvenance,
+  workspaceFiles,
   workspaceFileVersion,
 } from '@sim/db/schema'
 import { generateId } from '@sim/utils/id'
@@ -33,9 +35,11 @@ import {
   updateWorkspaceFileContent,
   uploadWorkspaceFile,
 } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
+import { WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT } from '@/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox'
 import {
   getCurrentWorkspaceFileVersion,
   queryWorkspaceFileVersions,
+  releaseExpiredWorkspaceFileVersions,
 } from '@/lib/uploads/contexts/workspace/workspace-file-versions'
 import { revertWorkspaceFileVersion } from '@/lib/workspace-files/application/file-versions'
 import { runCleanupFileVersions } from '@/background/cleanup-file-versions'
@@ -398,6 +402,49 @@ describe('workspace file version history in PostgreSQL', () => {
     ).rejects.toMatchObject({ code: 'conflict' })
   })
 
+  it('releases an expired file history atomically and leaves a restored file untouched', async () => {
+    const expired = await seedFile('one')
+    const restored = await seedFile('one')
+    for (const fixture of [expired, restored]) {
+      for (const content of ['two', 'three']) {
+        await updateWorkspaceFileContent(
+          fixture.workspaceId,
+          fixture.fileId,
+          fixture.aliceId,
+          Buffer.from(content),
+          undefined,
+          { version: { source: 'api', authorUserId: fixture.aliceId } }
+        )
+      }
+    }
+    const deletedAt = sql`now() - interval '40 days'`
+    await db.update(workspaceFiles).set({ deletedAt }).where(eq(workspaceFiles.id, expired.fileId))
+    const releasedKeys = (await versionRows(expired.fileId))
+      .filter((row) => row.supersededAt !== null)
+      .map((row) => row.key)
+
+    await releaseExpiredWorkspaceFileVersions(
+      db,
+      [expired.fileId, restored.fileId],
+      new Date(Date.now() - 30 * 24 * 60 * 60 * 1000)
+    )
+
+    expect((await versionRows(expired.fileId)).map((row) => row.version)).toEqual([3])
+    expect((await versionRows(restored.fileId)).map((row) => row.version)).toEqual([1, 2, 3])
+    const events = await db
+      .select({ id: outboxEvent.id, payload: outboxEvent.payload })
+      .from(outboxEvent)
+      .where(eq(outboxEvent.eventType, WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT))
+    const enqueuedKeys = events.map((event) => (event.payload as { key: string }).key)
+    expect(enqueuedKeys).toEqual(expect.arrayContaining(releasedKeys))
+    await db.delete(outboxEvent).where(
+      inArray(
+        outboxEvent.id,
+        events.map((event) => event.id)
+      )
+    )
+  })
+
   it('prunes superseded versions past retention but always keeps the newest ten', async () => {
     const fixture = await seedFile('v1')
     for (let index = 2; index <= 13; index++) {

apps/sim/lib/uploads/contexts/workspace/__integration__/file-versions.integration.ts

@@ -3,11 +3,12 @@ import {
   type WorkspaceFileRow,
   type WorkspaceFileVersionRow,
   type WorkspaceFileVersionSource,
+  workspaceFiles,
   workspaceFileVersion,
 } from '@sim/db/schema'
 import { sha256Hex } from '@sim/security/hash'
 import { generateId } from '@sim/utils/id'
-import { and, desc, eq, inArray, isNotNull } from 'drizzle-orm'
+import { and, desc, eq, inArray, isNotNull, lt, notInArray } from 'drizzle-orm'
 import {
   type CursorKey,
   type KeysetKey,
@@ -21,6 +22,7 @@ import {
 import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import type { WorkspaceFileRecord } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
 import type { WorkspaceFileSecretProvenanceSnapshot } from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance'
+import { enqueueWorkspaceFileStorageCleanups } from '@/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox'
 import { getWorkspaceFileSize } from '@/lib/uploads/shared/types'
 
 /** A coalescing version stops absorbing writes this long after it was opened. */
@@ -290,6 +292,64 @@ export async function deleteWorkspaceFileVersionInTx(
   return deleted?.key ?? null
 }
 
+/** Files whose history is released in one transaction. */
+const RELEASE_FILE_CHUNK_SIZE = 100
+/** Most cleanup events one outbox insert may carry. */
+const RELEASE_ENQUEUE_CHUNK_SIZE = 1000
+
+/**
+ * Releases the history of soft-deleted files about to be purged: per chunk, one transaction locks
+ * the files still deleted before `deletedBefore`, deletes their superseded version rows, and
+ * enqueues those objects on the durable storage-cleanup outbox. The row lock orders this against a
+ * restore, which updates the same row, so a file is either restored first and keeps its whole
+ * history or purged first and restored without it — version rows never outlive their bytes, and no
+ * object is left unreferenced. The current version is left to the file's own purge.
+ */
+export async function releaseExpiredWorkspaceFileVersions(
+  client: typeof db,
+  fileIds: readonly string[],
+  deletedBefore: Date
+): Promise<void> {
+  for (let start = 0; start < fileIds.length; start += RELEASE_FILE_CHUNK_SIZE) {
+    const chunk = fileIds.slice(start, start + RELEASE_FILE_CHUNK_SIZE)
+    await client.transaction(async (tx) => {
+      const expired = await tx
+        .select({ id: workspaceFiles.id, key: workspaceFiles.key })
+        .from(workspaceFiles)
+        .where(
+          and(
+            inArray(workspaceFiles.id, chunk),
+            isNotNull(workspaceFiles.deletedAt),
+            lt(workspaceFiles.deletedAt, deletedBefore)
+          )
+        )
+        .for('update')
+      if (expired.length === 0) return
+      const released = await tx
+        .delete(workspaceFileVersion)
+        .where(
+          and(
+            inArray(
+              workspaceFileVersion.fileId,
+              expired.map((file) => file.id)
+            ),
+            notInArray(
+              workspaceFileVersion.key,
+              expired.map((file) => file.key)
+            )
+          )
+        )
+        .returning({ key: workspaceFileVersion.key })
+      for (let offset = 0; offset < released.length; offset += RELEASE_ENQUEUE_CHUNK_SIZE) {
+        await enqueueWorkspaceFileStorageCleanups(
+          tx,
+          released.slice(offset, offset + RELEASE_ENQUEUE_CHUNK_SIZE).map((row) => row.key)
+        )
+      }
+    })
+  }
+}
+
 /** Deletes superseded versions beyond {@link MAX_SUPERSEDED_FILE_VERSIONS}, returning their keys. */
 async function pruneExcessWorkspaceFileVersionsInTx(
   tx: DbTransaction,