fix(file-search): send legacy workspace index payloads, isolate realtime test DB, best-effort post-upload Project check (#9041)

* fix(file-search): send legacy workspace index payloads, isolate realtime test DB, best-effort post-upload Project check

- Workspace index jobs use the { workspaceId } wire shape so a rolled-back worker can decode them; Project jobs keep { owner }
- Realtime integration setup clears inherited role-specific DB URLs and runs as the realtime role on the test database, via a shared clearInheritedDatabaseUrls helper
- After a successful local upload write, a failed retired-Project check logs and returns 204 instead of 500

* fix(uploads): name the expired-session sweep as the post-write cleanup backstop

Author: waleedlatif1

apps/realtime/vitest.integration.setup.ts

@@ -1,10 +1,16 @@
-import { readTestDatabaseUrl, readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
+import {
+  clearInheritedDatabaseUrls,
+  readTestDatabaseUrl,
+  readTestRedisUrl,
+} from '@sim/db/testing/test-infrastructure'
 
 const redisUrl = readTestRedisUrl()
 if (!redisUrl) throw new Error('Realtime integration tests require disposable TEST_REDIS_URL')
 
+clearInheritedDatabaseUrls()
 Object.assign(process.env, {
   NODE_ENV: 'test',
+  SIM_DB_ROLE: 'realtime',
   DATABASE_URL: readTestDatabaseUrl(),
   REDIS_URL: redisUrl,
   BETTER_AUTH_URL: 'http://localhost:3000',

apps/sim/app/api/v2/uploads/[uploadId]/parts/[partNumber]/route.test.ts

@@ -1,3 +1,4 @@
+import { dbChainMockFns } from '@sim/testing/mocks/database.mock'
 import { uploadSessionMock, uploadSessionMockFns } from '@sim/testing/mocks/upload-session.mock'
 import { NextRequest } from 'next/server'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
@@ -18,6 +19,7 @@ vi.mock('@/lib/uploads/upload-session/provider', () => ({
 vi.mock('@/lib/uploads/upload-session/service', () => uploadSessionMock)
 
 import { OrchestrationError } from '@/lib/core/orchestration/types'
+import { PROJECT_FILE_UPLOAD_BINDING_KEY } from '@/lib/uploads/upload-session/types'
 import { PUT } from '@/app/api/v2/uploads/[uploadId]/parts/[partNumber]/route'
 
 const mockExpectedUploadPartSize = uploadSessionMockFns.mockExpectedUploadPartSize
@@ -54,6 +56,24 @@ describe('PUT /api/v2/uploads/[uploadId]/parts/[partNumber]', () => {
     })
   })
 
+  it('answers 204 when the retired-Project check fails after the part is written', async () => {
+    mockVerifyUploadSessionToken.mockReturnValue({
+      ...SESSION,
+      purpose: 'project_file',
+      storageProvider: 'local',
+      providerUploadId: null,
+      finalKey: 'project/project-1/file.bin',
+      metadata: { [PROJECT_FILE_UPLOAD_BINDING_KEY]: { entityId: 'project-1' } },
+    })
+    dbChainMockFns.transaction.mockRejectedValueOnce(new Error('database unavailable'))
+
+    const response = await request()
+
+    expect(response.status).toBe(204)
+    expect(mockWriteLocalMultipartPart).toHaveBeenCalledOnce()
+    expect(dbChainMockFns.transaction).toHaveBeenCalledOnce()
+  })
+
   it('maps a streamed-size failure to 400', async () => {
     mockWriteLocalMultipartPart.mockRejectedValue(
       new MockLocalUploadBodyError('Part 1 has 2 bytes; expected 3')

apps/sim/app/api/v2/uploads/[uploadId]/parts/[partNumber]/route.ts

@@ -105,8 +105,17 @@ export const PUT = withRouteHandler(
       }
       throw error
     }
-    if (await queueRetiredProjectUploadCleanup(session))
-      return v2Error('CONFLICT', 'The Project no longer exists')
+    try {
+      if (await queueRetiredProjectUploadCleanup(session))
+        return v2Error('CONFLICT', 'The Project no longer exists')
+    } catch (error) {
+      // The object is already written. The expired-session sweep cleans it up, as will completion's
+      // Project re-authorization once Project uploads can complete.
+      logger.warn('Could not check for a retired Project after upload', {
+        uploadId: session.id,
+        error,
+      })
+    }
     return new NextResponse(null, { status: 204 })
   },
   {

apps/sim/app/api/v2/uploads/[uploadId]/route.test.ts

@@ -1,3 +1,4 @@
+import { dbChainMockFns } from '@sim/testing/mocks/database.mock'
 import { uploadSessionMock, uploadSessionMockFns } from '@sim/testing/mocks/upload-session.mock'
 import { NextRequest } from 'next/server'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
@@ -17,6 +18,7 @@ vi.mock('@/lib/uploads/upload-session/provider', () => ({
 
 vi.mock('@/lib/uploads/upload-session/service', () => uploadSessionMock)
 
+import { PROJECT_FILE_UPLOAD_BINDING_KEY } from '@/lib/uploads/upload-session/types'
 import { PUT } from '@/app/api/v2/uploads/[uploadId]/route'
 
 const { mockGetOwnedUploadSession, mockUploadSessionObjectMetadata: mockMetadata } =
@@ -117,6 +119,23 @@ describe('PUT /api/v2/uploads/[uploadId]', () => {
     expect(mockWriteLocalPut).not.toHaveBeenCalled()
   })
 
+  it('answers 204 when the retired-Project check fails after the object is written', async () => {
+    mockGetOwnedUploadSession.mockReturnValue({
+      ...SESSION,
+      purpose: 'project_file',
+      storageContext: 'project',
+      finalKey: 'project/project-1/file.bin',
+      metadata: { [PROJECT_FILE_UPLOAD_BINDING_KEY]: { entityId: 'project-1' } },
+    })
+    dbChainMockFns.transaction.mockRejectedValueOnce(new Error('database unavailable'))
+
+    const response = await request()
+
+    expect(response.status).toBe(204)
+    expect(mockWriteLocalPut).toHaveBeenCalledOnce()
+    expect(dbChainMockFns.transaction).toHaveBeenCalledOnce()
+  })
+
   it('maps exact-size streaming failures to a caller error', async () => {
     mockWriteLocalPut.mockRejectedValue(new MockLocalUploadBodyError('Upload exceeds 3 bytes'))
 

apps/sim/app/api/v2/uploads/[uploadId]/route.ts

@@ -95,8 +95,17 @@ export const PUT = withRouteHandler(
       }
       throw error
     }
-    if (await queueRetiredProjectUploadCleanup(session))
-      return v2Error('CONFLICT', 'The Project no longer exists')
+    try {
+      if (await queueRetiredProjectUploadCleanup(session))
+        return v2Error('CONFLICT', 'The Project no longer exists')
+    } catch (error) {
+      // The object is already written. The expired-session sweep cleans it up, as will completion's
+      // Project re-authorization once Project uploads can complete.
+      logger.warn('Could not check for a retired Project after upload', {
+        uploadId: session.id,
+        error,
+      })
+    }
     return new NextResponse(null, { status: 204 })
   },
   {

apps/sim/lib/workspace-files/search/dispatcher.integration.ts

@@ -261,9 +261,9 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
 
       const first = await prepareWorkspaceFileSearchDispatch()
       const second = await prepareWorkspaceFileSearchDispatch()
-      expect([...first.payloads, ...second.payloads].map((payload) => payload.owner)).toEqual([
-        { entityType: 'workspace', entityId: 'later-workspace' },
-      ])
+      const claimed = [...first.payloads, ...second.payloads]
+      expect(claimed.map((payload) => payload.workspaceId)).toEqual(['later-workspace'])
+      expect(claimed[0]).not.toHaveProperty('owner')
       const [queued] =
         await connection`SELECT count(*)::int AS count FROM file_search_dispatch_queue
         WHERE entity_type = 'project'`
@@ -272,9 +272,10 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
       await holder`COMMIT`
       const resumed = await prepareWorkspaceFileSearchDispatch()
       expect(resumed.payloads.length).toBeGreaterThan(0)
-      expect(resumed.payloads.every((payload) => payload.owner?.entityType === 'project')).toBe(
-        true
-      )
+      for (const payload of resumed.payloads) {
+        expect(payload.owner?.entityType).toBe('project')
+        expect(payload).not.toHaveProperty('workspaceId')
+      }
     } finally {
       await holder`ROLLBACK`
       holder.release()
@@ -290,17 +291,16 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
       await holder`SELECT pg_advisory_xact_lock(hashtextextended('project:busy-project', 0))`
       const result = await prepareWorkspaceFileSearchDispatch()
       expect(result.payloads).toHaveLength(1)
-      expect(result.payloads[0].owner).toEqual({
-        entityType: 'workspace',
-        entityId: 'free-workspace',
-      })
+      expect(result.payloads[0].workspaceId).toBe('free-workspace')
+      expect(result.payloads[0]).not.toHaveProperty('owner')
       const [waiting] = await connection`SELECT count(*)::int AS count
         FROM file_search_dispatch_queue WHERE entity_type = 'project' AND entity_id = 'busy-project'`
       expect(waiting.count).toBe(1)
       await holder`COMMIT`
       const resumed = await prepareWorkspaceFileSearchDispatch()
       expect(resumed.payloads).toHaveLength(1)
       expect(resumed.payloads[0].owner).toEqual({ entityType: 'project', entityId: 'busy-project' })
+      expect(resumed.payloads[0]).not.toHaveProperty('workspaceId')
     } finally {
       await holder`ROLLBACK`
       holder.release()
@@ -311,13 +311,16 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
     await seedQueue('shared-id', 2, 1)
     await seedQueue('shared-id', 2, 0, 'project')
     const result = await prepareWorkspaceFileSearchDispatch()
-    expect(
-      result.payloads.filter((payload) => payload.owner?.entityType === 'workspace')
-    ).toHaveLength(1)
-    expect(
-      result.payloads.filter((payload) => payload.owner?.entityType === 'project')
-    ).toHaveLength(2)
-    expect(result.payloads.every((payload) => payload.owner?.entityId === 'shared-id')).toBe(true)
+    const scopes = result.payloads.map((payload) =>
+      'workspaceId' in payload
+        ? `legacy:${payload.workspaceId}`
+        : `${payload.owner?.entityType}:${payload.owner?.entityId}`
+    )
+    expect([...scopes].sort()).toEqual([
+      'legacy:shared-id',
+      'project:shared-id',
+      'project:shared-id',
+    ])
     const [invalid] =
       await connection`SELECT count(*)::int AS count FROM workspace_file_search_revision
       WHERE entity_type = 'project' AND workspace_id IS NOT NULL`
@@ -775,7 +778,7 @@ describe('workspace file search dispatch PostgreSQL deadlines', () => {
         payload: {
           dispatchToken: expect.any(String),
           fileId,
-          owner: { entityType: 'workspace', entityId: workspaceId },
+          workspaceId,
           sourceContentUpdatedAt: '2026-09-16T00:00:00.000Z',
         },
       }),

apps/sim/lib/workspace-files/search/dispatcher.ts

@@ -62,6 +62,7 @@ import {
 import { lockFileSearchOwners } from '@/lib/workspace-files/search/owner-policy'
 import {
   fileSearchOwnerFields,
+  fileSearchOwnerScope,
   resolveFileSearchOwner,
   staleFileSearchDependencyBuilds,
 } from '@/lib/workspace-files/search/scope'
@@ -510,7 +511,7 @@ async function claimQueuedOwnerJobs(
     .delete(fileSearchDispatchQueue)
     .where(and(queuedOwnerFilter(owners), notExists(pendingRevision), ...keepsStaleRevision))
   return rows.map((row) => ({
-    owner: rowOwner(row),
+    ...fileSearchOwnerScope(rowOwner(row)),
     fileId: row.fileId,
     sourceContentUpdatedAt: new Date(row.sourceContentUpdatedAt).toISOString(),
     dispatchToken: now.toISOString(),

apps/sim/lib/workspace-files/search/scope.ts

@@ -35,6 +35,14 @@ export function resolveFileSearchOwner(input: unknown): EditableFileOwner {
   throw new Error('Invalid file search owner')
 }
 
+/**
+ * Encodes an owner for an index job. Workspace jobs keep the legacy `{ workspaceId }` shape so a
+ * rolled-back worker can still decode them; only Project jobs carry `owner`.
+ */
+export function fileSearchOwnerScope(owner: EditableFileOwner): FileSearchOwnerScope {
+  return owner.entityType === 'workspace' ? { workspaceId: owner.entityId } : { owner }
+}
+
 /** Writes canonical owner columns and the legacy workspace key used by deployed workers. */
 export function fileSearchOwnerFields(owner: EditableFileOwner) {
   return {

apps/sim/vitest.integration.setup.ts

@@ -1,6 +1,10 @@
 import { readFileSync } from 'node:fs'
 import { createServer } from 'node:http'
-import { readTestDatabaseUrl, readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
+import {
+  clearInheritedDatabaseUrls,
+  readTestDatabaseUrl,
+  readTestRedisUrl,
+} from '@sim/db/testing/test-infrastructure'
 import { afterAll } from 'vitest'
 
 /**
@@ -20,15 +24,7 @@ const environmentSource = readFileSync(new URL('./lib/core/config/env.ts', impor
 for (const entry of environmentSource.matchAll(/^\s+([A-Z][A-Z0-9_]*)\s*:/gm)) {
   delete process.env[entry[1]]
 }
-for (const key of Object.keys(process.env)) {
-  if (
-    key.startsWith('DATABASE_URL') ||
-    key.startsWith('DATABASE_REPLICA_URL') ||
-    key === 'MIGRATION_DATABASE_URL'
-  ) {
-    delete process.env[key]
-  }
-}
+clearInheritedDatabaseUrls()
 Object.assign(process.env, {
   NODE_ENV: 'test',
   DATABASE_URL: databaseUrl,

packages/db/testing/test-infrastructure.ts

@@ -38,6 +38,22 @@ export function readTestDatabaseUrl(): string {
   return value
 }
 
+/**
+ * Deletes every inherited database URL. `@sim/db` prefers a role-specific `DATABASE_URL_<ROLE>`
+ * over `DATABASE_URL`, so one left in the shell would bypass the test database.
+ */
+export function clearInheritedDatabaseUrls(): void {
+  for (const key of Object.keys(process.env)) {
+    if (
+      key.startsWith('DATABASE_URL') ||
+      key.startsWith('DATABASE_REPLICA_URL') ||
+      key === 'MIGRATION_DATABASE_URL'
+    ) {
+      delete process.env[key]
+    }
+  }
+}
+
 /**
  * Returns the validated `TEST_REDIS_URL`, or undefined when a local run has no Redis. In CI a
  * missing value throws: every Redis-gated suite would otherwise pass by skipping.