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<Doc0UploadInput, 'tenantId' | 'projectId' | 'actorId'>;
+type Attempts = {
+  prepare(input: UploadAttemptIdentity): Promise<{ state: UploadAttemptState; replay: boolean }>;
+  startDispatch(input: UploadAttemptIdentity): Promise<boolean>;
+};
+type Doc0 = {
+  upload(input: Doc0UploadInput): Promise<Doc0UploadReceipt>;
+  inspect(input: Doc0InspectionInput): Promise<Doc0Inspection>;
+};
+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<boolean>;
+    maxBytes: number;
+  }) {
+    if (!Number.isSafeInteger(deps.maxBytes) || deps.maxBytes <= 0) throw new Error('Invalid upload limit');
+  }
+
+  async execute(input: Doc0UploadInput): Promise<CoordinatedUpload> {
+    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' };
+    }
+  }
+}

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();
+  });
+
 });

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<PostgresUploadAttemptStore['prepare']>(async () => ({ state: 'prepared' as const, replay: false })), startDispatch: vi.fn<PostgresUploadAttemptStore['startDispatch']>(async () => true) };
+  const doc0 = { upload: vi.fn<Doc0UploadClient['upload']>(async () => receipt), inspect: vi.fn<Doc0UploadClient['inspect']>(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');
+  });
+});