diff --git a/apps/sim/lib/core/application/workspace-authorization.ts b/apps/sim/lib/core/application/workspace-authorization.ts index 721c59d7007..67de9320502 100644 --- a/apps/sim/lib/core/application/workspace-authorization.ts +++ b/apps/sim/lib/core/application/workspace-authorization.ts @@ -191,7 +191,10 @@ export function requireAllowedWorkspacePrincipal( } } -function requirePermission(permission: PermissionType | null, required: PermissionType): void { +function requirePermission( + permission: PermissionType | null, + required: PermissionType +): asserts permission is PermissionType { if (permission === null) { throw new NoWorkspaceAccessError() } @@ -324,7 +327,7 @@ export async function requireCurrentHumanRole -): Promise { +): Promise { const permission = await resolveEffectiveWorkspacePermission( userId, context.workspaceId, @@ -333,6 +336,7 @@ export async function requireCurrentHumanRole; -export const MemoryNotesAccessResult = z.strictObject({ +export const MemoryNotesAccessResult = z.object({ success: z.literal(true), version: z.literal(1), - actor: z.strictObject({ userId: z.string().min(1), chatId: z.uuid() }), + actor: z.object({ userId: z.string().min(1), chatId: z.uuid() }), target: z.discriminatedUnion("scope", [ - z.strictObject({ + z.object({ scope: z.literal("personal"), permissions: permissionsSchema.extend({ sharedMutation: z.literal(false) }), }), projectSchema, - z.strictObject({ + z.object({ scope: z.literal("workspace"), id: z.uuid(), name: z.string(), @@ -55,4 +56,4 @@ export const MemoryCleanupInput = z.strictObject({ ]), }); export type MemoryCleanupInput = z.infer; -export const MemoryCleanupResult = z.strictObject({ success: z.literal(true) }); +export const MemoryCleanupResult = z.object({ success: z.literal(true) }); diff --git a/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts b/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts index 9473a1b62ee..72a6033d0a5 100644 --- a/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts +++ b/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts @@ -12,6 +12,7 @@ import { } from '@sim/db/schema' import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' import { featureFlagsMock, featureFlagsMockFns } from '@sim/testing/mocks/feature-flags.mock' +import { sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import { and, eq, inArray } from 'drizzle-orm' import { NextRequest } from 'next/server' @@ -26,7 +27,7 @@ import { resolveMemoryNotesAccess, } from '@/lib/mothership/memories/application/resolve-access' import { executeSimControl } from '@/lib/mothership/transport/control' -import { splitForkProject } from '@/lib/projects/membership' +import { lockProject, splitForkProject } from '@/lib/projects/membership' import { POST } from '@/app/api/mothership/memories/access/route' vi.mock('@/lib/core/config/feature-flags', () => featureFlagsMock) @@ -255,6 +256,41 @@ it('resolves overlapping workspace checks that share one Project without deadloc expect(result.target).toMatchObject({ scope: 'workspace', currentProject: { id: projectId } }) }) +it('answers while a writer holds the account row and the Project exclusively', async () => { + let release = () => {} + const released = new Promise((resolve) => { + release = resolve + }) + let held = () => {} + const holding = new Promise((resolve) => { + held = resolve + }) + const writer = db.transaction(async (tx) => { + await tx.select({ id: user.id }).from(user).where(eq(user.id, actorId)).for('update') + await lockProject(tx, projectId) + held() + await released + }) + await holding + try { + for (const organizationChat of [false, true]) { + const result = await Promise.race([ + resolveMemoryNotesAccess.execute({ + principal: principal(organizationChat), + input: { target: { scope: 'workspace', workspaceId }, action: 'read' }, + }), + sleep(2_000).then(() => { + throw new Error('Memory access waited on a writer') + }), + ]) + expect(result.target).toMatchObject({ currentProject: { id: projectId } }) + } + } finally { + release() + await writer + } +}) + it('does not turn write in one environment into shared Project write', async () => { await db .update(permissions) diff --git a/apps/sim/lib/mothership/memories/application/resolve-access.ts b/apps/sim/lib/mothership/memories/application/resolve-access.ts index db6b6592676..a52ba63bc29 100644 --- a/apps/sim/lib/mothership/memories/application/resolve-access.ts +++ b/apps/sim/lib/mothership/memories/application/resolve-access.ts @@ -1,15 +1,12 @@ import type { OrganizationDelegatedPrincipal, SubjectDelegatedPrincipal } from '@sim/auth/principal' import { db } from '@sim/db' -import { copilotChats, workspace } from '@sim/db/schema' +import { copilotChats, project, workspace } from '@sim/db/schema' +import { permissionSatisfies } from '@sim/platform-authz/workspace' import { and, eq, isNull } from 'drizzle-orm' import { defineWorkspaceOperation } from '@/lib/core/application' import { authorizeOrganizationOperation } from '@/lib/core/application/organization-authorization' import { defineOrganizationOperation } from '@/lib/core/application/organization-operation' -import { - InsufficientWorkspacePermissionsError, - requireCurrentHumanRole, -} from '@/lib/core/application/workspace-authorization' -import { isFeatureEnabled } from '@/lib/core/config/feature-flags' +import { requireCurrentHumanRole } from '@/lib/core/application/workspace-authorization' import { OrchestrationError } from '@/lib/core/orchestration/types' import type { DbTransaction } from '@/lib/db/types' import { defineAuthorizedChatUseCase } from '@/lib/mothership/chat/application/authorized-chat-use-case' @@ -21,15 +18,13 @@ import type { } from '@/lib/mothership/generated/memory-notes-access' import { assertWorkspaceCapability } from '@/lib/permission-groups/capability-assertions' import { PermissionGroupCapabilityError } from '@/lib/permission-groups/capability-error' -import { acquirePermissionGroupOrgReadLock } from '@/lib/permission-groups/locks' import { loadProjectAccess } from '@/lib/projects/application/authorization' import { requireCurrentCopilotProjectInvocation, requireProjectFileOwnerCapabilities, requireProjectFileOwnerRole, } from '@/lib/projects/files/application/authorization' -import { lockWorkspaceProject } from '@/lib/projects/membership' -import { requireProjectApiEnabled } from '@/lib/projects/rollout.server' +import { isProjectsEnabled, requireProjectApiEnabled } from '@/lib/projects/rollout.server' import { requireCurrentFileSubject } from '@/lib/workspace-files/application/subject' export const MEMORY_NOTES_ACCESS_AUDIENCE = 'sim:memory-access' @@ -52,6 +47,18 @@ const organizationOperation = defineOrganizationOperation({ delegationAudience: MEMORY_NOTES_ACCESS_AUDIENCE, }) type MemoryPrincipal = SubjectDelegatedPrincipal | OrganizationDelegatedPrincipal +/** The worker acts on the answer after commit, so a consistent unlocked snapshot is enough. */ +const READ_SNAPSHOT = { isolationLevel: 'repeatable read', accessMode: 'read only' } as const + +async function workspaceProject(tx: DbTransaction, workspaceId: string) { + const [row] = await tx + .select({ id: project.id, organizationId: project.organizationId }) + .from(workspace) + .innerJoin(project, eq(project.id, workspace.projectId)) + .where(eq(workspace.id, workspaceId)) + .limit(1) + return row ?? null +} async function workspaceAccess(tx: DbTransaction, userId: string, workspaceId: string) { const [target] = await tx @@ -63,10 +70,9 @@ async function workspaceAccess(tx: DbTransaction, userId: string, workspaceId: s }) .from(workspace) .where(and(eq(workspace.id, workspaceId), isNull(workspace.archivedAt))) - .for('share') .limit(1) if (!target) throw new OrchestrationError('not_found', 'Workspace not found') - await requireCurrentHumanRole(userId, target, 'read', { executor: tx }) + const permission = await requireCurrentHumanRole(userId, target, 'read', { executor: tx }) await assertWorkspaceCapability( userId, workspaceId, @@ -74,14 +80,7 @@ async function workspaceAccess(tx: DbTransaction, userId: string, workspaceId: s target.workspaceOrganizationId, tx ) - let sharedMutation = true - try { - await requireCurrentHumanRole(userId, target, 'write', { executor: tx }) - } catch (error) { - if (!(error instanceof InsufficientWorkspacePermissionsError)) throw error - sharedMutation = false - } - return { target, sharedMutation } + return { target, sharedMutation: permissionSatisfies(permission, 'write') } } async function projectAccess( @@ -99,12 +98,10 @@ async function projectAccess( ...(scope.organizationId ? { organizationId: scope.organizationId } : {}), ...(scope.workspaceId ? { workspaceId: scope.workspaceId } : {}), }, - 'shared' + 'snapshot' ) if (scope.organizationId !== access.record.organizationId) throw new OrchestrationError('not_found', 'Project not found in this conversation') - if (access.record.organizationId) - await acquirePermissionGroupOrgReadLock(tx, access.record.organizationId) await requireCurrentCopilotProjectInvocation( tx, { @@ -112,9 +109,10 @@ async function projectAccess( invocation: { kind: 'chat', chatId }, }, access, - scope + scope, + 'snapshot' ) - const owner = await requireProjectFileOwnerRole(tx, principal, access, 'read') + const owner = await requireProjectFileOwnerRole(tx, principal, access, 'read', 'snapshot') let sharedMutation = owner.canWrite if (sharedMutation) { try { @@ -167,8 +165,9 @@ const resolveAccess = defineAuthorizedChatUseCase({ }, }, async execute({ principal, context, input }): Promise { + const projectsEnabled = input.target.scope !== 'personal' && (await isProjectsEnabled()) return db.transaction(async (tx) => { - await requireCurrentFileSubject(tx, principal) + await requireCurrentFileSubject(tx, principal, 'snapshot') const [chat] = await tx .select({ userId: copilotChats.userId, @@ -179,7 +178,6 @@ const resolveAccess = defineAuthorizedChatUseCase({ }) .from(copilotChats) .where(and(eq(copilotChats.id, context.chatId), isNull(copilotChats.deletedAt))) - .for('share') .limit(1) if ( !chat || @@ -204,9 +202,11 @@ const resolveAccess = defineAuthorizedChatUseCase({ } const origin = chat.workspaceId && input.target.scope !== 'personal' - ? await lockWorkspaceProject(tx, chat.workspaceId) + ? await workspaceProject(tx, chat.workspaceId) : null - if (chat.workspaceId) await workspaceAccess(tx, principal.subjectUserId, chat.workspaceId) + const originAccess = chat.workspaceId + ? await workspaceAccess(tx, principal.subjectUserId, chat.workspaceId) + : null const scope = { organizationId: chat.organizationId ?? origin?.organizationId ?? null, ...(chat.workspaceId ? { workspaceId: chat.workspaceId } : {}), @@ -218,7 +218,7 @@ const resolveAccess = defineAuthorizedChatUseCase({ permissions: { read: true, privateMutation: true, sharedMutation: false }, } } else if (input.target.scope === 'project') { - await requireProjectApiEnabled() + if (!projectsEnabled) await requireProjectApiEnabled() if (chat.workspaceId && origin?.id !== input.target.projectId) throw new OrchestrationError('not_found', 'Project not found in this conversation') target = await projectAccess(tx, principal, context.chatId, input.target.projectId, scope) @@ -227,13 +227,15 @@ const resolveAccess = defineAuthorizedChatUseCase({ throw new OrchestrationError('not_found', 'Workspace not found in this conversation') const membership = chat.workspaceId ? origin - : await lockWorkspaceProject(tx, input.target.workspaceId) + : await workspaceProject(tx, input.target.workspaceId) if (!membership || membership.organizationId !== scope.organizationId) throw new OrchestrationError('not_found', 'Workspace not found in this conversation') - const access = await workspaceAccess(tx, principal.subjectUserId, input.target.workspaceId) + const access = + originAccess ?? + (await workspaceAccess(tx, principal.subjectUserId, input.target.workspaceId)) if (access.target.workspaceOrganizationId !== scope.organizationId) throw new OrchestrationError('not_found', 'Workspace not found in this conversation') - const currentProject = (await isFeatureEnabled('projects')) + const currentProject = projectsEnabled ? await projectAccess(tx, principal, context.chatId, membership.id, scope) : null target = { @@ -252,7 +254,7 @@ const resolveAccess = defineAuthorizedChatUseCase({ actor: { userId: principal.subjectUserId, chatId: context.chatId }, target, } - }) + }, READ_SNAPSHOT) }, }) diff --git a/apps/sim/lib/mothership/memories/cleanup-outbox.test.ts b/apps/sim/lib/mothership/memories/cleanup-outbox.test.ts index 57603f42054..25c1da8673b 100644 --- a/apps/sim/lib/mothership/memories/cleanup-outbox.test.ts +++ b/apps/sim/lib/mothership/memories/cleanup-outbox.test.ts @@ -15,7 +15,7 @@ vi.mock('@/lib/mothership/server/agent-url', () => mothershipAgentUrlMock) import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox' -const { mockEnqueueOutboxEvent } = outboxServiceMockFns +const { mockEnqueueOutboxEvents } = outboxServiceMockFns const executor = { insert: vi.fn() } const targets = [ { kind: 'private_owner' as const, userId: 'user-1' }, @@ -35,17 +35,6 @@ describe('enqueueMemoryCleanup', () => { it('queues nothing when this Sim has no worker credential', async () => { await enqueueMemoryCleanup(executor, targets) - expect(mockEnqueueOutboxEvent).not.toHaveBeenCalled() - }) - - it('queues one credential-bound event per destination and target', async () => { - setEnv({ COPILOT_API_KEY: 'configured-key' }) - await enqueueMemoryCleanup(executor, targets) - expect(mockEnqueueOutboxEvent).toHaveBeenCalledTimes(targets.length) - for (const [, , payload] of mockEnqueueOutboxEvent.mock.calls) - expect(payload).toMatchObject({ - baseUrl: 'https://agent.example.test', - credentialFingerprint: expect.any(String), - }) + expect(mockEnqueueOutboxEvents).not.toHaveBeenCalled() }) }) diff --git a/apps/sim/lib/mothership/memories/cleanup-outbox.ts b/apps/sim/lib/mothership/memories/cleanup-outbox.ts index 3dbaa16ef55..eb67bcf1267 100644 --- a/apps/sim/lib/mothership/memories/cleanup-outbox.ts +++ b/apps/sim/lib/mothership/memories/cleanup-outbox.ts @@ -1,8 +1,15 @@ -import { createHash } from 'node:crypto' import type { db } from '@sim/db' +import { createLogger } from '@sim/logger' +import { chunkArray } from '@sim/utils/helpers' +import { omit } from '@sim/utils/object' import { z } from 'zod' import { env } from '@/lib/core/config/env' -import { enqueueOutboxEvent, type OutboxHandlerRegistry } from '@/lib/core/outbox/service' +import { + deferOutboxHandler, + enqueueOutboxEvents, + MAX_BULK_ENQUEUE_EVENTS, + type OutboxHandlerRegistry, +} from '@/lib/core/outbox/service' import { SIM_AGENT_API_URL } from '@/lib/mothership/constants' import { MemoryCleanupInput, @@ -11,13 +18,28 @@ import { import { MEMORY_CLEANUP_EVENT } from '@/lib/mothership/memories/cleanup-event' import { fetchGo } from '@/lib/mothership/request/go/fetch' import { mothershipRequestHeaders } from '@/lib/mothership/request/headers' -import { getMothershipSourceEnvHeaders } from '@/lib/mothership/server/agent-url' +import { + getMothershipSourceEnvHeaders, + MOTHERSHIP_SOURCE_ENV_HEADER, +} from '@/lib/mothership/server/agent-url' + +const logger = createLogger('MemoryCleanup') + +/** Waits for a worker or configuration that is not there yet without spending the retry budget. */ +const UNAVAILABLE_RETRY_MS = 60 * 60 * 1000 +/** + * Statuses meaning the worker does not serve cleanup yet, not that this request failed: a worker + * that predates the route answers 404 to the hosted key and 401 to a self-hosted customer key. + */ +const ENDPOINT_MISSING_STATUSES = new Set([401, 404, 405, 501]) -const cleanupPayloadSchema = z.strictObject({ +/** Non-strict so rows enqueued with the retired `credentialFingerprint` field still parse. */ +const cleanupPayloadSchema = z.object({ request: MemoryCleanupInput, baseUrl: z.url(), - credentialFingerprint: z.string().nullable(), - sourceHeaders: z.record(z.string(), z.string()), + sourceHeaders: z.object({ + [MOTHERSHIP_SOURCE_ENV_HEADER]: z.enum(['dev', 'staging', 'prod']).optional(), + }), }) function destinations(): string[] { @@ -31,10 +53,6 @@ function destinations(): string[] { ] } -function credentialFingerprint(): string | null { - return env.COPILOT_API_KEY ? createHash('sha256').update(env.COPILOT_API_KEY).digest('hex') : null -} - /** * Captures configured routes before deletion; notes can outlive a user's environment selection. * Without `COPILOT_API_KEY` this Sim never reached a worker, so it holds no notes to erase and @@ -44,50 +62,55 @@ export async function enqueueMemoryCleanup( executor: Pick, targets: MemoryCleanupInput['target'][] ): Promise { - const fingerprint = credentialFingerprint() - if (!fingerprint) return + if (!env.COPILOT_API_KEY) return const sourceHeaders = getMothershipSourceEnvHeaders() - for (const baseUrl of destinations()) { - for (const target of targets) { - await enqueueOutboxEvent( - executor, - MEMORY_CLEANUP_EVENT, - cleanupPayloadSchema.parse({ - request: { version: 1, target }, - baseUrl, - credentialFingerprint: fingerprint, - sourceHeaders, - }) - ) - } + const payloads = destinations().flatMap((baseUrl) => + targets.map((target) => + cleanupPayloadSchema.parse({ request: { version: 1, target }, baseUrl, sourceHeaders }) + ) + ) + for (const batch of chunkArray(payloads, MAX_BULK_ENQUEUE_EVENTS)) { + await enqueueOutboxEvents(executor, MEMORY_CLEANUP_EVENT, batch) } } +function waitForWorker(reason: string, eventId: string, detail?: Record) { + logger.warn(`Memory cleanup waiting: ${reason}`, { eventId, ...detail }) + return deferOutboxHandler(reason, UNAVAILABLE_RETRY_MS, false) +} + +/** + * Sends the processor's current key only to a destination the processor itself configures, and + * replays the enqueuing deployment's source env so the worker erases that deployment's notes. + * A missing key, destination, or worker endpoint waits without consuming attempts: erasure must + * not dead-letter while configuration or the worker rollout catches up. + */ export const memoryCleanupOutboxHandlers = { [MEMORY_CLEANUP_EVENT]: async (rawPayload, context) => { const payload = cleanupPayloadSchema.parse(rawPayload) - if ( - !payload.credentialFingerprint || - payload.credentialFingerprint !== credentialFingerprint() - ) { - throw new Error('Memory cleanup credential binding changed; operator reconciliation required') - } - if ( - !destinations().includes(payload.baseUrl) || - JSON.stringify(payload.sourceHeaders) !== JSON.stringify(getMothershipSourceEnvHeaders()) - ) { - throw new Error( - 'Memory cleanup destination binding changed; operator reconciliation required' - ) - } + if (!env.COPILOT_API_KEY) + return waitForWorker('COPILOT_API_KEY is not configured', context.eventId) + if (!destinations().includes(payload.baseUrl)) + return waitForWorker('destination is not configured', context.eventId, { + baseUrl: payload.baseUrl, + }) + const sourceEnv = payload.sourceHeaders[MOTHERSHIP_SOURCE_ENV_HEADER] const response = await fetchGo(`${payload.baseUrl}/api/memories/cleanup`, { method: 'POST', - headers: mothershipRequestHeaders(), + headers: { + ...omit(mothershipRequestHeaders(), [MOTHERSHIP_SOURCE_ENV_HEADER]), + ...(sourceEnv ? { [MOTHERSHIP_SOURCE_ENV_HEADER]: sourceEnv } : {}), + }, body: JSON.stringify(payload.request), signal: AbortSignal.any([context.signal, AbortSignal.timeout(10_000)]), redirect: 'error', }) - if (!response.ok) throw new Error(`Memory cleanup failed (${response.status})`) + if (!response.ok) { + await response.body?.cancel().catch(() => {}) + if (ENDPOINT_MISSING_STATUSES.has(response.status)) + return waitForWorker(`endpoint unavailable (${response.status})`, context.eventId) + throw new Error(`Memory cleanup failed (${response.status})`) + } MemoryCleanupResult.parse(await response.json()) }, } satisfies OutboxHandlerRegistry diff --git a/apps/sim/lib/mothership/memories/cleanup.integration.ts b/apps/sim/lib/mothership/memories/cleanup.integration.ts index 211f13bd464..dd361ac1b2a 100644 --- a/apps/sim/lib/mothership/memories/cleanup.integration.ts +++ b/apps/sim/lib/mothership/memories/cleanup.integration.ts @@ -11,13 +11,13 @@ const userId = generateId() const workspaceId = generateId() const projectId = generateId() const received: { body: unknown; key: string | string[] | undefined }[] = [] -let fail = true +let status = 503 const server = createServer(async (req, res) => { let body = '' for await (const chunk of req) body += chunk received.push({ body: JSON.parse(body), key: req.headers['x-api-key'] }) - res.writeHead(fail ? 503 : 200, { 'content-type': 'application/json' }) - res.end(JSON.stringify({ success: !fail })) + res.writeHead(status, { 'content-type': 'application/json' }) + res.end(JSON.stringify({ success: status === 200 })) }) let runtime: { deletion: typeof import('@/lib/users/account-deletion') @@ -117,7 +117,7 @@ it('rolls cleanup back with deletion and retries committed account/target erasur expect( (await pending()).every((event) => event.status === 'pending' && event.attempts === 1) ).toBe(true) - fail = false + status = 200 await db .update(outboxEvent) .set({ availableAt: new Date(0) }) @@ -139,18 +139,33 @@ it('rolls cleanup back with deletion and retries committed account/target erasur expect(received.every((request) => request.key === 'memory-cleanup-fixture-key')).toBe(true) const first = events[0] - await db - .update(outboxEvent) - .set({ - status: 'pending', - availableAt: new Date(0), - payload: { ...toRecord(first.payload), credentialFingerprint: 'changed-credential' }, - }) - .where(and(eq(outboxEvent.id, first.id), eq(outboxEvent.status, 'completed'))) + const replay = async (payload: unknown = first.payload) => { + await db + .update(outboxEvent) + .set({ status: 'pending', availableAt: new Date(0), attempts: 9, payload }) + .where(eq(outboxEvent.id, first.id)) + await runtime.outbox.processOutboxEventById( + first.id, + runtime.cleanup.memoryCleanupOutboxHandlers + ) + const [row] = await db.select().from(outboxEvent).where(eq(outboxEvent.id, first.id)) + return row + } + const waitsAnHour = (row: typeof outboxEvent.$inferSelect) => { + expect(row).toMatchObject({ status: 'pending', attempts: 9 }) + expect(row.availableAt.getTime()).toBeGreaterThan(Date.now() + 50 * 60_000) + } + + status = 404 const before = received.length - await runtime.outbox.processOutboxEventById(first.id, runtime.cleanup.memoryCleanupOutboxHandlers) - expect(received).toHaveLength(before) - const [blocked] = await db.select().from(outboxEvent).where(eq(outboxEvent.id, first.id)) - expect(blocked.status).toBe('pending') - expect(blocked.lastError).toContain('credential binding changed') + waitsAnHour(await replay()) + expect(received).toHaveLength(before + 1) + + status = 200 + waitsAnHour(await replay({ ...toRecord(first.payload), baseUrl: 'https://unconfigured.test' })) + expect(received).toHaveLength(before + 1) + + const legacy = await replay({ ...toRecord(first.payload), credentialFingerprint: 'retired' }) + expect(legacy.status).toBe('completed') + expect(received).toHaveLength(before + 2) }, 30_000) diff --git a/apps/sim/lib/projects/application/authorization.ts b/apps/sim/lib/projects/application/authorization.ts index 75fe2291a15..f91e22c75a5 100644 --- a/apps/sim/lib/projects/application/authorization.ts +++ b/apps/sim/lib/projects/application/authorization.ts @@ -192,16 +192,17 @@ function projectVisibility( * Loads and holds canonical Project membership for a resource-specific access policy. `lock` is * the mode the Project is held in: `shared` for a caller that only needs it fenced against * lifecycle and payer changes, `exclusive` for one that will take it exclusively later in the - * transaction, which must never upgrade a shared hold. + * transaction, which must never upgrade a shared hold. `snapshot` holds nothing and relies on + * the caller's read-only snapshot, for a caller that acts on nothing in the transaction. */ export async function loadProjectAccess( tx: DbTransaction, userId: string, input: ProjectAuthorizationInput & { projectId: string }, - lock: 'shared' | 'exclusive' = 'exclusive' + lock: 'shared' | 'exclusive' | 'snapshot' = 'exclusive' ) { if (lock === 'shared') await lockProjectShared(tx, input.projectId) - else await lockProject(tx, input.projectId) + else if (lock === 'exclusive') await lockProject(tx, input.projectId) const [record] = await tx.select().from(project).where(eq(project.id, input.projectId)).limit(1) if ( !record || @@ -210,7 +211,7 @@ export async function loadProjectAccess( throw new OrchestrationError('not_found', 'Project not found') return projectVisibility( record, - await loadProjectMemberships(tx, userId, [record], 'hold'), + await loadProjectMemberships(tx, userId, [record], lock === 'snapshot' ? 'snapshot' : 'hold'), input ) } diff --git a/apps/sim/lib/projects/files/application/authorization.ts b/apps/sim/lib/projects/files/application/authorization.ts index 9d799fa7849..3c2986c6554 100644 --- a/apps/sim/lib/projects/files/application/authorization.ts +++ b/apps/sim/lib/projects/files/application/authorization.ts @@ -98,7 +98,10 @@ function requirePrincipal( } } -/** Rechecks canonical conversation ownership and authoring capability while authorization locks are held. */ +/** + * Rechecks canonical conversation ownership and authoring capability while authorization locks + * are held; `snapshot` reads the same rows unlocked inside the caller's read-only snapshot. + */ export async function requireCurrentCopilotProjectInvocation( tx: DbTransaction, principal: Pick< @@ -106,14 +109,15 @@ export async function requireCurrentCopilotProjectInvocation( 'subjectUserId' | 'invocation' >, access: Awaited>, - scope: Awaited> + scope: Awaited>, + mode: 'hold' | 'snapshot' = 'hold' ) { if (scope.organizationId !== access.record.organizationId) { throw new OrchestrationError('not_found', 'Project not found in this conversation') } let workspaceId: string | null if (principal.invocation.kind === 'chat') { - const [chat] = await tx + const chatQuery = tx .select({ userId: copilotChats.userId, workspaceId: copilotChats.workspaceId, @@ -123,8 +127,8 @@ export async function requireCurrentCopilotProjectInvocation( }) .from(copilotChats) .where(and(eq(copilotChats.id, principal.invocation.chatId), isNull(copilotChats.deletedAt))) - .for('share') .limit(1) + const [chat] = await (mode === 'hold' ? chatQuery.for('share') : chatQuery) if ( !chat || chat.userId !== principal.subjectUserId || @@ -158,7 +162,7 @@ export async function requireCurrentCopilotProjectInvocation( if (!workspaceId || workspaceId !== scope.workspaceId) { throw new OrchestrationError('not_found', 'Project not found in this conversation') } - const [origin] = await tx + const originQuery = tx .select({ workspaceId: workspace.id, workspaceOrganizationId: workspace.organizationId, @@ -166,22 +170,24 @@ export async function requireCurrentCopilotProjectInvocation( }) .from(workspace) .where(and(eq(workspace.id, workspaceId), isNull(workspace.archivedAt))) - .for('share') .limit(1) + const [origin] = await (mode === 'hold' ? originQuery.for('share') : originQuery) if (!origin || origin.workspaceOrganizationId !== scope.organizationId) throw new OrchestrationError('not_found', 'Workspace not found in this conversation') - // Hold origin grants without upgrading the target's existing SHARE locks. - await tx - .select({ id: permissions.id }) - .from(permissions) - .where( - and( - eq(permissions.userId, principal.subjectUserId), - eq(permissions.entityType, 'workspace'), - eq(permissions.entityId, workspaceId) + if (mode === 'hold') { + // Hold origin grants without upgrading the target's existing SHARE locks. + await tx + .select({ id: permissions.id }) + .from(permissions) + .where( + and( + eq(permissions.userId, principal.subjectUserId), + eq(permissions.entityType, 'workspace'), + eq(permissions.entityId, workspaceId) + ) ) - ) - .for('share') + .for('share') + } await requireCurrentHumanRole(principal.subjectUserId, origin, 'read', { executor: tx }) await assertWorkspaceCapability( principal.subjectUserId, @@ -197,10 +203,11 @@ export async function requireProjectFileOwnerRole( tx: DbTransaction, principal: ProjectFilePrincipal | SubjectDelegatedPrincipal | OrganizationDelegatedPrincipal, access: Awaited>, - accessMode: 'read' | 'write' + accessMode: 'read' | 'write', + mode: 'hold' | 'snapshot' = 'hold' ): Promise { if (access.record.archivedAt) throw new OrchestrationError('conflict', 'Project is archived') - await requireCurrentFileSubject(tx, principal) + await requireCurrentFileSubject(tx, principal, mode) const canWrite = access.orgAdmin || access.active.some((row) => row.permission === 'admin') || diff --git a/apps/sim/lib/workspace-files/application/subject.ts b/apps/sim/lib/workspace-files/application/subject.ts index a11bc5906e7..81145e11930 100644 --- a/apps/sim/lib/workspace-files/application/subject.ts +++ b/apps/sim/lib/workspace-files/application/subject.ts @@ -17,19 +17,23 @@ export async function requireFileSubject(principal: FileSubjectPrincipal): Promi return userId } -/** Holds the acting account stable through a file transaction without substituting an owner. */ +/** + * Holds the acting account stable through a file transaction without substituting an owner. + * `snapshot` reads it without a lock, for a caller in a read-only snapshot that acts on nothing. + */ export async function requireCurrentFileSubject( tx: DbTransaction, - principal: FileSubjectPrincipal + principal: FileSubjectPrincipal, + mode: 'hold' | 'snapshot' = 'hold' ): Promise { // actorless-unsupported: Project files and cross-owner copies require a human; their operation registries reject workspace keys and executors. const userId = requirePrincipalSubjectUserId(principal) - const [actor] = await tx + const query = tx .select({ banned: user.banned, banExpires: user.banExpires, suspendedAt: user.suspendedAt }) .from(user) .where(eq(user.id, userId)) - .for('share') .limit(1) + const [actor] = await (mode === 'hold' ? query.for('share') : query) if (!actor || isAccountBlocked(actor)) { throw new OrchestrationError('forbidden', 'User account is suspended') }