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

import { readFile } from 'node:fs/promises';
import postgres from 'postgres';
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;
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(); }
  });
  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();
  });

});