0pt/tests/analytics/analytics-governance-postgres.realinfra.test.ts

/**
 * tests/analytics/analytics-governance-postgres.realinfra.test.ts
 * ================================================================
 * sql/009 against a REAL Postgres.
 *
 * A migration nobody ran is a migration that does not work, and a trigger
 * nobody fired is a comment. So this applies sql/008 and sql/009 into a
 * throwaway schema and then attempts, against the database, every write the
 * constraints exist to refuse:
 *
 *   BI-P031  UPDATE a CERTIFIED metric's meaning in place -> must RAISE.
 *            This is the layer that holds when something writes without going
 *            through the application: a migration script, a psql session, an
 *            admin tool. A TypeScript guard does not survive `psql`.
 *   BI-P030  UPDATE a published semantic model version's content -> must RAISE.
 *            Two ACTIVE versions of one model -> must violate a unique index.
 *   BI-P028  A currency-formatted metric with no currency -> must be refused.
 *            Certification without an actor or the two definitions -> refused.
 *   BI-P018  A `compiled_query_hash` that is not a 64-hex digest -> refused.
 *   BI-P021  UPDATE or DELETE a usage-ledger row -> must RAISE (append-only).
 *
 * Gated on ANALYTICS_TEST_DATABASE_URL, matching the convention in
 * analytics-domain-postgres.realinfra.test.ts. With it set and Postgres down,
 * this FAILS rather than skipping.
 *
 *   docker run -d --name pg -p 25433:5432 -e POSTGRES_PASSWORD=test \
 *     -e POSTGRES_DB=opt postgres:16-alpine
 *   ANALYTICS_TEST_DATABASE_URL=postgres://postgres:test@127.0.0.1:25433/opt \
 *     pnpm vitest run tests/analytics/analytics-governance-postgres.realinfra.test.ts
 */
import fs from 'node:fs';
import path from 'node:path';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import postgres from 'postgres';

const TEST_DATABASE_URL = process.env.ANALYTICS_TEST_DATABASE_URL;
const run = TEST_DATABASE_URL ? describe : describe.skip;
const SCHEMA = `bi_p009_${Date.now().toString(36)}`;

let sql: ReturnType<typeof postgres>;

/** Did the statement RAISE? Returns the message, or undefined if it succeeded. */
async function refusalOf(fn: () => Promise<unknown>): Promise<string | undefined> {
  try {
    await fn();
    return undefined;
  } catch (e) {
    return String((e as Error).message);
  }
}

beforeAll(async () => {
  if (!TEST_DATABASE_URL) return;
  // The search_path is pinned on the CONNECTION rather than prefixed to every
  // statement. Prefixing makes each call a multi-statement simple query, and
  // postgres.js then returns the LAST statement's result — which silently
  // turned `SELECT state FROM ...` into the connection object in the first
  // version of this file, and made two assertions compare against nonsense.
  const admin = postgres(TEST_DATABASE_URL, { max: 1, onnotice: () => undefined });
  await admin.unsafe(`CREATE SCHEMA IF NOT EXISTS ${SCHEMA}`);
  await admin.end({ timeout: 5 });

  sql = postgres(TEST_DATABASE_URL, {
    max: 2,
    onnotice: () => undefined,
    connection: { search_path: SCHEMA },
  });
  for (const file of [
    'sql/008_analytics_domain_model.sql',
    'sql/009_analytics_execution_metrics_governance.sql',
  ]) {
    await sql.unsafe(fs.readFileSync(path.join(process.cwd(), file), 'utf8'));
  }

  // A dataset and a metric to govern.
  await sql.unsafe(`
    INSERT INTO opt_analytics_dataset
      (dataset_id, tenant_id, project_id, name, tier, schema_contract_id, schema_contract_version,
       owner_id, source_ref, freshness_slo_seconds, retention_days, classification, status)
    VALUES ('ds_sales','t_alpha','p_main','sales','SERVING','c_sales','1.0.0',
            'u_1','0pt://sourcing/sales', 3600, 90, 'none', 'active')
    ON CONFLICT DO NOTHING;`);
}, 120_000);

afterAll(async () => {
  if (!TEST_DATABASE_URL || !sql) return;
  await sql.unsafe(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`);
  await sql.end({ timeout: 5 });
});

run('sql/009 applies, and replays', () => {
  it('is idempotent — a replay onto an already-migrated schema succeeds', async () => {
    const text = fs.readFileSync(
      path.join(process.cwd(), 'sql/009_analytics_execution_metrics_governance.sql'),
      'utf8',
    );
    // The restore drill replays every file in order onto a live database.
    await expect(sql.unsafe(text)).resolves.toBeDefined();
  });

  it('added every BI-P018 column the execution record needs', async () => {
    const rows = await sql.unsafe(
      `SELECT column_name FROM information_schema.columns
       WHERE table_schema = '${SCHEMA}' AND table_name = 'opt_analytics_query_execution'`,
    );
    const names = new Set((rows as unknown as Array<{ column_name: string }>).map((r) => r.column_name));
    for (const col of [
      'principal_id',
      'workload',
      'compiled_query_hash',
      'datasets',
      'metrics',
      'cache_status',
      'cost_units',
    ]) {
      expect(names.has(col), `missing ${col}`).toBe(true);
    }
    // And the one it must NOT have.
    expect(names.has('query_text')).toBe(false);
    expect(names.has('sql')).toBe(false);
  });
});

run('BI-P031 — a CERTIFIED metric cannot be modified in place, enforced by the DATABASE', () => {
  const base = '';

  beforeAll(async () => {
    await sql.unsafe(`${base}
      INSERT INTO opt_analytics_metric_definition
        (metric_id, tenant_id, project_id, name, dataset_id, aggregation, measure_field,
         allowed_dimension_ids, current_version, owner_id, description,
         metric_type, time_dimension, unit, currency, display_format, status, filters_json,
         certification, business_definition, technical_definition, meaning_hash,
         certified_at, certified_by)
      VALUES ('m_revenue','t_alpha','p_main','net_revenue','ds_sales','sum','amount_sek',
              ARRAY['country'], 3, 'finance@siax.io', 'Invoiced revenue net of credits',
              'simple','invoiced_at','SEK','SEK','currency','ACTIVE','[]',
              'certified','What we invoiced','sum(amount_sek)', repeat('a',64),
              now(), 'u_1')
      ON CONFLICT (metric_id) DO NOTHING;`);
  });

  it('REFUSES an in-place change to the meaning hash', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} UPDATE opt_analytics_metric_definition
                  SET meaning_hash = repeat('b',64) WHERE metric_id = 'm_revenue'`),
    );
    expect(message, "a certified metric's meaning was changed in place, in the database").toBeDefined();
    expect(message).toContain('BI-P031');
    expect(message).toContain('mint version 4');
  });

  it('ALLOWS an editorial change that leaves the meaning hash alone', async () => {
    await expect(
      sql.unsafe(`${base} UPDATE opt_analytics_metric_definition
                  SET description = 'Clearer wording' WHERE metric_id = 'm_revenue'`),
    ).resolves.toBeDefined();
  });

  it('ALLOWS a meaning change that comes with a version bump', async () => {
    await expect(
      sql.unsafe(`${base} UPDATE opt_analytics_metric_definition
                  SET meaning_hash = repeat('c',64), current_version = 4
                  WHERE metric_id = 'm_revenue'`),
    ).resolves.toBeDefined();
    // Put it back for the remaining assertions.
    await sql.unsafe(`${base} UPDATE opt_analytics_metric_definition
                      SET current_version = 3 WHERE metric_id = 'm_revenue'`);
  });

  it('refuses a currency-formatted metric with no currency', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} INSERT INTO opt_analytics_metric_definition
        (metric_id, tenant_id, project_id, name, dataset_id, aggregation, measure_field,
         owner_id, display_format, currency, time_dimension)
        VALUES ('m_bad','t_alpha','p_main','bad_metric','ds_sales','sum','x','u','currency',NULL,'t')`),
    );
    expect(message).toContain('currency_pairing');
  });

  it('refuses a currency on a non-currency format', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} INSERT INTO opt_analytics_metric_definition
        (metric_id, tenant_id, project_id, name, dataset_id, aggregation, measure_field,
         owner_id, display_format, currency, time_dimension)
        VALUES ('m_bad2','t_alpha','p_main','bad_metric2','ds_sales','sum','x','u','integer','SEK','t')`),
    );
    expect(message).toContain('currency_pairing');
  });

  it('refuses certification without an actor, a time, and both definitions', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} INSERT INTO opt_analytics_metric_definition
        (metric_id, tenant_id, project_id, name, dataset_id, aggregation, measure_field,
         owner_id, display_format, time_dimension, certification)
        VALUES ('m_bad3','t_alpha','p_main','bad_metric3','ds_sales','sum','x','u','decimal','t','certified')`),
    );
    expect(message).toContain('certification_act');
  });

  it('refuses an unknown lifecycle status', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} UPDATE opt_analytics_metric_definition
                  SET status = 'PROBABLY_FINE' WHERE metric_id = 'm_revenue'`),
    );
    expect(message).toContain('metric_status');
  });
});

run('BI-P030 — a published semantic model version is immutable, and only one is active', () => {
  const base = '';

  beforeAll(async () => {
    for (const [id, version, status] of [
      ['smv_1', 1, 'active'],
      ['smv_2', 2, 'draft'],
    ] as const) {
      await sql.unsafe(`${base}
        INSERT INTO opt_analytics_semantic_model_version
          (semantic_model_version_id, model_id, tenant_id, project_id, version,
           datasets, relationships_json, metrics_json, dimensions_json, status, hash)
        VALUES ('${id}','sm_sales','t_alpha','p_main',${version},
                ARRAY['ds_sales'],'[]','[]','[]','${status}', repeat('${version}',64))
        ON CONFLICT DO NOTHING;`);
    }
  });

  it("REFUSES an update to a published version's content", async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} UPDATE opt_analytics_semantic_model_version
                  SET metrics_json = '[{"id":"m_new"}]' WHERE semantic_model_version_id = 'smv_1'`),
    );
    expect(message, 'a pinned semantic model version was rewritten under a dashboard').toBeDefined();
    expect(message).toContain('BI-P030');
  });

  it('ALLOWS a status transition, which is not a content change', async () => {
    await expect(
      sql.unsafe(`${base} UPDATE opt_analytics_semantic_model_version
                  SET status = 'superseded' WHERE semantic_model_version_id = 'smv_1'`),
    ).resolves.toBeDefined();
    await sql.unsafe(`${base} UPDATE opt_analytics_semantic_model_version
                      SET status = 'active' WHERE semantic_model_version_id = 'smv_1'`);
  });

  it('refuses TWO active versions of the same model — half the dashboards would read the other one', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} UPDATE opt_analytics_semantic_model_version
                  SET status = 'active' WHERE semantic_model_version_id = 'smv_2'`),
    );
    expect(message).toContain('opt_analytics_semantic_model_one_active');
  });

  it('refuses a hash that is not a sha256 digest', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base}
        INSERT INTO opt_analytics_semantic_model_version
          (semantic_model_version_id, model_id, tenant_id, project_id, version,
           datasets, relationships_json, metrics_json, dimensions_json, status, hash)
        VALUES ('smv_bad','sm_sales','t_alpha','p_main',9,ARRAY['ds_sales'],'[]','[]','[]','draft','not-a-hash')`),
    );
    expect(message).toBeDefined();
  });
});

run('BI-P018 / BI-P021 — the execution record and the usage ledger', () => {
  const base = '';

  it('refuses a compiled_query_hash that is not a 64-hex digest', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} INSERT INTO opt_analytics_query_execution
        (execution_id, tenant_id, project_id, engine_query_id, state, provider, compiled_query_hash)
        VALUES ('e_bad','t_alpha','p_main','q1','queued','clickhouse','SELECT * FROM sales')`),
    );
    expect(message).toContain('hash_shape');
  });

  it('accepts a proper execution record', async () => {
    await expect(
      sql.unsafe(`${base} INSERT INTO opt_analytics_query_execution
        (execution_id, tenant_id, project_id, engine_query_id, state, provider,
         compiled_query_hash, principal_id, workload, cache_status, cost_units)
        VALUES ('e_ok','t_alpha','p_main','q1','succeeded','clickhouse',
                repeat('a',64),'u_1','dashboard','hit',0.5)`),
    ).resolves.toBeDefined();
  });

  it('refuses an unknown cache status and an unknown workload', async () => {
    expect(
      await refusalOf(() =>
        sql.unsafe(`${base} UPDATE opt_analytics_query_execution
                    SET cache_status = 'probably' WHERE execution_id = 'e_ok'`),
      ),
    ).toContain('cache_status');
    expect(
      await refusalOf(() =>
        sql.unsafe(`${base} UPDATE opt_analytics_query_execution
                    SET workload = 'vibes' WHERE execution_id = 'e_ok'`),
      ),
    ).toContain('workload');
  });

  it('the usage ledger is APPEND-ONLY — update and delete both raise', async () => {
    await sql.unsafe(`${base} INSERT INTO opt_analytics_usage_ledger
      (usage_id, tenant_id, principal_id, query_id, engine, rows_scanned, bytes_scanned,
       execution_ms, cost_units, terminated_by_budget)
      VALUES ('u_1','t_alpha','p_1','q1','clickhouse',1000,2000,50,0.25,true)`);

    expect(
      await refusalOf(() =>
        sql.unsafe(`${base} UPDATE opt_analytics_usage_ledger SET cost_units = 0 WHERE usage_id = 'u_1'`),
      ),
      'a billing row was editable',
    ).toBeDefined();
    expect(
      await refusalOf(() =>
        sql.unsafe(`${base} DELETE FROM opt_analytics_usage_ledger WHERE usage_id = 'u_1'`),
      ),
      'a billing row was deletable',
    ).toBeDefined();
  });

  it('an analyst statement refusal must carry a reason', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} INSERT INTO opt_analytics_sql_statement
        (statement_id, tenant_id, principal_id, statement_normalized, statement_hash, outcome)
        VALUES ('s_bad','t_alpha','u_1','SELECT ?', repeat('a',64), 'refused')`),
    );
    expect(message, 'a refusal with no reason was accepted — an unreviewable audit trail').toBeDefined();
  });

  it('a freshness row must carry a disclosure a UI can render', async () => {
    const message = await refusalOf(() =>
      sql.unsafe(`${base} INSERT INTO opt_analytics_dataset_freshness
        (tenant_id, dataset_id, state, expected_interval_ms, disclosure)
        VALUES ('t_alpha','ds_sales','STALE',3600000,'   ')`),
    );
    expect(message).toBeDefined();
  });

  it('freshness defaults to UNKNOWN, never to FRESH', async () => {
    await sql.unsafe(`${base} INSERT INTO opt_analytics_dataset_freshness
      (tenant_id, dataset_id, expected_interval_ms, disclosure)
      VALUES ('t_beta','ds_sales',3600000,'not yet computed')`);
    const rows = await sql.unsafe(`${base} SELECT state FROM opt_analytics_dataset_freshness
                                   WHERE tenant_id = 't_beta'`);
    expect((rows as unknown as Array<{ state: string }>)[0].state).toBe('UNKNOWN');
  });
});