fix(files): read file versions against one snapshot of the file

Author: waleedlatif1

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

@@ -868,7 +868,7 @@
       "get": {
         "operationId": "listFileVersions",
         "summary": "List File Versions",
-        "description": "List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into the current version while it is under ten minutes old and written in the last five. Renames and moves are not versions. An empty file is version 1 until its first content replaces it. Retention keeps the newest ten, so numbers can have gaps.\n\nOAuth scope: `api:read`.",
+        "description": "List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into a version under ten minutes old and written in the last five. Renames and moves are not versions. Retention removes older versions by age and plan but keeps the newest ten, so numbers can have gaps.\n\nOAuth scope: `api:read`.",
         "x-sim-operation": "files.versions.list",
         "x-oauth-scope": "api:read",
         "tags": ["Files"],

apps/sim/lib/api/contracts/v2/openapi/files-audit.ts

@@ -447,7 +447,7 @@ const declaredRoutes = [
       operationId: 'listFileVersions',
       summary: 'List File Versions',
       description:
-        'List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into the current version while it is under ten minutes old and written in the last five. Renames and moves are not versions. An empty file is version 1 until its first content replaces it. Retention keeps the newest ten, so numbers can have gaps.',
+        'List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into a version under ten minutes old and written in the last five. Renames and moves are not versions. Retention removes older versions by age and plan but keeps the newest ten, so numbers can have gaps.',
       errors: RESOURCE_ERRORS,
       success: { description: 'A page of file versions.' },
     }),

apps/sim/lib/api/mcp/generated/v2-operations.ts

@@ -1425,7 +1425,7 @@ export const V2_MCP_OPERATIONS = {
     contract: v2ListFileVersionsContract,
     summary: 'List File Versions',
     description:
-      'List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into the current version while it is under ten minutes old and written in the last five. Renames and moves are not versions. An empty file is version 1 until its first content replaces it. Retention keeps the newest ten, so numbers can have gaps.\n\nOAuth scope: `api:read`.',
+      'List the versions of a file, newest first by default. Each write that changes the bytes records one; identical rewrites do not. Collaborative edits, and repeated workflow writes by one author, fold into a version under ten minutes old and written in the last five. Renames and moves are not versions. Retention removes older versions by age and plan but keeps the newest ten, so numbers can have gaps.\n\nOAuth scope: `api:read`.',
     handler: () => import('@/app/api/v2/files/[fileId]/versions/route').then((route) => route.GET),
   },
   listKnowledgeBases: {

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

@@ -1,6 +1,6 @@
 /** Real PostgreSQL transactions and local object storage for workspace file version history. */
 import { mkdtempSync } from 'node:fs'
-import { access, rm } from 'node:fs/promises'
+import { access, mkdir, rm, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import path from 'node:path'
 import { db, dbFor } from '@sim/db'
@@ -39,6 +39,7 @@ import {
 import { WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT } from '@/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox'
 import {
   getCurrentWorkspaceFileVersion,
+  getWorkspaceFileVersion,
   queryWorkspaceFileVersions,
   releaseWorkspaceFileVersionsForPurgeInTx,
 } from '@/lib/uploads/contexts/workspace/workspace-file-versions'
@@ -314,9 +315,17 @@ describe('workspace file version history in PostgreSQL', () => {
     )
     /** A content write that replaced the bytes without recording a version, as a build predating history would. */
     const unrecordedKey = `${fixture.firstKey}-unrecorded`
+    const unrecordedContent = 'third, never recorded'
+    const unrecordedPath = path.join(fixtureStorage.root, unrecordedKey)
+    await mkdir(path.dirname(unrecordedPath), { recursive: true })
+    await writeFile(unrecordedPath, unrecordedContent)
     await db
       .update(workspaceFiles)
-      .set({ key: unrecordedKey, contentUpdatedAt: new Date() })
+      .set({
+        key: unrecordedKey,
+        sizeBytes: Buffer.byteLength(unrecordedContent),
+        contentUpdatedAt: new Date(),
+      })
       .where(eq(workspaceFiles.id, fixture.fileId))
     const file = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
     if (!file) throw new Error('file missing')
@@ -330,6 +339,7 @@ describe('workspace file version history in PostgreSQL', () => {
       isCurrent: true,
     })
     const listed = await queryWorkspaceFileVersions(file, { sortOrder: 'desc', limit: 10 })
+    expect(listed.versions[0].size).toBe(Buffer.byteLength(unrecordedContent))
     expect(listed.versions.map((row) => [row.version, row.isCurrent, row.source])).toEqual([
       [3, true, 'unknown'],
       [2, false, 'api'],
@@ -366,7 +376,34 @@ describe('workspace file version history in PostgreSQL', () => {
       [3, 'unknown'],
       [4, 'api'],
     ])
-    expect((await versionRows(fixture.fileId))[2].key).toBe(unrecordedKey)
+    const materialized = (await versionRows(fixture.fileId))[2]
+    expect(materialized.key).toBe(unrecordedKey)
+    expect(materialized.sizeBytes).toBe(Buffer.byteLength(unrecordedContent))
+    expect(await readVersionBytes(fixture.workspaceId, fixture.fileId, materialized.key)).toBe(
+      unrecordedContent
+    )
+  })
+
+  it('reads versions against the file as committed, not a record loaded before a write', async () => {
+    const fixture = await seedFile('original')
+    const stale = await getWorkspaceFile(fixture.workspaceId, fixture.fileId)
+    if (!stale) throw new Error('file missing')
+    await updateWorkspaceFileContent(
+      fixture.workspaceId,
+      fixture.fileId,
+      fixture.aliceId,
+      Buffer.from('second'),
+      undefined,
+      { version: { source: 'api', authorUserId: fixture.aliceId } }
+    )
+
+    const listed = await queryWorkspaceFileVersions(stale, { sortOrder: 'desc', limit: 10 })
+    expect(listed.versions.map((row) => [row.version, row.isCurrent])).toEqual([
+      [2, true],
+      [1, false],
+    ])
+    expect((await getCurrentWorkspaceFileVersion(stale)).version).toBe(2)
+    await expect(getWorkspaceFileVersion(stale, 3)).resolves.toBeNull()
   })
 
   it('never folds deliberate writes, and repoints the head for identical bytes', async () => {

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

@@ -485,6 +485,59 @@ function implicitCurrentVersion(
   }
 }
 
+/**
+ * Runs `read` in one read-only snapshot that also re-reads the file's content columns, so a content
+ * write committing mid-read can never pair one write's file record with another write's version
+ * rows. A file row deleted since the caller loaded it keeps the caller's record.
+ */
+function withVersionSnapshot<T>(
+  file: WorkspaceFileVersionSubject,
+  read: (tx: DbTransaction, file: WorkspaceFileVersionSubject) => Promise<T>
+): Promise<T> {
+  return db.transaction(
+    async (tx) => {
+      const [row] = await tx
+        .select({
+          key: workspaceFiles.key,
+          sizeBytes: workspaceFiles.sizeBytes,
+          contentType: workspaceFiles.contentType,
+          userId: workspaceFiles.userId,
+          uploadedAt: workspaceFiles.uploadedAt,
+          updatedAt: workspaceFiles.updatedAt,
+          contentUpdatedAt: workspaceFiles.contentUpdatedAt,
+        })
+        .from(workspaceFiles)
+        .where(eq(workspaceFiles.id, file.id))
+        .limit(1)
+      const snapshot: WorkspaceFileVersionSubject = row
+        ? {
+            id: file.id,
+            key: row.key,
+            size: getWorkspaceFileSize(row),
+            type: row.contentType,
+            uploadedBy: row.userId,
+            uploadedAt: row.uploadedAt,
+            updatedAt: row.updatedAt,
+            contentUpdatedAt: row.contentUpdatedAt,
+          }
+        : file
+      return read(tx, snapshot)
+    },
+    { isolationLevel: 'repeatable read', accessMode: 'read only' }
+  )
+}
+
+/** The version holding the file's current bytes, recorded or implicit, within a snapshot. */
+async function currentVersionInSnapshot(
+  tx: DbTransaction,
+  file: WorkspaceFileVersionSubject
+): Promise<WorkspaceFileVersionRecord> {
+  const head = await loadWorkspaceFileVersionHead(file.id, tx)
+  return head && isVersionHeadCurrent(head, file)
+    ? toVersionRecord(head, file)
+    : implicitCurrentVersion(file, head)
+}
+
 const VERSION_KEYSET: readonly KeysetKey<{ version: number }>[] = [
   numberKey(workspaceFileVersion.version, (row) => row.version),
 ]
@@ -495,35 +548,37 @@ export async function queryWorkspaceFileVersions(
   options: { sortOrder: ListSortOrder; limit: number; after?: CursorKey[] }
 ): Promise<{ versions: WorkspaceFileVersionRecord[]; nextKeys: CursorKey[] | null }> {
   const resume = resumeKeyset(VERSION_KEYSET, options.after, options.sortOrder)
-  const [rows, head] = await Promise.all([
-    db
+  return withVersionSnapshot(file, async (tx, current) => {
+    const rows = await tx
       .select(versionSummaryColumns)
       .from(workspaceFileVersion)
-      .where(and(eq(workspaceFileVersion.fileId, file.id), resume))
+      .where(and(eq(workspaceFileVersion.fileId, current.id), resume))
       .orderBy(...listOrderBy(keysetColumns(VERSION_KEYSET), options.sortOrder))
-      .limit(options.limit + 1),
-    loadWorkspaceFileVersionHead(file.id),
-  ])
-  const records = rows.map((row) => toVersionRecord(row, file))
-  /**
-   * The implicit current version numbers above every row, so it leads a descending list and ends an
-   * ascending one; a cursor already past it leaves it out. Over-fetching by one row still decides
-   * whether another page follows, since the cut keeps the first `limit` records either way.
-   */
-  const implicit = isVersionHeadCurrent(head, file) ? null : implicitCurrentVersion(file, head)
-  const resumeAfter = options.after?.[0]
-  if (
-    implicit &&
-    (typeof resumeAfter !== 'number' ||
-      (options.sortOrder === 'desc'
-        ? implicit.version < resumeAfter
-        : implicit.version > resumeAfter))
-  ) {
-    if (options.sortOrder === 'desc') records.unshift(implicit)
-    else records.push(implicit)
-  }
-  const page = keysetPage(VERSION_KEYSET, records, options.limit)
-  return { versions: page.data, nextKeys: page.nextCursorKeys }
+      .limit(options.limit + 1)
+    const head = await loadWorkspaceFileVersionHead(current.id, tx)
+    const records = rows.map((row) => toVersionRecord(row, current))
+    /**
+     * The implicit current version numbers above every row, so it leads a descending list and ends
+     * an ascending one; a cursor already past it leaves it out. Over-fetching by one row still
+     * decides whether another page follows, since the cut keeps the first `limit` records either way.
+     */
+    const implicit = isVersionHeadCurrent(head, current)
+      ? null
+      : implicitCurrentVersion(current, head)
+    const resumeAfter = options.after?.[0]
+    if (
+      implicit &&
+      (typeof resumeAfter !== 'number' ||
+        (options.sortOrder === 'desc'
+          ? implicit.version < resumeAfter
+          : implicit.version > resumeAfter))
+    ) {
+      if (options.sortOrder === 'desc') records.unshift(implicit)
+      else records.push(implicit)
+    }
+    const page = keysetPage(VERSION_KEYSET, records, options.limit)
+    return { versions: page.data, nextKeys: page.nextCursorKeys }
+  })
 }
 
 /**
@@ -542,13 +597,10 @@ export async function findWorkspaceFileVersionKeys(keys: readonly string[]): Pro
 }
 
 /** The version holding the file's current bytes, recorded or implicit. */
-export async function getCurrentWorkspaceFileVersion(
+export function getCurrentWorkspaceFileVersion(
   file: WorkspaceFileVersionSubject
 ): Promise<WorkspaceFileVersionRecord> {
-  const head = await loadWorkspaceFileVersionHead(file.id)
-  return head && isVersionHeadCurrent(head, file)
-    ? toVersionRecord(head, file)
-    : implicitCurrentVersion(file, head)
+  return withVersionSnapshot(file, currentVersionInSnapshot)
 }
 
 /**
@@ -582,18 +634,22 @@ export function currentWorkspaceFileVersionNumberSql() {
 }
 
 /** One version of a file, or null when it never existed or retention removed it. */
-export async function getWorkspaceFileVersion(
+export function getWorkspaceFileVersion(
   file: WorkspaceFileVersionSubject,
   version: number
 ): Promise<WorkspaceFileVersionRecord | null> {
-  const [row] = await db
-    .select(versionSummaryColumns)
-    .from(workspaceFileVersion)
-    .where(and(eq(workspaceFileVersion.fileId, file.id), eq(workspaceFileVersion.version, version)))
-    .limit(1)
-  if (row) return toVersionRecord(row, file)
-  const current = await getCurrentWorkspaceFileVersion(file)
-  return current.version === version ? current : null
+  return withVersionSnapshot(file, async (tx, current) => {
+    const [row] = await tx
+      .select(versionSummaryColumns)
+      .from(workspaceFileVersion)
+      .where(
+        and(eq(workspaceFileVersion.fileId, current.id), eq(workspaceFileVersion.version, version))
+      )
+      .limit(1)
+    if (row) return toVersionRecord(row, current)
+    const latest = await currentVersionInSnapshot(tx, current)
+    return latest.version === version ? latest : null
+  })
 }
 
 /**