fix(files): address review admission and collaboration gaps

Author: mzxchandra

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

@@ -2,7 +2,7 @@
   "openapi": "3.1.0",
   "info": {
     "title": "Sim API v2 — Files & Audit Logs",
-    "description": "Version 2 of the Sim REST API for workspace and Project files, resumable uploads, public shares, and organization audit logs.",
+    "description": "Version 2 of the Sim REST API for workspace files, resumable uploads, public shares, and organization audit logs.",
     "version": "2.0.0",
     "contact": {
       "name": "Sim Support",
@@ -23,7 +23,7 @@
   "tags": [
     {
       "name": "Files",
-      "description": "Create, upload, download, organize, share, version, and delete workspace and Project files."
+      "description": "Create, upload, download, organize, share, version, and delete workspace files."
     },
     {
       "name": "Audit Logs",

apps/realtime/src/handlers/file-doc-editor.integration.ts

@@ -202,7 +202,7 @@ describe('Project document editor attribution over real Redis', () => {
   )
 
   check(
-    'a late prefix snapshot cannot replace a newer editor and unknown newer edits erase attribution',
+    'a late prefix snapshot and durable merge preserve the accepted editor without carrying it into a new generation',
     async () => {
       const f = await fixture()
       await store.publishClientUpdateAndWait(
@@ -226,10 +226,35 @@ describe('Project document editor attribution over real Redis', () => {
         ei: entry.id,
       })
       expect((await store.getStreamSnapshot(f.name, f.generation))?.editor).toEqual(bob)
-      await redis.xAdd(`filedoc:stream:${f.name}`, '*', {
-        u: Buffer.from(update(f.doc, '?')).toString('base64'),
-      })
-      expect((await store.getStreamSnapshot(f.name, f.generation))?.editor).toBeNull()
+      await store.publishAndWait(f.name, update(f.doc, '-durable'), f.generation)
+      const merged = await store.getStreamSnapshot(f.name, f.generation)
+      expect(merged?.editor).toEqual(bob)
+      const restored = new Y.Doc()
+      const replacement = new Y.Doc()
+      try {
+        if (!merged) throw new Error('Missing merged snapshot')
+        Y.applyUpdate(restored, merged.docState)
+        expect(restored.getText('body').toString()).toBe('AB-durable')
+        await store.invalidateDocument(f.name, 2)
+        const replacementId = generateId()
+        replacement.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
+        replacement.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.docIdKey, replacementId)
+        await store.seedIfEmpty(f.name, Y.encodeStateAsUpdate(replacement), 2)
+        await store.publishAndWait(f.name, update(replacement, 'only-durable'), replacementId)
+        const next = await store.getStreamSnapshot(f.name, replacementId)
+        expect(next?.editor).toBeNull()
+        const nextDoc = new Y.Doc()
+        try {
+          if (!next) throw new Error('Missing replacement snapshot')
+          Y.applyUpdate(nextDoc, next.docState)
+          expect(nextDoc.getText('body').toString()).toBe('only-durable')
+        } finally {
+          nextDoc.destroy()
+        }
+      } finally {
+        restored.destroy()
+        replacement.destroy()
+      }
     }
   )
 })

apps/realtime/src/handlers/file-doc-store.test.ts

@@ -224,7 +224,7 @@ function makeClient(): TestRedisClient {
         if (members.includes(member)) return 0
         const id = nextId()
         const arr = b().streams.get(key) ?? []
-        arr.push({ id, message: { [field]: value } })
+        arr.push({ id, message: { [field]: value, eu: opts.arguments[6], ec: opts.arguments[7] } })
         b().streams.set(key, arr)
         members.push(member)
         const capacity = Number(capacityText)
@@ -263,6 +263,9 @@ function makeClient(): TestRedisClient {
             [marker]: '1',
             [generationField]: expectedGeneration,
             ...(compactionField ? { [compactionField]: '1' } : {}),
+            eu: opts.arguments[7],
+            ec: opts.arguments[8],
+            ei: opts.arguments[9],
           },
         })
         b().streams.set(key, arr)
@@ -393,6 +396,57 @@ async function newStore(): Promise<FileDocStore> {
 }
 
 describe('FileDocStore', () => {
+  it('retains only an accepted editor across durable merges, compaction, and generation replacement', async () => {
+    const store = await newStore()
+    const source = docWithText('base')
+    source.getMap('config').set('docId', 'generation-1')
+    source.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
+    const replica = new Y.Doc()
+    try {
+      await store.seedIfEmpty(NAME, Y.encodeStateAsUpdate(source), 10)
+      let vector = Y.encodeStateVector(source)
+      source.getText('body').insert(4, '-durable')
+      await store.publishAndWait(NAME, Y.encodeStateAsUpdate(source, vector), 'generation-1')
+      expect((await store.getStreamSnapshot(NAME, 'generation-1'))?.editor).toBeNull()
+      vector = Y.encodeStateVector(source)
+      source.getText('body').insert(source.getText('body').length, '-writer')
+      const actor = { userId: 'actual-editor', connectionId: 'authenticated-socket' }
+      await store.publishClientUpdateAndWait(
+        NAME,
+        'edit-1',
+        Y.encodeStateAsUpdate(source, vector),
+        'generation-1',
+        actor
+      )
+      vector = Y.encodeStateVector(source)
+      source.getText('body').insert(source.getText('body').length, '-external')
+      await store.publishAndWait(NAME, Y.encodeStateAsUpdate(source, vector), 'generation-1')
+      const snapshot = await store.getStreamSnapshot(NAME, 'generation-1')
+      expect(snapshot?.editor).toEqual(actor)
+      Y.applyUpdate(replica, snapshot!.docState)
+      expect(replica.getText('body').toString()).toBe('base-durable-writer-external')
+      await store.attachRoom(NAME, replica)
+      storeInternals(store).rooms.get(NAME)!.uncompactedDeltaBytes = 12 * 1024 * 1024
+      await storeInternals(store).maybeCompact(NAME)
+      expect((await store.getStreamSnapshot(NAME, 'generation-1'))?.editor).toEqual(actor)
+      store.detachRoom(NAME)
+      await store.invalidateDocument(NAME, 20)
+      const replacement = docWithText('replacement')
+      replacement.getMap('config').set('docId', 'generation-2')
+      replacement.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
+      try {
+        await store.seedIfEmpty(NAME, Y.encodeStateAsUpdate(replacement), 20)
+        expect((await store.getStreamSnapshot(NAME, 'generation-2'))?.editor).toBeNull()
+      } finally {
+        replacement.destroy()
+      }
+    } finally {
+      store.detachRoom(NAME)
+      source.destroy()
+      replica.destroy()
+    }
+  })
+
   beforeEach(() => {
     state.backing = {
       streams: new Map(),

apps/realtime/src/handlers/file-doc-store.ts

@@ -329,11 +329,9 @@ function advanceEditor(
       ? message.ei
       : id
   if (current && !isAfterStreamId(sourceId, current.id)) return current
-  const editor =
-    message.eu?.trim() && message.ec?.trim()
-      ? { userId: message.eu, connectionId: message.ec }
-      : null
-  return { id: sourceId, editor }
+  /** Durable external merges carry no editor; retain only a previously authenticated writer. */
+  if (!message.eu?.trim() || !message.ec?.trim()) return current
+  return { id: sourceId, editor: { userId: message.eu, connectionId: message.ec } }
 }
 
 /** One locally-open room the store tracks: its doc and the last stream id applied to it. */

apps/sim/app/api/projects/[id]/files/[fileId]/content/route.ts

@@ -10,20 +10,23 @@ import {
   internalSessionAuth,
 } from '@/lib/api/server/routes'
 import { presentProjectFileContent } from '@/lib/projects/files/api'
-import { readProjectFileContent, updateProjectFileContent } from '@/lib/projects/files/application'
+import {
+  downloadProjectFileContent,
+  updateProjectFileContent,
+} from '@/lib/projects/files/application'
 import { MAX_WORKSPACE_FILE_INLINE_BODY_BYTES } from '@/lib/workspace-files/orchestration'
 
 export const GET = defineInternalBinaryRoute({
   contract: readProjectFileContentContract,
   auth: internalSessionAuth,
   headSafe: false,
-  operation: readProjectFileContent.operation,
+  operation: downloadProjectFileContent.operation,
   rateLimit: internalRateLimits.none({
     reason: 'Authenticated internal file byte delivery follows the workspace download policy',
   }),
   errorPolicy: internalOrchestrationErrorPolicy,
   mapInput: ({ params }) => ({ projectId: params.id, fileId: params.fileId }),
-  useCase: readProjectFileContent,
+  useCase: downloadProjectFileContent,
   present: presentProjectFileContent,
 })
 

apps/sim/app/api/projects/[id]/files/[fileId]/versions/[version]/content/route.ts

@@ -6,12 +6,12 @@ import {
   internalSessionAuth,
 } from '@/lib/api/server/routes'
 import { presentProjectFileVersionContent } from '@/lib/projects/files/api'
-import { readProjectFileVersionContent } from '@/lib/projects/files/application'
+import { downloadProjectFileVersionContent } from '@/lib/projects/files/application'
 export const GET = defineInternalBinaryRoute({
   contract: readProjectFileVersionContentContract,
   auth: internalSessionAuth,
   headSafe: false,
-  operation: readProjectFileVersionContent.operation,
+  operation: downloadProjectFileVersionContent.operation,
   rateLimit: internalRateLimits.none({
     reason: 'Authenticated historical byte delivery follows the workspace download policy',
   }),
@@ -21,6 +21,6 @@ export const GET = defineInternalBinaryRoute({
     fileId: params.fileId,
     version: params.version,
   }),
-  useCase: readProjectFileVersionContent,
+  useCase: downloadProjectFileVersionContent,
   present: presentProjectFileVersionContent,
 })

apps/sim/app/api/v2/projects/[projectId]/files/[fileId]/content/route.ts

@@ -11,7 +11,10 @@ import {
 } from '@/lib/api/server/routes'
 import { presentProjectFileContent } from '@/lib/projects/files/api'
 import { toV2ProjectFile } from '@/lib/projects/files/api/presenters'
-import { readProjectFileContent, updateProjectFileContent } from '@/lib/projects/files/application'
+import {
+  downloadProjectFileContent,
+  updateProjectFileContent,
+} from '@/lib/projects/files/application'
 import { projectFileOperations } from '@/lib/projects/files/application/operations'
 import { MAX_WORKSPACE_FILE_INLINE_BODY_BYTES } from '@/lib/workspace-files/orchestration'
 
@@ -22,7 +25,7 @@ export const GET = defineV2BinaryRoute({
   rateLimit: v2RateLimits.publicApi,
   errorPolicy: v2OrchestrationErrorPolicy,
   mapInput: ({ params }) => ({ projectId: params.projectId, fileId: params.fileId }),
-  useCase: readProjectFileContent,
+  useCase: downloadProjectFileContent,
   present: presentProjectFileContent,
   headSafe: false,
 })
@@ -34,6 +37,18 @@ export const PUT = defineV2JsonRoute({
   rateLimit: v2RateLimits.publicApi,
   errorPolicy: v2OrchestrationErrorPolicy,
   parseOptions: { maxBodyBytes: MAX_WORKSPACE_FILE_INLINE_BODY_BYTES },
+  beforeParse: async ({ principal, params }) => {
+    if (typeof params.projectId === 'string' && typeof params.fileId === 'string')
+      await updateProjectFileContent.authorize({
+        principal,
+        input: {
+          projectId: params.projectId,
+          fileId: params.fileId,
+          content: '',
+          encoding: 'utf-8',
+        },
+      })
+  },
   mapInput: ({ params, body }) => ({
     projectId: params.projectId,
     fileId: params.fileId,

apps/sim/app/api/v2/projects/[projectId]/files/[fileId]/versions/[version]/content/route.ts

@@ -6,7 +6,7 @@ import {
   v2RateLimits,
 } from '@/lib/api/server/routes'
 import { presentProjectFileVersionContent } from '@/lib/projects/files/api'
-import { readProjectFileVersionContent } from '@/lib/projects/files/application'
+import { downloadProjectFileVersionContent } from '@/lib/projects/files/application'
 import { projectFileOperations } from '@/lib/projects/files/application/operations'
 export const GET = defineV2BinaryRoute({
   contract: v2ReadProjectFileVersionContentContract,
@@ -20,6 +20,6 @@ export const GET = defineV2BinaryRoute({
     fileId: params.fileId,
     version: params.version,
   }),
-  useCase: readProjectFileVersionContent,
+  useCase: downloadProjectFileVersionContent,
   present: presentProjectFileVersionContent,
 })

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-provider.test.ts

@@ -122,6 +122,76 @@ function syncStep1Frame(doc: Y.Doc): Uint8Array {
 }
 
 describe('FileDocProvider', () => {
+  it('never replays rejected document state after write regrant', async () => {
+    vi.useFakeTimers()
+    const old = createProvider()
+    const server = new Y.Doc()
+    server.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.docIdKey, 'doc-1')
+    server.getText('body').insert(0, 'accepted')
+    Y.applyUpdate(old.doc, Y.encodeStateAsUpdate(server), old.provider)
+    const reset = vi.fn()
+    old.provider.on('write-access-lost', reset)
+    try {
+      acceptJoin(old.fire, old.doc.clientID, 'doc-1')
+      await vi.advanceTimersByTimeAsync(0)
+      old.doc.getText('body').insert(8, '-rejected')
+      await vi.advanceTimersByTimeAsync(UPDATE_BATCH_TEST_WINDOW_MS)
+      old.fire(FILE_DOC_EVENTS.PERMISSION, { fileId: 'file-1', canWrite: false })
+      old.fire(FILE_DOC_EVENTS.PERMISSION, { fileId: 'file-1', canWrite: true })
+      expect(old.provider.canWrite).toBe(false)
+      old.emit.mockClear()
+      old.fire(FILE_DOC_EVENTS.MESSAGE, syncStep1Frame(server))
+      old.doc.getText('body').insert(0, 'also-rejected-')
+      await vi.advanceTimersByTimeAsync(UPDATE_BATCH_TEST_WINDOW_MS)
+      expect(reset).toHaveBeenCalledTimes(1)
+      expect(emittedMessages(old.emit)).toHaveLength(0)
+      expect(
+        old.emit.mock.calls.filter(([event]) => event === FILE_DOC_EVENTS.UPDATE)
+      ).toHaveLength(0)
+      old.provider.destroy()
+      const journal = new PendingFileDocUpdateJournal({ ...workspaceScope, fileId: 'file-1' })
+      expect(await journal.load('doc-1')).toBeNull()
+      expect(server.getText('body').toString()).toBe('accepted')
+      const fresh = createProvider()
+      try {
+        fresh.fire(FILE_DOC_EVENTS.JOIN_SUCCESS, {
+          schemaVersion: FILE_DOC_SCHEMA_VERSION,
+          fileId: 'file-1',
+          clientId: fresh.doc.clientID,
+          docId: 'doc-1',
+          acknowledgedUpdates: true,
+          canWrite: false,
+        })
+        await vi.advanceTimersByTimeAsync(0)
+        const frame = encoding.createEncoder()
+        encoding.writeVarUint(frame, FILE_DOC_MESSAGE_TYPE.SYNC)
+        syncProtocol.writeSyncStep2(frame, server)
+        fresh.fire(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(frame))
+        expect(fresh.doc.getText('body').toString()).toBe('accepted')
+        expect(fresh.provider.canWrite).toBe(false)
+        server.getText('body').insert(8, '-peer')
+        const update = encoding.createEncoder()
+        encoding.writeVarUint(update, FILE_DOC_MESSAGE_TYPE.SYNC)
+        syncProtocol.writeUpdate(update, Y.encodeStateAsUpdate(server))
+        fresh.fire(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(update))
+        expect(fresh.doc.getText('body').toString()).toBe('accepted-peer')
+        fresh.fire(FILE_DOC_EVENTS.PERMISSION, { fileId: 'file-1', canWrite: true })
+        expect(fresh.provider.canWrite).toBe(true)
+        expect(fresh.doc.getText('body').toString()).not.toContain('rejected')
+      } finally {
+        fresh.provider.destroy()
+        fresh.awareness.destroy()
+        fresh.doc.destroy()
+      }
+    } finally {
+      old.provider.destroy()
+      old.awareness.destroy()
+      old.doc.destroy()
+      server.destroy()
+      vi.useRealTimers()
+    }
+  })
+
   it('sends the workspace owner while accepting deployed ownerless join replies', async () => {
     vi.useFakeTimers()
     const { socket, emit, fire } = createSocket(true)