stack/apps/gateway/src/hierarchy/hierarchy-audit.integration.test.ts

import { mkdtemp, rm } from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { Test, type TestingModule } from '@nestjs/testing';
import {
  companies,
  createPgliteDb,
  eq,
  estates,
  hierarchyAuditEvents,
  hierarchyOutbox,
  platformProjects,
  runPgliteMigrations,
  type DbHandle,
} from '@mosaicstack/db';
import { DB } from '../database/database.module.js';
import {
  HierarchyAuditIdempotencyConflictError,
  HierarchyAuditRepository,
  HierarchyNodeNotFoundError,
  type AppendHierarchyEventInput,
} from './hierarchy-audit.repository.js';

/**
 * Repository-level §6.4 witnesses for the hierarchy audit machinery
 * (contract 1 §5.2, REQ-AUD-001): same-transaction atomicity of state +
 * event + outbox, rollback leaving no residue, idempotent replay, snapshot
 * parent chains, events surviving target deletion, per-target ordering, and
 * the outbox claim/complete/release CAS. The schema-level constraints are
 * witnessed in packages/db/src/hierarchy-audit.witness.test.ts.
 */
describe('hierarchy audit repository integration', (): void => {
  let dataDir: string;
  let handle: DbHandle;
  let moduleRef: TestingModule;
  let repo: HierarchyAuditRepository;

  const input = (
    overrides: Partial<AppendHierarchyEventInput> = {},
  ): AppendHierarchyEventInput => ({
    actorId: 'user-actor',
    verb: 'create',
    targetKind: 'company',
    targetId: randomUUID(),
    targetSnapshot: { id: 'x', slug: 'x', name: 'x', parentChain: [] },
    correlationId: 'corr-1',
    idempotencyKey: `key-${randomUUID()}`,
    ...overrides,
  });

  beforeAll(async (): Promise<void> => {
    dataDir = await mkdtemp(join(tmpdir(), 'mosaic-gateway-hierarchy-audit-'));
    handle = createPgliteDb(dataDir);
    await runPgliteMigrations(handle);
    moduleRef = await Test.createTestingModule({
      providers: [HierarchyAuditRepository, { provide: DB, useValue: handle.db }],
    }).compile();
    repo = moduleRef.get(HierarchyAuditRepository);
  });

  afterAll(async (): Promise<void> => {
    await moduleRef.close();
    await handle.close();
    await rm(dataDir, { recursive: true, force: true });
  });

  it('commits state, event, and outbox record atomically in one transaction', async () => {
    const companyId = randomUUID();
    const key = `key-${randomUUID()}`;
    await handle.db.transaction(async (tx) => {
      await tx.insert(companies).values({ id: companyId, name: 'Atomic Co', slug: 'atomic-co' });
      const snapshot = await repo.snapshot(tx, 'company', companyId);
      const result = await repo.append(tx, {
        ...input({ targetId: companyId, idempotencyKey: key }),
        targetSnapshot: { ...snapshot },
      });
      expect(result.replayed).toBe(false);
      expect(result.event.idempotencyKey).toBe(key);
    });
    const events = await handle.db
      .select()
      .from(hierarchyAuditEvents)
      .where(eq(hierarchyAuditEvents.idempotencyKey, key));
    expect(events).toHaveLength(1);
    const outbox = await handle.db
      .select()
      .from(hierarchyOutbox)
      .where(eq(hierarchyOutbox.eventId, events[0]!.id));
    expect(outbox).toHaveLength(1);
    expect(outbox[0]).toMatchObject({
      status: 'pending',
      idempotencyKey: key,
      correlationId: 'corr-1',
    });
  });

  it('a rolled-back transaction leaves no state, no event, and no outbox record', async () => {
    const companyId = randomUUID();
    const key = `key-${randomUUID()}`;
    await expect(
      handle.db.transaction(async (tx) => {
        await tx.insert(companies).values({ id: companyId, name: 'Doomed Co', slug: 'doomed-co' });
        await repo.append(tx, input({ targetId: companyId, idempotencyKey: key }));
        throw new Error('deliberate rollback');
      }),
    ).rejects.toThrow('deliberate rollback');
    const [companyRows, eventRows, outboxRows] = await Promise.all([
      handle.db.select().from(companies).where(eq(companies.id, companyId)),
      handle.db
        .select()
        .from(hierarchyAuditEvents)
        .where(eq(hierarchyAuditEvents.idempotencyKey, key)),
      handle.db.select().from(hierarchyOutbox).where(eq(hierarchyOutbox.idempotencyKey, key)),
    ]);
    expect(companyRows).toHaveLength(0);
    expect(eventRows).toHaveLength(0);
    expect(outboxRows).toHaveLength(0);
  });

  it('replays a duplicate idempotency key without inserting a second event or outbox record', async () => {
    const first = input();
    const original = await handle.db.transaction(async (tx) => repo.append(tx, first));
    const replay = await handle.db.transaction(async (tx) => repo.append(tx, first));
    expect(original.replayed).toBe(false);
    expect(replay.replayed).toBe(true);
    expect(replay.event.id).toBe(original.event.id);
    const outbox = await handle.db
      .select()
      .from(hierarchyOutbox)
      .where(eq(hierarchyOutbox.eventId, original.event.id));
    expect(outbox).toHaveLength(1);
  });

  it('throws on a duplicate idempotency key carrying different event content', async () => {
    const first = input();
    await handle.db.transaction(async (tx) => repo.append(tx, first));
    await expect(
      handle.db.transaction(async (tx) =>
        repo.append(tx, { ...first, verb: 'rename', targetId: randomUUID() }),
      ),
    ).rejects.toThrow(HierarchyAuditIdempotencyConflictError);
  });

  it('throws on a duplicate idempotency key whose transfer destination differs', async () => {
    const from = { kind: 'company' as const, id: randomUUID(), slug: 'src-co' };
    const to = { kind: 'company' as const, id: randomUUID(), slug: 'dst-co' };
    const first = input({
      verb: 'transfer',
      targetKind: 'estate',
      transferFrom: from,
      transferTo: to,
    });
    const original = await handle.db.transaction(async (tx) => repo.append(tx, first));
    expect(original.replayed).toBe(false);
    // Identical retry replays; a retry re-routed to a different destination must conflict.
    const replay = await handle.db.transaction(async (tx) => repo.append(tx, first));
    expect(replay.replayed).toBe(true);
    await expect(
      handle.db.transaction(async (tx) =>
        repo.append(tx, { ...first, transferTo: { ...to, id: randomUUID() } }),
      ),
    ).rejects.toThrow(HierarchyAuditIdempotencyConflictError);
  });

  it('builds root-first parent chains and rejects unknown nodes', async () => {
    const companyId = randomUUID();
    const estateId = randomUUID();
    const projectId = randomUUID();
    await handle.db.transaction(async (tx) => {
      await tx.insert(companies).values({ id: companyId, name: 'Chain Co', slug: 'chain-co' });
      await tx
        .insert(estates)
        .values({ id: estateId, name: 'Chain Estate', slug: 'chain-estate', companyId });
      await tx
        .insert(platformProjects)
        .values({ id: projectId, name: 'Chain Project', slug: 'chain-project', estateId });
    });
    const snapshot = await repo.snapshot(handle.db, 'platform_project', projectId);
    expect(snapshot).toMatchObject({ id: projectId, slug: 'chain-project', name: 'Chain Project' });
    expect(snapshot.parentChain).toEqual([
      { kind: 'company', id: companyId, slug: 'chain-co' },
      { kind: 'estate', id: estateId, slug: 'chain-estate' },
    ]);
    await expect(repo.snapshot(handle.db, 'estate', randomUUID())).rejects.toThrow(
      HierarchyNodeNotFoundError,
    );
  });

  it('keeps events readable, in per-target seq order, after the target row is deleted', async () => {
    const companyId = randomUUID();
    await handle.db.transaction(async (tx) => {
      await tx.insert(companies).values({ id: companyId, name: 'Mortal Co', slug: 'mortal-co' });
      const snapshot = await repo.snapshot(tx, 'company', companyId);
      await repo.append(tx, input({ targetId: companyId, targetSnapshot: { ...snapshot } }));
    });
    await handle.db.transaction(async (tx) => {
      const snapshot = await repo.snapshot(tx, 'company', companyId);
      await repo.append(tx, {
        ...input({ verb: 'delete', targetId: companyId }),
        targetSnapshot: { ...snapshot },
      });
      await tx.delete(companies).where(eq(companies.id, companyId));
    });
    const events = await repo.eventsForTarget(companyId);
    expect(events.map((e) => e.verb)).toEqual(['create', 'delete']);
    expect(events[1]!.seq).toBeGreaterThan(events[0]!.seq);
    expect((events[1]!.targetSnapshot as { id: string }).id).toBe(companyId);
  });

  it('claims the oldest pending outbox record exactly once, completes and releases by CAS', async () => {
    // Drain records left pending by earlier cases so ordering is deterministic.
    for (;;) {
      const drained = await repo.claimPendingOutbox();
      if (!drained) break;
      await repo.completeOutbox(drained.id);
    }
    const older = await handle.db.transaction(async (tx) => repo.append(tx, input()));
    const newer = await handle.db.transaction(async (tx) => repo.append(tx, input()));

    const claimed = await repo.claimPendingOutbox();
    expect(claimed).not.toBeNull();
    expect(claimed!.eventId).toBe(older.event.id);
    expect(claimed!.status).toBe('processing');

    // Delivery fails: release returns it to pending and it is claimable again.
    await repo.releaseOutbox(claimed!.id);
    const reclaimed = await repo.claimPendingOutbox();
    expect(reclaimed!.id).toBe(claimed!.id);

    await repo.completeOutbox(reclaimed!.id);
    const done = await handle.db
      .select()
      .from(hierarchyOutbox)
      .where(eq(hierarchyOutbox.id, reclaimed!.id));
    expect(done[0]!.status).toBe('delivered');
    expect(done[0]!.deliveredAt).not.toBeNull();
    // completeOutbox is CAS-guarded on 'processing': completing again is a no-op.
    await repo.completeOutbox(reclaimed!.id);

    const second = await repo.claimPendingOutbox();
    expect(second!.eventId).toBe(newer.event.id);
    await repo.completeOutbox(second!.id);
    expect(await repo.claimPendingOutbox()).toBeNull();
  });
});