0pt/tests/p0/dataset-postgres.realinfra.test.ts

/**
 * Real-Postgres proof for the dataset registry + immutable version metadata.
 *
 * Set DATASET_TEST_DATABASE_URL to a disposable Postgres 16 instance. When it
 * is unset the suite skips honestly; Gitea CI should provide it for the
 * blocking real-infrastructure job before this capability is promoted.
 */
import fs from "node:fs";
import path from "node:path";
import postgres from "postgres";
import { afterAll, beforeAll, describe, expect, it } from "vitest";

import { closeRepositorySqlForTests } from "@/src/repositories/db";
import { DatasetService } from "@/src/services/dataset-service";
import { DatasetVersionService } from "@/src/services/dataset-version-service";

const TEST_DATABASE_URL = process.env.DATASET_TEST_DATABASE_URL;
const run = TEST_DATABASE_URL ? describe : describe.skip;
let admin: ReturnType<typeof postgres> | undefined;
let reachable = false;
let unreachableReason = "";

beforeAll(async () => {
  if (!TEST_DATABASE_URL) {
    console.warn("[realinfra] DATASET_TEST_DATABASE_URL unset — skipping honestly.");
    return;
  }
  try {
    admin = postgres(TEST_DATABASE_URL, { prepare: false, ssl: false, max: 3 });
    await Promise.race([
      admin`SELECT 1`,
      new Promise((_, reject) => setTimeout(() => reject(new Error("connect timeout")), 5_000)),
    ]);
    for (const migration of ["003_datasets.sql", "011_aud0_evidence_outbox.sql", "012_dataset_versions.sql"]) {
      await admin.unsafe(fs.readFileSync(path.join(process.cwd(), "sql", migration), "utf8"));
    }
    // Aktivera den durabla AUD0-outboxen: createAud0OutboxRepositoryFromEnv()
    // läser DATABASE_URL (se src/repositories/db.ts) — utan den blir klienten
    // Noop och inga producer-rader skrivs (testet skulle ljugit om durabilitet).
    process.env.DATABASE_URL = TEST_DATABASE_URL;
    reachable = true;
  } catch (error) {
    unreachableReason = error instanceof Error ? error.message : String(error);
  }
}, 30_000);

afterAll(async () => {
  delete process.env.DATABASE_URL;
  await admin?.end({ timeout: 5 });
  await closeRepositorySqlForTests();
});

function datasetService(): DatasetService {
  return new DatasetService({ databaseUrl: TEST_DATABASE_URL });
}

function versionService(): DatasetVersionService {
  return new DatasetVersionService({ databaseUrl: TEST_DATABASE_URL });
}

run("Dataset registry + versions against LIVE Postgres", () => {
  it("persists a dataset across independent service instances", async (ctx) => {
    if (!reachable) {
      console.warn(`[realinfra] Postgres unavailable (${unreachableReason}) — skipping honestly.`);
      ctx.skip();
      return;
    }
    const stamp = `${Date.now()}_${Math.random().toString(16).slice(2)}`;
    const datasetId = `ds_realpg_${stamp}`;
    const tenantId = `t_realpg_${stamp}`;

    await datasetService().registerDataset({
      dataset_id: datasetId,
      tenant_id: tenantId,
      project_id: "p_main",
      name: "Real Postgres dataset",
      dataset_type: "curated",
      created_by: "realinfra-test",
    });

    const reloaded = await datasetService().getDataset(datasetId);
    expect(reloaded.dataset_id).toBe(datasetId);
    expect(reloaded.tenant_id).toBe(tenantId);

    const rows = await admin!<{ tenant_id: string; name: string }[]>`
      SELECT tenant_id, name FROM opt_datasets WHERE dataset_id = ${datasetId}`;
    expect(rows).toEqual([{ tenant_id: tenantId, name: "Real Postgres dataset" }]);
  });

  it("persists monotonic immutable versions across fresh service instances", async (ctx) => {
    if (!reachable) return ctx.skip();
    const stamp = `${Date.now()}_${Math.random().toString(16).slice(2)}`;
    const datasetId = `ds_versions_${stamp}`;
    const tenantId = `t_versions_${stamp}`;
    await datasetService().registerDataset({
      dataset_id: datasetId,
      tenant_id: tenantId,
      project_id: "p_main",
      name: "Versioned dataset",
      dataset_type: "curated",
      created_by: "realinfra-test",
    });

    const v1 = await versionService().createVersion({
      dataset_id: datasetId,
      tenant_id: tenantId,
      project_id: "p_main",
      artifact_refs: ["st0re:artifact:art_realpg_1"],
      schema_version: "1",
      record_count: 10,
      created_by: "realinfra-test",
      idempotency_key: `idem_${stamp}_1`,
    });
    const v2 = await versionService().createVersion({
      dataset_id: datasetId,
      tenant_id: tenantId,
      project_id: "p_main",
      artifact_refs: ["st0re:artifact:art_realpg_2"],
      schema_version: "1",
      record_count: 11,
      created_by: "realinfra-test",
      idempotency_key: `idem_${stamp}_2`,
    });

    expect([v1.version, v2.version]).toEqual([1, 2]);
    const fresh = versionService();
    expect((await fresh.listVersions(datasetId, tenantId)).map((item) => item.version)).toEqual([1, 2]);
    expect((await fresh.getVersion(datasetId, tenantId, 2)).artifact_refs).toEqual(["st0re:artifact:art_realpg_2"]);
  });

  it("replays the same idempotency key without minting another version and rejects payload drift", async (ctx) => {
    if (!reachable) return ctx.skip();
    const stamp = `${Date.now()}_${Math.random().toString(16).slice(2)}`;
    const datasetId = `ds_idem_${stamp}`;
    const tenantId = `t_idem_${stamp}`;
    await datasetService().registerDataset({
      dataset_id: datasetId,
      tenant_id: tenantId,
      project_id: "p_main",
      name: "Idempotent dataset",
      dataset_type: "derived",
      created_by: "realinfra-test",
    });
    const input = {
      dataset_id: datasetId,
      tenant_id: tenantId,
      project_id: "p_main",
      artifact_refs: ["st0re:artifact:art_idem_1"],
      schema_version: "1",
      created_by: "realinfra-test",
      idempotency_key: `idem_${stamp}`,
    } as const;

    const first = await versionService().createVersion(input);
    const replay = await versionService().createVersion(input);
    expect(replay.version).toBe(first.version);
    expect(await versionService().listVersions(datasetId, tenantId)).toHaveLength(1);

    await expect(versionService().createVersion({
      ...input,
      artifact_refs: ["st0re:artifact:art_DIFFERENT"],
    })).rejects.toMatchObject({ code: "CONFLICT" });
  });

  it("foreign tenant cannot read or version another tenant's dataset", async (ctx) => {
    if (!reachable) return ctx.skip();
    const stamp = `${Date.now()}_${Math.random().toString(16).slice(2)}`;
    const datasetId = `ds_tenant_${stamp}`;
    const owner = `t_owner_${stamp}`;
    const foreign = `t_foreign_${stamp}`;
    await datasetService().registerDataset({
      dataset_id: datasetId,
      tenant_id: owner,
      project_id: "p_main",
      name: "Tenant isolated",
      dataset_type: "raw",
      created_by: "realinfra-test",
    });

    await expect(versionService().listVersions(datasetId, foreign)).rejects.toMatchObject({ code: "NOT_FOUND" });
    await expect(versionService().createVersion({
      dataset_id: datasetId,
      tenant_id: foreign,
      project_id: "p_main",
      artifact_refs: ["st0re:artifact:art_foreign"],
      schema_version: "1",
      created_by: "intruder",
    })).rejects.toMatchObject({ code: "NOT_FOUND" });
  });

  it("durable AUD0 producer rows are written alongside dataset operations", async (ctx) => {
    if (!reachable) return ctx.skip();
    const stamp = `${Date.now()}_${Math.random().toString(16).slice(2)}`;
    const datasetId = `ds_evidence_${stamp}`;
    const tenantId = `t_evidence_${stamp}`;
    await datasetService().registerDataset({
      dataset_id: datasetId,
      tenant_id: tenantId,
      project_id: "p_main",
      name: "Evidence dataset",
      dataset_type: "curated",
      created_by: "realinfra-test",
    });
    await versionService().createVersion({
      dataset_id: datasetId,
      tenant_id: tenantId,
      project_id: "p_main",
      artifact_refs: ["st0re:artifact:art_evidence"],
      schema_version: "1",
      created_by: "realinfra-test",
    });

    const evidence = await admin!<{ evidence_kind: string; status: string }[]>`
      SELECT evidence_kind, status
      FROM opt_aud0_evidence_outbox
      WHERE tenant_id = ${tenantId}
      ORDER BY created_at`;
    expect(evidence.length).toBeGreaterThanOrEqual(3); // register + version event + lineage
    expect(evidence.every((row) => row.status === "pending")).toBe(true);
    expect(new Set(evidence.map((row) => row.evidence_kind))).toEqual(new Set(["event", "openlineage"]));
  });
});