sql/018_document_upload_attempts.sql

@@ -0,0 +1,23 @@
+-- Internal coordination only: raw bytes remain in the document storage service.
+CREATE TABLE opt_document_upload_attempts (
+  tenant_id text NOT NULL CHECK (length(tenant_id) BETWEEN 1 AND 256),
+  project_id text NOT NULL CHECK (length(project_id) BETWEEN 1 AND 256),
+  actor_id text NOT NULL CHECK (length(actor_id) BETWEEN 1 AND 256),
+  attempt_id text NOT NULL CHECK (length(attempt_id) BETWEEN 1 AND 256),
+  source_sha256 text NOT NULL CHECK (source_sha256 ~ '^[0-9a-f]{64}$'),
+  source_size_bytes bigint NOT NULL CHECK (source_size_bytes > 0 AND source_size_bytes <= 9007199254740991),
+  request_fingerprint text NOT NULL CHECK (request_fingerprint ~ '^[0-9a-f]{64}$'),
+  state text NOT NULL CHECK (state IN ('prepared', 'dispatch_started')),
+  created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+  updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+  PRIMARY KEY (tenant_id, attempt_id)
+);
+ALTER TABLE opt_document_upload_attempts ENABLE ROW LEVEL SECURITY;
+ALTER TABLE opt_document_upload_attempts FORCE ROW LEVEL SECURITY;
+CREATE POLICY opt_document_upload_attempt_scope ON opt_document_upload_attempts
+  USING (tenant_id = current_setting('app.opt_tenant_id', true)
+    AND project_id = current_setting('app.opt_project_id', true)
+    AND actor_id = current_setting('app.opt_actor_id', true))
+  WITH CHECK (tenant_id = current_setting('app.opt_tenant_id', true)
+    AND project_id = current_setting('app.opt_project_id', true)
+    AND actor_id = current_setting('app.opt_actor_id', true));

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

@@ -0,0 +1,83 @@
+import type { RepositorySql } from '../repositories/db';
+import type postgres from 'postgres';
+
+/** All scope values must come from authenticated and project-authorized context. */
+export interface UploadAttemptIdentity {
+  tenantId: string;
+  projectId: string;
+  actorId: string;
+  /** Also used as the original DOC0 idempotency key by the future caller. */
+  attemptId: string;
+  sourceSha256: string;
+  sourceSizeBytes: number;
+  requestFingerprint: string;
+}
+export type UploadAttemptState = 'prepared' | 'dispatch_started';
+interface Row {
+  source_sha256: string;
+  source_size_bytes: string;
+  request_fingerprint: string;
+  state: UploadAttemptState;
+}
+
+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) {
+    throw new Error('Invalid document upload attempt');
+  }
+}
+
+/**
+ * Internal, one-way coordination primitive; not an authentication, project
+ * authorization or content-integrity check. Callers must persist the exact
+ * DOC0-compatible fingerprint and verify actual uploaded bytes themselves.
+ * RLS session settings are set only inside each transaction; the database
+ * role must not be superuser or BYPASSRLS. A started dispatch is never
+ * automatically retried or declared absent. No customer route is wired.
+ */
+export class PostgresUploadAttemptStore {
+  constructor(private readonly sql: RepositorySql) {}
+
+  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');
+      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>;
+  }
+
+  async prepare(input: UploadAttemptIdentity): Promise<{ state: UploadAttemptState; replay: boolean }> {
+    return this.scoped(input, async (tx) => {
+      const inserted = await tx<Row[]>`
+        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[]>`
+        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) {
+        throw new Error('Document upload attempt cannot be reused');
+      }
+      return { state: row.state, replay: inserted.length === 0 };
+    });
+  }
+
+  /** Exactly one claimant may start dispatch. Persist this before any network write. */
+  async startDispatch(input: UploadAttemptIdentity): Promise<boolean> {
+    return this.scoped(input, async (tx) => {
+      const rows = await tx<{ attempt_id: string }[]>`
+        UPDATE opt_document_upload_attempts SET state='dispatch_started', 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}
+          AND state='prepared' RETURNING attempt_id`;
+      return rows.length === 1;
+    });
+  }
+}

tests/document-workbench/upload-attempt-store.realinfra.test.ts

@@ -0,0 +1,60 @@
+import { readFile } from 'node:fs/promises';
+import postgres from 'postgres';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { PostgresUploadAttemptStore, type UploadAttemptIdentity } from '../../src/document-workbench/upload-attempt-store';
+
+const url = process.env.OPT_ATTEMPT_TEST_DATABASE_URL;
+describe.skipIf(!url)('upload attempt manifest with real PostgreSQL RLS', () => {
+  let admin: ReturnType<typeof postgres>;
+  let app: ReturnType<typeof postgres>;
+  let store: PostgresUploadAttemptStore;
+  let scopedUrl: string;
+  const sample: UploadAttemptIdentity = {
+    tenantId: 'tenant-a', projectId: 'project-a', actorId: 'actor-a', attemptId: 'attempt-a',
+    sourceSha256: 'a'.repeat(64), sourceSizeBytes: 12, requestFingerprint: 'b'.repeat(64),
+  };
+
+  beforeAll(async () => {
+    if (process.env.OPT_ATTEMPT_TEST_ALLOW_LOCAL !== '1' || new URL(url!).hostname !== '127.0.0.1') {
+      throw new Error('Only an explicitly enabled isolated loopback test database may be used');
+    }
+    admin = postgres(url!, { max: 1, prepare: false });
+    await admin.unsafe(await readFile('sql/018_document_upload_attempts.sql', 'utf8'));
+    await admin.unsafe('CREATE ROLE opt_attempt_test LOGIN NOSUPERUSER NOBYPASSRLS');
+    await admin.unsafe('GRANT USAGE ON SCHEMA public TO opt_attempt_test');
+    await admin.unsafe('GRANT SELECT, INSERT, UPDATE ON opt_document_upload_attempts TO opt_attempt_test');
+    const appUrl = new URL(url!);
+    appUrl.username = 'opt_attempt_test';
+    scopedUrl = appUrl.toString();
+    app = postgres(scopedUrl, { max: 8, prepare: false });
+    store = new PostgresUploadAttemptStore(app);
+  });
+  afterAll(async () => { await app?.end(); await admin?.end(); });
+
+  it('rejects a database role which bypasses RLS', async () => {
+    await expect(new PostgresUploadAttemptStore(admin).prepare(sample)).rejects.toThrow(/RLS-enforced/);
+  });
+  it('persists matching retries and rejects changed or differently scoped identities', async () => {
+    expect(await store.prepare(sample)).toEqual({ state: 'prepared', replay: false });
+    expect(await store.prepare(sample)).toEqual({ state: 'prepared', replay: true });
+    await expect(store.prepare({ ...sample, sourceSha256: 'c'.repeat(64) })).rejects.toThrow();
+    await expect(store.prepare({ ...sample, projectId: 'project-b' })).rejects.toThrow('Document upload attempt cannot be reused');
+    await expect(store.prepare({ ...sample, actorId: 'actor-b' })).rejects.toThrow('Document upload attempt cannot be reused');
+    expect(await store.prepare({ ...sample, tenantId: 'tenant-b' })).toEqual({ state: 'prepared', replay: false });
+    expect(await app`SELECT * FROM opt_document_upload_attempts`).toHaveLength(0);
+  });
+  it('allows only one concurrent dispatch claimant and survives a new connection', async () => {
+    const attempt = { ...sample, attemptId: 'race' };
+    expect(await store.startDispatch(attempt)).toBe(false);
+    await store.prepare(attempt);
+    expect(await store.startDispatch({ ...attempt, actorId: 'actor-b' })).toBe(false);
+    expect(await store.startDispatch({ ...attempt, sourceSha256: 'c'.repeat(64) })).toBe(false);
+    const results = await Promise.all(Array.from({ length: 8 }, () => store.startDispatch(attempt)));
+    expect(results.filter(Boolean)).toHaveLength(1);
+    expect(await store.startDispatch(attempt)).toBe(false);
+    expect(await store.prepare(attempt)).toEqual({ state: 'dispatch_started', replay: true });
+    const second = postgres(scopedUrl, { max: 1, prepare: false });
+    try { expect(await new PostgresUploadAttemptStore(second).prepare(attempt)).toEqual({ state: 'dispatch_started', replay: true }); }
+    finally { await second.end(); }
+  });
+});