0pt/src/document-workbench/upload-coordinator.ts

import {
  Doc0UploadOutcomeUnknownError,
  prepareDoc0UploadIdentity,
  type Doc0Inspection,
  type Doc0InspectionInput,
  type Doc0UploadInput,
  type Doc0UploadReceipt,
} from './doc0-upload-client';
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>;
  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 {
        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,
        requestFingerprint: details.requestFingerprint,
      });
      return { status: 'outcome_unknown', observation };
    } catch {
      return { status: 'outcome_unknown' };
    }
  }
}