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 => ({ 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 => { 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 => { 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(); }); });