sql/021_upload_receipt.sql

@@ -0,0 +1,8 @@
+-- Persist the verified DOC0 receipt on the durable upload attempt so an
+-- idempotent retry returns the same 201 instead of "outcome unknown".
+-- Additive and re-runnable.
+ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS document_id text;
+ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS version_id text;
+ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS receipt_checksum text;
+ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS pipeline_stage text;
+ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS verified_at timestamptz;

src/document-workbench/upload-attempt-store.ts

@@ -13,6 +13,12 @@ export interface UploadAttemptIdentity {
   requestFingerprint: string;
 }
 export type UploadAttemptState = 'prepared' | 'dispatch_started';
+export interface StoredUploadReceipt {
+  documentId: string;
+  versionId: string;
+  checksumSha256: string;
+  pipelineStage: string;
+}
 interface Row {
   source_sha256: string;
   source_size_bytes: string;
@@ -21,10 +27,15 @@ interface Row {
 }
 
 function validate(input: UploadAttemptIdentity): void {
-  if (![input.tenantId, input.projectId, input.actorId, input.attemptId].every(
-    (value) => typeof value === 'string' && value.trim().length > 0 && value.length <= 256,
-  ) || !/^[0-9a-f]{64}$/.test(input.sourceSha256) || !/^[0-9a-f]{64}$/.test(input.requestFingerprint)
-    || !Number.isSafeInteger(input.sourceSizeBytes) || input.sourceSizeBytes <= 0) {
+  if (
+    ![input.tenantId, input.projectId, input.actorId, input.attemptId].every(
+      (value) => typeof value === 'string' && value.trim().length > 0 && value.length <= 256,
+    ) ||
+    !/^[0-9a-f]{64}$/.test(input.sourceSha256) ||
+    !/^[0-9a-f]{64}$/.test(input.requestFingerprint) ||
+    !Number.isSafeInteger(input.sourceSizeBytes) ||
+    input.sourceSizeBytes <= 0
+  ) {
     throw new Error('Invalid document upload attempt');
   }
 }
@@ -40,12 +51,16 @@ function validate(input: UploadAttemptIdentity): void {
 export class PostgresUploadAttemptStore {
   constructor(private readonly sql: RepositorySql) {}
 
-  private async scoped<T>(input: UploadAttemptIdentity, action: (tx: postgres.TransactionSql) => Promise<T>): Promise<T> {
+  private async scoped<T>(
+    input: UploadAttemptIdentity,
+    action: (tx: postgres.TransactionSql) => Promise<T>,
+  ): Promise<T> {
     validate(input);
     return this.sql.begin(async (tx) => {
       const roles = await tx<{ rolsuper: boolean; rolbypassrls: boolean }[]>`
         SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user`;
-      if (!roles[0] || roles[0].rolsuper || roles[0].rolbypassrls) throw new Error('Upload attempts require an RLS-enforced database role');
+      if (!roles[0] || roles[0].rolsuper || roles[0].rolbypassrls)
+        throw new Error('Upload attempts require an RLS-enforced database role');
       await tx`SELECT set_config('app.opt_tenant_id', ${input.tenantId}, true), set_config('app.opt_project_id', ${input.projectId}, true), set_config('app.opt_actor_id', ${input.actorId}, true)`;
       return action(tx);
     }) as Promise<T>;
@@ -57,11 +72,18 @@ export class PostgresUploadAttemptStore {
         INSERT INTO opt_document_upload_attempts (tenant_id,project_id,actor_id,attempt_id,source_sha256,source_size_bytes,request_fingerprint,state)
         VALUES (${input.tenantId},${input.projectId},${input.actorId},${input.attemptId},${input.sourceSha256},${input.sourceSizeBytes},${input.requestFingerprint},'prepared')
         ON CONFLICT (tenant_id,attempt_id) DO NOTHING RETURNING source_sha256,source_size_bytes,request_fingerprint,state`;
-      const rows = inserted.length ? inserted : await tx<Row[]>`
+      const rows = inserted.length
+        ? inserted
+        : await tx<Row[]>`
         SELECT source_sha256,source_size_bytes,request_fingerprint,state FROM opt_document_upload_attempts
         WHERE tenant_id=${input.tenantId} AND project_id=${input.projectId} AND actor_id=${input.actorId} AND attempt_id=${input.attemptId}`;
       const row = rows[0];
-      if (!row || row.source_sha256 !== input.sourceSha256 || row.source_size_bytes !== String(input.sourceSizeBytes) || row.request_fingerprint !== input.requestFingerprint) {
+      if (
+        !row ||
+        row.source_sha256 !== input.sourceSha256 ||
+        row.source_size_bytes !== String(input.sourceSizeBytes) ||
+        row.request_fingerprint !== input.requestFingerprint
+      ) {
         throw new Error('Document upload attempt cannot be reused');
       }
       return { state: row.state, replay: inserted.length === 0 };
@@ -80,4 +102,59 @@ export class PostgresUploadAttemptStore {
       return rows.length === 1;
     });
   }
+
+  /**
+   * Persist the verified DOC0 receipt so a retry with the same idempotency key
+   * returns the SAME 201 receipt instead of "outcome unknown". Only the fields
+   * DOC0 itself returned are stored; storage/extraction are not re-proven here.
+   */
+  async recordReceipt(input: UploadAttemptIdentity, receipt: StoredUploadReceipt): Promise<void> {
+    if (
+      ![receipt.documentId, receipt.versionId, receipt.pipelineStage].every(
+        (value) => typeof value === 'string' && value.trim().length > 0 && value.length <= 256,
+      ) ||
+      !/^[0-9a-f]{64}$/.test(receipt.checksumSha256)
+    ) {
+      throw new Error('Invalid upload receipt');
+    }
+    await this.scoped(input, async (tx) => {
+      const rows = await tx<{ attempt_id: string }[]>`
+        UPDATE opt_document_upload_attempts
+        SET document_id=${receipt.documentId}, version_id=${receipt.versionId},
+            receipt_checksum=${receipt.checksumSha256}, pipeline_stage=${receipt.pipelineStage},
+            verified_at=clock_timestamp(), updated_at=clock_timestamp()
+        WHERE tenant_id=${input.tenantId} AND project_id=${input.projectId} AND actor_id=${input.actorId}
+          AND attempt_id=${input.attemptId} AND source_sha256=${input.sourceSha256}
+          AND source_size_bytes=${input.sourceSizeBytes} AND request_fingerprint=${input.requestFingerprint}
+        RETURNING attempt_id`;
+      if (rows.length !== 1) throw new Error('Upload attempt receipt could not be recorded');
+    });
+  }
+
+  async getReceipt(input: UploadAttemptIdentity): Promise<StoredUploadReceipt | undefined> {
+    return this.scoped(input, async (tx) => {
+      const rows = await tx<
+        {
+          document_id: string | null;
+          version_id: string | null;
+          receipt_checksum: string | null;
+          pipeline_stage: string | null;
+        }[]
+      >`
+        SELECT document_id, version_id, receipt_checksum, pipeline_stage
+        FROM opt_document_upload_attempts
+        WHERE tenant_id=${input.tenantId} AND project_id=${input.projectId} AND actor_id=${input.actorId}
+          AND attempt_id=${input.attemptId} AND source_sha256=${input.sourceSha256}
+          AND source_size_bytes=${input.sourceSizeBytes} AND request_fingerprint=${input.requestFingerprint}`;
+      const row = rows[0];
+      if (!row?.document_id || !row.version_id || !row.receipt_checksum || !row.pipeline_stage)
+        return undefined;
+      return {
+        documentId: row.document_id,
+        versionId: row.version_id,
+        checksumSha256: row.receipt_checksum,
+        pipelineStage: row.pipeline_stage,
+      };
+    });
+  }
 }

src/document-workbench/upload-coordinator.ts

@@ -6,12 +6,15 @@ import {
   type Doc0UploadInput,
   type Doc0UploadReceipt,
 } from './doc0-upload-client';
-import type { UploadAttemptIdentity, UploadAttemptState } from './upload-attempt-store';
+import type { StoredUploadReceipt, 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>;
+  /** Optional so lightweight tests/implementations keep working. */
+  recordReceipt?(input: UploadAttemptIdentity, receipt: StoredUploadReceipt): Promise<void>;
+  getReceipt?(input: UploadAttemptIdentity): Promise<StoredUploadReceipt | undefined>;
 };
 type Doc0 = {
   upload(input: Doc0UploadInput): Promise<Doc0UploadReceipt>;
@@ -28,12 +31,14 @@ export type CoordinatedUpload =
  * 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;
-  }) {
+  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');
   }
 
@@ -48,27 +53,50 @@ export class DocumentUploadCoordinator {
     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,
+      ...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) };
+        const receipt = await this.deps.doc0.upload(snapshot);
+        // Persist the verified receipt so retrying with the same idempotency
+        // key returns this exact 201 instead of "outcome unknown". A journal
+        // write failure must not turn a real upload into an error response.
+        try {
+          await this.deps.attempts.recordReceipt?.(attempt, receipt);
+        } catch {
+          /* receipt stays unpersisted; a retry falls back to observation */
+        }
+        return { status: 'receipt_verified', receipt };
       } 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;
       }
     }
+    // A retry of an already-dispatched attempt: if the first dispatch returned
+    // a verified receipt, return it again (same document, same checksum).
+    try {
+      const stored = await this.deps.attempts.getReceipt?.(attempt);
+      if (stored) return { status: 'receipt_verified', receipt: stored };
+    } catch {
+      /* fall through to observation */
+    }
     // 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,
+        ...scope,
+        requestId: snapshot.requestId,
+        idempotencyKey: snapshot.idempotencyKey,
+        sourceSha256: details.sourceSha256,
+        sourceSizeBytes: details.sourceSizeBytes,
         requestFingerprint: details.requestFingerprint,
       });
       return { status: 'outcome_unknown', observation };

tests/document-workbench/upload-coordinator.test.ts

@@ -1,15 +1,38 @@
 import { describe, expect, it, vi } from 'vitest';
-import { Doc0UploadOutcomeUnknownError, prepareDoc0UploadIdentity, type Doc0UploadClient } from '../../src/document-workbench/doc0-upload-client';
+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 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 attempts = {
+    prepare: vi.fn<PostgresUploadAttemptStore['prepare']>(async () => ({
+      state: 'prepared' as const,
+      replay: false,
+    })),
+    startDispatch: vi.fn<PostgresUploadAttemptStore['startDispatch']>(async () => true),
+    recordReceipt: vi.fn<PostgresUploadAttemptStore['recordReceipt']>(async () => undefined),
+    getReceipt: vi.fn<PostgresUploadAttemptStore['getReceipt']>(async () => undefined),
+  };
+  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 };
@@ -20,47 +43,99 @@ describe('document upload coordination', () => {
     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(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]);
+    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);
+    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();
+    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);
+    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'));
+    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();
+    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();
+    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('persists the verified receipt after a successful upload', async () => {
+    const f = fixture();
+    expect(await f.coordinator.execute(input)).toEqual({ status: 'receipt_verified', receipt });
+    expect(f.attempts.recordReceipt).toHaveBeenCalledWith(
+      expect.objectContaining({ attemptId: 'i1' }),
+      receipt,
+    );
+  });
+  it('returns the SAME verified receipt on an idempotent retry without a second POST', async () => {
+    const f = fixture();
+    f.attempts.startDispatch.mockResolvedValueOnce(false);
+    f.attempts.getReceipt.mockResolvedValueOnce(receipt);
+    expect(await f.coordinator.execute(input)).toEqual({ status: 'receipt_verified', receipt });
+    expect(f.doc0.upload).not.toHaveBeenCalled();
+    expect(f.doc0.inspect).not.toHaveBeenCalled();
   });
   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' } });
+    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'));
+    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();
+    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; });
+    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');