diff --git a/sql/021_upload_receipt.sql b/sql/021_upload_receipt.sql new file mode 100644 index 0000000..a9eace9 --- /dev/null +++ b/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; diff --git a/src/document-workbench/upload-attempt-store.ts b/src/document-workbench/upload-attempt-store.ts index 0fa0aea..0756111 100644 --- a/src/document-workbench/upload-attempt-store.ts +++ b/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(input: UploadAttemptIdentity, action: (tx: postgres.TransactionSql) => Promise): Promise { + private async scoped( + input: UploadAttemptIdentity, + action: (tx: postgres.TransactionSql) => Promise, + ): Promise { 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; @@ -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` + const rows = inserted.length + ? inserted + : await tx` 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 { + 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 { + 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, + }; + }); + } } diff --git a/src/document-workbench/upload-coordinator.ts b/src/document-workbench/upload-coordinator.ts index a338247..e3da821 100644 --- a/src/document-workbench/upload-coordinator.ts +++ b/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; type Attempts = { prepare(input: UploadAttemptIdentity): Promise<{ state: UploadAttemptState; replay: boolean }>; startDispatch(input: UploadAttemptIdentity): Promise; + /** Optional so lightweight tests/implementations keep working. */ + recordReceipt?(input: UploadAttemptIdentity, receipt: StoredUploadReceipt): Promise; + getReceipt?(input: UploadAttemptIdentity): Promise; }; type Doc0 = { upload(input: Doc0UploadInput): Promise; @@ -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; - maxBytes: number; - }) { + 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'); } @@ -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 }; diff --git a/tests/document-workbench/upload-coordinator.test.ts b/tests/document-workbench/upload-coordinator.test.ts index 684189d..277033f 100644 --- a/tests/document-workbench/upload-coordinator.test.ts +++ b/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(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 attempts = { + prepare: vi.fn(async () => ({ + state: 'prepared' as const, + replay: false, + })), + startDispatch: vi.fn(async () => true), + recordReceipt: vi.fn(async () => undefined), + getReceipt: vi.fn(async () => undefined), + }; + 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 }; @@ -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');