From ce41d601f79bf7f2c804865df35debe062a313c3 Mon Sep 17 00:00:00 2001 From: sax3l Date: Mon, 2 Nov 2026 07:17:32 +0100 Subject: [PATCH] feat: coordinate journaled document upload attempts --- src/document-workbench/upload-coordinator.ts | 79 +++++++++++++++++++ .../upload-attempt-store.realinfra.test.ts | 17 +++- .../upload-coordinator.test.ts | 68 ++++++++++++++++ 3 files changed, 163 insertions(+), 1 deletion(-) create mode 100644 src/document-workbench/upload-coordinator.ts create mode 100644 tests/document-workbench/upload-coordinator.test.ts diff --git a/src/document-workbench/upload-coordinator.ts b/src/document-workbench/upload-coordinator.ts new file mode 100644 index 0000000..a338247 --- /dev/null +++ b/src/document-workbench/upload-coordinator.ts @@ -0,0 +1,79 @@ +import { + Doc0UploadOutcomeUnknownError, + prepareDoc0UploadIdentity, + type Doc0Inspection, + type Doc0InspectionInput, + type Doc0UploadInput, + type Doc0UploadReceipt, +} from './doc0-upload-client'; +import type { UploadAttemptIdentity, UploadAttemptState } from './upload-attempt-store'; + +type Scope = Pick; +type Attempts = { + prepare(input: UploadAttemptIdentity): Promise<{ state: UploadAttemptState; replay: boolean }>; + startDispatch(input: UploadAttemptIdentity): Promise; +}; +type Doc0 = { + upload(input: Doc0UploadInput): Promise; + inspect(input: Doc0InspectionInput): Promise; +}; +export type CoordinatedUpload = + | { status: 'receipt_verified'; receipt: Doc0UploadReceipt } + | { status: 'outcome_unknown'; observation?: Doc0Inspection }; + +/** + * Internal only: no customer routes are wired. Scope must be derived from an + * authenticated principal and independently authorized by the injected policy. + * The DOC0 journal, RLS database role, storage and project authorization must + * be deployed and verified before this coordinator can serve customer data. + */ +export class DocumentUploadCoordinator { + constructor(private readonly deps: { + attempts: Attempts; + doc0: Doc0; + authorize: (scope: Scope) => Promise; + maxBytes: number; + }) { + if (!Number.isSafeInteger(deps.maxBytes) || deps.maxBytes <= 0) throw new Error('Invalid upload limit'); + } + + async execute(input: Doc0UploadInput): Promise { + if (!(input.bytes instanceof Uint8Array) || input.bytes.byteLength > this.deps.maxBytes) { + throw new Error('Invalid upload input'); + } + // Snapshot both metadata and bytes before the first await: a caller cannot + // mutate the request between the durable fingerprint and the actual POST. + const snapshot: Doc0UploadInput = { ...input, bytes: new Uint8Array(input.bytes) }; + const details = prepareDoc0UploadIdentity(snapshot); + const scope = { tenantId: snapshot.tenantId, projectId: snapshot.projectId, actorId: snapshot.actorId }; + if ((await this.deps.authorize(scope)) !== true) throw new Error('Document upload not authorized'); + const attempt: UploadAttemptIdentity = { + ...scope, attemptId: snapshot.idempotencyKey, sourceSha256: details.sourceSha256, + sourceSizeBytes: details.sourceSizeBytes, requestFingerprint: details.requestFingerprint, + }; + await this.deps.attempts.prepare(attempt); + // A database error here also stops the operation: the update might have + // committed and must never be followed by an unrecorded network write. + if ((await this.deps.attempts.startDispatch(attempt)) === true) { + try { + return { status: 'receipt_verified', receipt: await this.deps.doc0.upload(snapshot) }; + } catch (error) { + // The client marks every potentially committed POST as unknown. Any + // other error propagates, while the one-way dispatch remains recorded. + if (!(error instanceof Doc0UploadOutcomeUnknownError)) throw error; + } + } + // Observation only. Even not_found does not prove that a write cannot + // still arrive; completed_recorded does not prove storage or extraction. + try { + const observation = await this.deps.doc0.inspect({ + ...scope, requestId: snapshot.requestId, idempotencyKey: snapshot.idempotencyKey, + sourceSha256: details.sourceSha256, sourceSizeBytes: details.sourceSizeBytes, + requestFingerprint: details.requestFingerprint, + }); + return { status: 'outcome_unknown', observation }; + } catch { + return { status: 'outcome_unknown' }; + } + } +} diff --git a/tests/document-workbench/upload-attempt-store.realinfra.test.ts b/tests/document-workbench/upload-attempt-store.realinfra.test.ts index a17b61c..47ccb65 100644 --- a/tests/document-workbench/upload-attempt-store.realinfra.test.ts +++ b/tests/document-workbench/upload-attempt-store.realinfra.test.ts @@ -1,6 +1,7 @@ import { readFile } from 'node:fs/promises'; import postgres from 'postgres'; -import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; +import { DocumentUploadCoordinator } from '../../src/document-workbench/upload-coordinator'; import { PostgresUploadAttemptStore, type UploadAttemptIdentity } from '../../src/document-workbench/upload-attempt-store'; const url = process.env.OPT_ATTEMPT_TEST_DATABASE_URL; @@ -57,4 +58,18 @@ describe.skipIf(!url)('upload attempt manifest with real PostgreSQL RLS', () => try { expect(await new PostgresUploadAttemptStore(second).prepare(attempt)).toEqual({ state: 'dispatch_started', replay: true }); } finally { await second.end(); } }); + it('coordinates concurrent uploads through the real RLS store without resending', async () => { + const doc0 = { upload: vi.fn(async () => ({ documentId: 'doc', versionId: 'v', checksumSha256: 'fixture', pipelineStage: 'accepted' })), + inspect: vi.fn(async () => ({ status: 'not_found' as const })) }; + const coordinator = new DocumentUploadCoordinator({ attempts: store, doc0, authorize: async () => true, maxBytes: 100 }); + const input = { tenantId: 'tenant-a', projectId: 'project-a', actorId: 'actor-a', requestId: 'request', idempotencyKey: 'coordinator-race', + filename: 'fil åäö.txt', mimeType: 'text/plain', classification: 'confidential' as const, bytes: new TextEncoder().encode('fixture') }; + const results = await Promise.all(Array.from({ length: 8 }, () => coordinator.execute(input))); + expect(results.filter((result) => result.status === 'receipt_verified')).toHaveLength(1); + expect(results.filter((result) => result.status === 'outcome_unknown')).toHaveLength(7); + expect(doc0.upload).toHaveBeenCalledOnce(); + await expect(coordinator.execute({ ...input, bytes: new TextEncoder().encode('different') })).rejects.toThrow(/cannot be reused/); + expect(doc0.upload).toHaveBeenCalledOnce(); + }); + }); diff --git a/tests/document-workbench/upload-coordinator.test.ts b/tests/document-workbench/upload-coordinator.test.ts new file mode 100644 index 0000000..684189d --- /dev/null +++ b/tests/document-workbench/upload-coordinator.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it, vi } from 'vitest'; +import { Doc0UploadOutcomeUnknownError, prepareDoc0UploadIdentity, type Doc0UploadClient } from '../../src/document-workbench/doc0-upload-client'; +import { DocumentUploadCoordinator } from '../../src/document-workbench/upload-coordinator'; +import type { PostgresUploadAttemptStore } from '../../src/document-workbench/upload-attempt-store'; + +const input = { tenantId: 't1', projectId: 'p1', actorId: 'a1', requestId: 'r1', idempotencyKey: 'i1', + filename: 'rapport åäö.txt', mimeType: 'text/plain', classification: 'confidential' as const, + bytes: new TextEncoder().encode('original') }; +const receipt = { documentId: 'd1', versionId: 'v1', checksumSha256: 'test', pipelineStage: 'accepted' }; +function fixture() { + const attempts = { prepare: vi.fn(async () => ({ state: 'prepared' as const, replay: false })), startDispatch: vi.fn(async () => true) }; + const doc0 = { upload: vi.fn(async () => receipt), inspect: vi.fn(async () => ({ status: 'not_found' as const })) }; + const authorize = vi.fn(async () => true); + const coordinator = new DocumentUploadCoordinator({ attempts, doc0, authorize, maxBytes: 100 }); + return { attempts, doc0, authorize, coordinator }; +} +describe('document upload coordination', () => { + it('authorizes and persists the exact fingerprint before a single write', async () => { + const f = fixture(); + expect(await f.coordinator.execute(input)).toEqual({ status: 'receipt_verified', receipt }); + expect(f.authorize).toHaveBeenCalledWith({ tenantId: 't1', projectId: 'p1', actorId: 'a1' }); + const attempt = f.attempts.prepare.mock.calls[0][0]; + expect(attempt).toEqual({ tenantId: 't1', projectId: 'p1', actorId: 'a1', attemptId: 'i1', + sourceSha256: prepareDoc0UploadIdentity(input).sourceSha256, sourceSizeBytes: input.bytes.length, + requestFingerprint: prepareDoc0UploadIdentity(input).requestFingerprint }); + expect(f.attempts.startDispatch).toHaveBeenCalledWith(attempt); + expect(f.attempts.prepare.mock.invocationCallOrder[0]).toBeLessThan(f.attempts.startDispatch.mock.invocationCallOrder[0]); + expect(f.attempts.startDispatch.mock.invocationCallOrder[0]).toBeLessThan(f.doc0.upload.mock.invocationCallOrder[0]); + }); + it('denies before persistence or network when authorization fails', async () => { + const f = fixture(); f.authorize.mockResolvedValueOnce(false); + await expect(f.coordinator.execute(input)).rejects.toThrow(/not authorized/); + expect(f.attempts.prepare).not.toHaveBeenCalled(); expect(f.doc0.upload).not.toHaveBeenCalled(); + }); + it('rejects truthy but non-boolean authorization results', async () => { + const f = fixture(); f.authorize.mockResolvedValueOnce('allow' as unknown as boolean); + await expect(f.coordinator.execute(input)).rejects.toThrow(/not authorized/); + expect(f.attempts.prepare).not.toHaveBeenCalled(); + }); + it('does not dispatch after a failed or ambiguous database transition', async () => { + const f = fixture(); f.attempts.startDispatch.mockRejectedValueOnce(new Error('connection lost')); + await expect(f.coordinator.execute(input)).rejects.toThrow(/connection lost/); + expect(f.doc0.upload).not.toHaveBeenCalled(); expect(f.doc0.inspect).not.toHaveBeenCalled(); + }); + it('only observes a previously started attempt and never sends a second POST', async () => { + const f = fixture(); f.attempts.startDispatch.mockResolvedValueOnce(false); + expect(await f.coordinator.execute(input)).toEqual({ status: 'outcome_unknown', observation: { status: 'not_found' } }); + expect(f.doc0.upload).not.toHaveBeenCalled(); expect(f.doc0.inspect).toHaveBeenCalledOnce(); + }); + it('keeps a journaled completion distinct from a verified receipt', async () => { + const f = fixture(); f.attempts.startDispatch.mockResolvedValueOnce(false); + f.doc0.inspect.mockResolvedValueOnce({ status: 'completed_recorded', documentId: 'doc', versionId: 'version' }); + expect(await f.coordinator.execute(input)).toEqual({ status: 'outcome_unknown', observation: { status: 'completed_recorded', documentId: 'doc', versionId: 'version' } }); + expect(f.doc0.upload).not.toHaveBeenCalled(); + }); + it('keeps a possibly committed upload unknown even when inspection is unavailable', async () => { + const f = fixture(); f.doc0.upload.mockRejectedValueOnce(new Doc0UploadOutcomeUnknownError()); f.doc0.inspect.mockRejectedValueOnce(new Error('unavailable')); + expect(await f.coordinator.execute(input)).toEqual({ status: 'outcome_unknown' }); + expect(f.doc0.upload).toHaveBeenCalledOnce(); expect(f.doc0.inspect).toHaveBeenCalledOnce(); + }); + it('snapshots bytes and metadata before asynchronous authorization', async () => { + const f = fixture(); const mutable = { ...input, bytes: new Uint8Array(input.bytes) }; + f.authorize.mockImplementationOnce(async () => { mutable.bytes.fill(0); mutable.filename = 'changed.txt'; return true; }); + await f.coordinator.execute(mutable); + expect(f.doc0.upload.mock.calls[0][0].filename).toBe(input.filename); + expect(new TextDecoder().decode(f.doc0.upload.mock.calls[0][0].bytes)).toBe('original'); + }); +}); -- 2.54.0