mesh-catalog/modules/model-usage/test/usage.test.ts

// What holds model-usage to taking no usage event it did not write (novox/hq issue 276): a store that
// refuses is thrown, so the bus offers the event again; a redelivery writes nothing twice; the last
// delivery is spooled and taken; the spool replays once the store answers; an older event replayed
// late does not overwrite a newer reading; and an event without a body is taken, not a TypeError.
//
// The store is a fake keyed exactly as the table is, with the same rule for "latest". When
// TEST_DATABASE_URL names a throwaway postgres, the same rule is also checked against the real SQL.

import { test } from "node:test";
import assert from "node:assert/strict";
import { mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";

import type { Event } from "@novox/mesh-sdk/events";
import { rowsOf, writerFor } from "../dist/consume.js";
import { Spool, takeOrSpool } from "../dist/spool.js";
import { UsageStore } from "../dist/store.js";
import { getUsageTools } from "../dist/tools/index.js";

type Row = { licence: string; consumer: string; period: string; metric: string; value: number; raw?: unknown };

function fakeStore() {
  const rows = new Map<string, { row: Row; at: string; eventId?: string }>();
  let writes = 0;
  return {
    broken: false,
    rows,
    get writes() {
      return writes;
    },
    async upsert(row: Row, observed: { at?: string; eventId?: string } = {}) {
      if (this.broken) throw new Error("connect ECONNREFUSED 10.0.0.5:5432");
      writes++;
      const key = [row.licence, row.consumer, row.period, row.metric].join("|");
      const had = rows.get(key);
      const at = observed.at || new Date().toISOString();
      if (had && had.at > at) return; // the table's WHERE: an older reading changes nothing
      rows.set(key, { row, at, eventId: observed.eventId });
    },
  };
}

function usage(id: string, at: string, value: number, body?: unknown): Event {
  return {
    type: "anthropic-licence-manager.usage.reported",
    id,
    source: "anthropic-licence-manager",
    node: "laptop",
    at,
    body: body === undefined ? { rows: [{ licence: "team", consumer: "*", period: "2026-10", metric: "tokens", value }], raw: { v: 1 } } : body,
  };
}

async function setup() {
  const store = fakeStore();
  const dir = await mkdtemp(join(tmpdir(), "usage-"));
  const spool = await Spool.open(join(dir, "spool"));
  const write = writerFor(store, () => {});
  const handle = (e: Event) => takeOrSpool(e, write, spool, () => {});
  return { store, spool, write, handle, dir };
}

test("a store that refuses a row is thrown, so the event is offered again, and it is spooled", async () => {
  const { store, spool, handle } = await setup();
  store.broken = true;
  await assert.rejects(handle(usage("u1", "2026-10-06T10:00:00Z", 5)), /ECONNREFUSED/);
  assert.equal(spool.size, 1);
  assert.equal(spool.list()[0].held, false);
});

test("a redelivered event writes the same reading, never a second, and clears its spooled copy", async () => {
  const { store, spool, handle } = await setup();
  store.broken = true;
  await assert.rejects(handle(usage("u2", "2026-10-06T10:00:00Z", 5)));
  store.broken = false;
  await handle(usage("u2", "2026-10-06T10:00:00Z", 5));
  await handle(usage("u2", "2026-10-06T10:00:00Z", 5));
  assert.equal(store.rows.size, 1);
  assert.equal([...store.rows.values()][0].row.value, 5);
  assert.equal(spool.size, 0);
});

test("on its last delivery a usage event is spooled and taken", async () => {
  const { store, spool, handle } = await setup();
  store.broken = true;
  for (let i = 1; i < 5; i++) await assert.rejects(handle(usage("u3", "2026-10-06T10:00:00Z", 5)));
  await handle(usage("u3", "2026-10-06T10:00:00Z", 5));
  assert.equal(spool.list()[0].held, true);
  assert.equal(spool.list()[0].attempts, 5);
});

test("the spool replays once the store answers, and an older reading replayed late does not overwrite a newer", async () => {
  const { store, spool, write, handle } = await setup();
  store.broken = true;
  for (let i = 0; i < 5; i++) await handle(usage("old", "2026-10-06T10:00:00Z", 5)).catch(() => {});
  store.broken = false;
  await handle(usage("new", "2026-10-06T11:00:00Z", 9)); // a newer total arrives first
  const r = await spool.replay(write);
  assert.deepEqual(r, { replayed: 1, left: 0 });
  assert.equal([...store.rows.values()][0].row.value, 9);
  assert.equal([...store.rows.values()][0].eventId, "new");
});

test("an event without a body is taken without a TypeError; a row the store could never take is skipped, not retried", async () => {
  const { store, handle } = await setup();
  await handle(usage("n1", "2026-10-06T10:00:00Z", 0, null));
  await handle(usage("n2", "2026-10-06T10:00:00Z", 0, { rows: "nonsense" }));
  await handle(usage("n3", "2026-10-06T10:00:00Z", 0, { rows: [null, { licence: "team", consumer: "*", period: "p", metric: "m", value: "x" }] }));
  assert.equal(store.writes, 0);
  assert.deepEqual(rowsOf({ body: null }), { rows: [], skipped: [] });
  assert.equal(rowsOf({ body: { rows: [null] } }).skipped.length, 1);
});

test("model_usage_status reports the spool, and is offered even when the store cannot be opened", async () => {
  const { store, handle, dir } = await setup();
  store.broken = true;
  for (let i = 0; i < 5; i++) await handle(usage("s1", "2026-10-06T10:00:00Z", 1)).catch(() => {});
  const tools = getUsageTools(undefined, { USAGE_SPOOL: join(dir, "spool") });
  assert.deepEqual(tools.map((t) => t.name), ["model_usage_status"]);
  const got = (await tools[0].run({}, {} as never)) as { spool: { spooled: number; held: number } };
  assert.equal(got.spool.spooled, 1);
  assert.equal(got.spool.held, 1);
});

const url = process.env.TEST_DATABASE_URL;
test("against postgres: migrations are numbered and repeatable, and the upsert keeps the reading observed latest", { skip: !url && "TEST_DATABASE_URL is not set" }, async () => {
  const store = UsageStore.fromEnv({ DATABASE_URL: url });
  try {
    await store.migrate();
    assert.deepEqual(await store.migrate(), [], "a second run applies nothing");
    const row = { licence: "pg", consumer: "*", period: "2026-10", metric: "tokens", value: 5, raw: {} };
    await store.upsert(row, { at: "2026-10-06T10:00:00Z", eventId: "a" });
    await store.upsert({ ...row, value: 9 }, { at: "2026-10-06T11:00:00Z", eventId: "b" });
    await store.upsert(row, { at: "2026-10-06T10:00:00Z", eventId: "a" }); // the old one, replayed late
    await store.upsert({ ...row, value: 9 }, { at: "2026-10-06T11:00:00Z", eventId: "b" }); // redelivered
    const got = (await store.current("pg")).map((r) => r.value);
    assert.deepEqual(got, [9]);
  } finally {
    await store.close();
  }
});