Jyotisha/frontend/tests/database-rectification-telemetry.test.ts

import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import test from "node:test";

import { closeLocalPostgresDataPool, createLocalPostgresDataClient } from "../src/lib/db/local-postgres-client-core.ts";
import { RECTIFICATION_TELEMETRY_FIELDS } from "../src/lib/rectification-agentic/v9/telemetry.ts";
import { startPostgresFixture, type PostgresFixture } from "./helpers/postgres-fixture.ts";

const runnerPath = fileURLToPath(new URL("../scripts/db-migrate.mjs", import.meta.url));
const skipWithoutDocker = spawnSync("docker", ["version", "--format", "{{.Server.Version}}"], {
  stdio: "ignore",
}).status === 0
  ? false
  : "docker unavailable on this host";

const RECORD_SIGNATURE = "public.record_rectification_telemetry(uuid,uuid,integer,text,integer,integer,integer,integer,integer,integer,integer,integer,integer,integer,text,boolean,integer,text,text,text)";

// Fictional synthetic Case snapshot; no real person.
const snapshot = {
  birth_date: "1990-01-01",
  birth_place_label: "虚构市",
  latitude: 30,
  longitude: 120,
  timezone_id: "Asia/Shanghai",
  timezone_offset: 8,
  birth_time_source: "approximate",
  reported_birth_time: "05:00",
  active_birth_time: "05:00",
  birth_time_period: null,
  uncertainty_before_minutes: 30,
  uncertainty_after_minutes: 30,
};

function rpcError(error: unknown): string {
  if (!error || typeof error !== "object") return "";
  const value = error as { message?: unknown };
  return typeof value.message === "string" ? value.message : "";
}

function metrics(overrides: Record<string, unknown> = {}) {
  return {
    p_window_radius_minutes: 30,
    p_birth_time_source: "approximate",
    p_questions_total: 7,
    p_questions_targeted: 2,
    p_questions_guided: 1,
    p_questions_dated_probe: 1,
    p_questions_personality: 1,
    p_questions_open: 1,
    p_experiences_added: 4,
    p_range_width_minutes: 12,
    p_candidate_count: 4,
    p_top_two_gap_points: 7,
    p_stop_reason: "pool_exhausted",
    p_precision_gate_met: false,
    p_duration_seconds: 1800,
    p_algorithm_version: "rectification-v5-matrix-scoring-9",
    p_policy_version: "rectification-policy-v1",
    p_skill_version: "10.0.30",
    ...overrides,
  };
}

function seedUser(fixture: PostgresFixture, email: string): string {
  fixture.psqlAs(
    "identity_runtime",
    "identity-runtime-test-password",
    `insert into identity.users (name, email, email_verified, email_verified_at) values ('Fixture', '${email}', true, now());`,
  );
  return fixture.psql(`select id from identity.users where email = '${email}'`);
}

test("rectification telemetry: server-only write, dedupe, retention, aggregate-only admin read, account cascade", {
  skip: skipWithoutDocker,
}, async () => {
  const fixture = startPostgresFixture();
  const serviceUrl = fixture.connectionUrl("service_runtime", "service-runtime-test-password");
  try {
    const migration = spawnSync(process.execPath, [runnerPath], {
      encoding: "utf8",
      env: { ...process.env, SCHEMA_DATABASE_URL: fixture.connectionUrl("schema_owner", "schema-owner-test-password") },
    });
    assert.equal(migration.status, 0, migration.stderr);

    // 1. Columns: exactly the allowlist, no identity column.
    assert.equal(
      fixture.psql(`
        select string_agg(column_name, ',' order by ordinal_position)
        from information_schema.columns
        where table_schema = 'public' and table_name = 'rectification_telemetry'
      `),
      ["id", "recorded_week", ...RECTIFICATION_TELEMETRY_FIELDS].join(","),
    );
    assert.equal(
      fixture.psql(`
        select string_agg(column_name, ',' order by ordinal_position)
        from information_schema.columns
        where table_schema = 'public' and table_name = 'rectification_telemetry_reported_cases'
      `),
      "case_id",
    );

    // 2. RLS on, no table privilege for any runtime role.
    assert.equal(
      fixture.psql(`
        select string_agg(relname || '=' || relrowsecurity::text, ',' order by relname)
        from pg_class
        where relname in ('rectification_telemetry', 'rectification_telemetry_reported_cases')
      `),
      "rectification_telemetry=true,rectification_telemetry_reported_cases=true",
    );
    const tablePrivileges = fixture.psql(`
      select string_agg(role || ':' || tbl || ':' || has_table_privilege(role, tbl, 'SELECT,INSERT,UPDATE,DELETE')::text, ',')
      from unnest(array['anon', 'authenticated', 'app_runtime', 'admin_runtime', 'service_role', 'service_runtime']) as role,
        unnest(array['public.rectification_telemetry', 'public.rectification_telemetry_reported_cases']) as tbl
    `);
    assert.doesNotMatch(tablePrivileges, /:true/, tablePrivileges);

    // 3. Function privileges: write = service_role, summary = admin_runtime.
    assert.equal(
      fixture.psql(`
        select concat_ws(':',
          has_function_privilege('service_role', '${RECORD_SIGNATURE}', 'EXECUTE'),
          has_function_privilege('admin_runtime', '${RECORD_SIGNATURE}', 'EXECUTE'),
          has_function_privilege('authenticated', '${RECORD_SIGNATURE}', 'EXECUTE'),
          has_function_privilege('app_runtime', '${RECORD_SIGNATURE}', 'EXECUTE'),
          has_function_privilege('admin_runtime', 'public.rectification_telemetry_summary(integer)', 'EXECUTE'),
          has_function_privilege('service_role', 'public.rectification_telemetry_summary(integer)', 'EXECUTE'),
          has_function_privilege('authenticated', 'public.rectification_telemetry_summary(integer)', 'EXECUTE'),
          has_function_privilege('service_role', 'public.purge_expired_rectification_telemetry()', 'EXECUTE'),
          has_function_privilege('admin_runtime', 'public.purge_expired_rectification_telemetry()', 'EXECUTE')
        )
      `),
      "t:f:f:f:t:f:f:t:f",
    );
    assert.throws(() => fixture.psqlAs(
      "admin_runtime",
      "admin-runtime-test-password",
      "select count(*) from public.rectification_telemetry",
    ));

    // 4. Seed two fictional users with one Case each.
    const firstUser = seedUser(fixture, "telemetry-one@example.com");
    const secondUser = seedUser(fixture, "telemetry-two@example.com");
    fixture.psql(`
      with provider as (
        insert into public.model_providers (code, name, provider_type, encrypted_api_key, enabled)
        values ('telemetry-test', 'Telemetry Test', 'openai', 'test-ciphertext', true)
        returning id
      ), config as (
        insert into public.model_configs (model_id) values ('telemetry-default-model') returning id
      )
      insert into public.model_config_versions (
        config_id, version, provider_id, label, provider_model, enabled, is_default, status, published_at
      )
      select config.id, 1, provider.id, 'Default', 'gpt-test', true, true, 'published', now()
      from config cross join provider;
    `);
    const service = createLocalPostgresDataClient(serviceUrl, null, "service_role");
    const openCase = async (userId: string, requestId: string) => {
      const opened = await service.rpc("open_agentic_rectification_case", {
        p_user_id: userId,
        p_request_id: requestId,
        p_intent: "homepage",
        p_session_id: null,
        p_skill_name: "jyotish-birth-time-rectification",
        p_skill_version: "10.0.30",
        p_baseline_profile_fingerprint: "c".repeat(64),
        p_baseline_birth_snapshot: snapshot,
        p_candidate_range: { start_time: "04:30", end_time: "05:30" },
      });
      assert.equal(opened.error, null, rpcError(opened.error));
      return String((opened.data as Record<string, unknown>).case_id);
    };
    const firstCase = await openCase(firstUser, "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa1");
    const secondCase = await openCase(secondUser, "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa2");

    // 5. First write stores one row; the second write for the same Case is a no-op.
    const first = await service.rpc("record_rectification_telemetry", {
      p_user_id: firstUser, p_case_id: firstCase, ...metrics(),
    });
    assert.equal(first.error, null, rpcError(first.error));
    assert.equal(first.data, true);
    const again = await service.rpc("record_rectification_telemetry", {
      p_user_id: firstUser, p_case_id: firstCase, ...metrics({ p_stop_reason: "converged" }),
    });
    assert.equal(again.error, null, rpcError(again.error));
    assert.equal(again.data, false);
    assert.equal(fixture.psql("select count(*) || ':' || min(stop_reason) from public.rectification_telemetry"), "1:pool_exhausted");
    assert.equal(
      fixture.psql("select extract(isodow from recorded_week)::int from public.rectification_telemetry"),
      "1",
    );

    // 6. Ownership: another user's id cannot write for this Case.
    const foreign = await service.rpc("record_rectification_telemetry", {
      p_user_id: firstUser, p_case_id: secondCase, ...metrics(),
    });
    assert.match(rpcError(foreign.error), /rectification_telemetry_case_not_found/);

    // 7. A rejected row rolls back the dedupe mark, so a later valid write still lands.
    for (const bad of [
      { p_stop_reason: "free text" },
      { p_birth_time_source: "1990-01-01" },
      { p_algorithm_version: "v 1 with spaces" },
      { p_questions_total: 1 },
      { p_top_two_gap_points: 101 },
    ]) {
      const rejected = await service.rpc("record_rectification_telemetry", {
        p_user_id: secondUser, p_case_id: secondCase, ...metrics(bad),
      });
      assert.ok(rejected.error, JSON.stringify(bad));
    }
    assert.equal(
      fixture.psql(`select count(*) from public.rectification_telemetry_reported_cases where case_id = '${secondCase}'`),
      "0",
    );
    const second = await service.rpc("record_rectification_telemetry", {
      p_user_id: secondUser, p_case_id: secondCase,
      ...metrics({ p_stop_reason: "converged", p_precision_gate_met: true, p_range_width_minutes: 4, p_top_two_gap_points: null }),
    });
    assert.equal(second.error, null, rpcError(second.error));
    assert.equal(second.data, true);

    // 8. Retention: rows whose week started more than 180 days ago are purged.
    fixture.psql(`
      insert into public.rectification_telemetry (
        recorded_week, birth_time_source, questions_total, questions_targeted, questions_guided,
        questions_dated_probe, questions_personality, questions_open, experiences_added,
        candidate_count, stop_reason, duration_seconds
      ) values (
        (date_trunc('week', (timezone('UTC', now()))::date - 200))::date,
        'unknown', 3, 0, 0, 0, 0, 0, 1, 2, 'user_stopped', 60
      );
    `);
    assert.equal(fixture.psql("select count(*) from public.rectification_telemetry"), "3");
    const purged = await service.rpc("purge_expired_rectification_telemetry", {});
    assert.equal(purged.error, null, rpcError(purged.error));
    assert.equal(Number(purged.data), 1);
    assert.equal(fixture.psql("select count(*) from public.rectification_telemetry"), "2");

    // 9. Admin sees aggregates only.
    const summary = JSON.parse(fixture.psqlAs(
      "admin_runtime",
      "admin-runtime-test-password",
      "select public.rectification_telemetry_summary(12)::text",
    )) as Record<string, unknown>;
    assert.equal(summary.sessions, 2);
    assert.equal(summary.retention_days, 180);
    assert.deepEqual(
      (summary.stop_reasons as Array<{ key: string; count: number }>).map((row) => `${row.key}=${row.count}`),
      ["converged=1", "pool_exhausted=1", "user_no_more=0", "round_cap=0", "user_stopped=0", "error=0"],
    );
    assert.deepEqual(summary.precision_gate, { met: 1, not_met: 1, unknown: 0 });
    const width = Object.fromEntries(
      (summary.width_distribution as Array<{ key: string; count: number }>).map((row) => [row.key, row.count]),
    );
    assert.equal(width["1-5"], 1);
    assert.equal(width["11-20"], 1);
    assert.equal(width.unknown, 0);
    const gap = Object.fromEntries(
      (summary.gap_distribution as Array<{ key: string; count: number }>).map((row) => [row.key, row.count]),
    );
    assert.equal(gap["5-9"], 1);
    assert.equal(gap.single, 1);
    assert.equal((summary.weekly as unknown[]).length, 12);
    const serialized = JSON.stringify(summary);
    for (const forbidden of [firstUser, secondUser, firstCase, secondCase, "1990-01-01", "虚构市"]) {
      assert.equal(serialized.includes(forbidden), false, `summary must not contain ${forbidden}`);
    }
    assert.equal(
      JSON.parse(fixture.psqlAs(
        "admin_runtime",
        "admin-runtime-test-password",
        "select public.rectification_telemetry_summary(999)::text",
      )).weeks,
      26,
    );

    // 10. Account deletion removes the dedupe mark with the Case; the anonymous row stays unlinkable.
    fixture.psqlAs(
      "identity_runtime",
      "identity-runtime-test-password",
      `delete from identity.users where id = '${firstUser}'`,
    );
    assert.equal(
      fixture.psql(`
        select concat_ws(':',
          (select count(*) from public.agentic_rectification_cases where id = '${firstCase}'),
          (select count(*) from public.rectification_telemetry_reported_cases where case_id = '${firstCase}'),
          (select count(*) from public.rectification_telemetry_reported_cases where case_id = '${secondCase}'),
          (select count(*) from public.rectification_telemetry)
        )
      `),
      "0:0:1:2",
    );
  } finally {
    try {
      await closeLocalPostgresDataPool(serviceUrl);
    } finally {
      fixture.stop();
    }
  }
});