feat(gateway): report TNT identity events to m1ngsama ID

gateway/src/events.ts

@@ -0,0 +1,94 @@
+const FLUSH_MS = 30_000;
+const BATCH = 50;
+const MAX_PENDING = 1000;
+
+export type IdentityEvent = {
+  id: string;
+  handle: string;
+  type: `tnt.${string}`;
+  at: number;
+  detail?: Record<string, string | boolean>;
+};
+
+export type ReporterOptions = {
+  issuer: string;
+  secret: string;
+  /** Pending events survive a restart here. */
+  file: string;
+  fetch?: typeof fetch;
+};
+
+/** Reports terminal identity events to m1ngsama ID, keeping them until it accepts them. */
+export function createReporter(options: ReporterOptions) {
+  const request = options.fetch ?? fetch;
+  const subs = new Map<string, string>();
+  let pending: IdentityEvent[] = [];
+  let flushing: Promise<void> | null = null;
+
+  async function load() {
+    const file = Bun.file(options.file);
+    if (!(await file.exists())) return;
+    try {
+      pending = JSON.parse(await file.text()) as IdentityEvent[];
+    } catch {
+      pending = [];
+    }
+  }
+
+  const save = () => Bun.write(options.file, JSON.stringify(pending));
+
+  async function subFor(handle: string): Promise<string | null> {
+    const known = subs.get(handle);
+    if (known) return known;
+    const res = await request(`${options.issuer}/people/${encodeURIComponent(handle)}`);
+    if (res.status === 404) return null;
+    if (!res.ok) throw new Error(`people lookup answered ${res.status}`);
+    const { sub } = (await res.json()) as { sub?: unknown };
+    if (typeof sub !== "string") return null;
+    subs.set(handle, sub);
+    return sub;
+  }
+
+  async function send(): Promise<void> {
+    while (pending.length > 0) {
+      const batch = pending.slice(0, BATCH);
+      const events = [];
+      const unknown = new Set<string>();
+      for (const event of batch) {
+        const sub = await subFor(event.handle);
+        if (sub === null) unknown.add(event.id);
+        else events.push({ id: event.id, sub, type: event.type, at: event.at, actor: { kind: "self" }, detail: event.detail });
+      }
+      if (events.length > 0) {
+        const res = await request(`${options.issuer}/internal/events`, {
+          method: "POST",
+          headers: { authorization: `Basic ${btoa(`tnt:${options.secret}`)}`, "content-type": "application/json" },
+          body: JSON.stringify({ events }),
+        });
+        if (res.status !== 202) throw new Error(`events answered ${res.status}`);
+      }
+      // A handle with no person behind it has nobody to attribute the event to.
+      pending = pending.slice(batch.length);
+      await save();
+      if (unknown.size > 0) console.error(`identity events: dropped ${unknown.size} for handles m1ngsama ID does not know`);
+    }
+  }
+
+  function flush(): Promise<void> {
+    flushing ??= send()
+      .catch((error: unknown) => console.error(`identity events: ${pending.length} waiting, ${String(error)}`))
+      .finally(() => {
+        flushing = null;
+      });
+    return flushing;
+  }
+
+  async function report(event: Omit<IdentityEvent, "id" | "at">) {
+    pending.push({ id: crypto.randomUUID(), at: Date.now(), ...event });
+    if (pending.length > MAX_PENDING) pending = pending.slice(-MAX_PENDING);
+    await save();
+    void flush();
+  }
+
+  return { load, report, flush, start: () => setInterval(() => void flush(), FLUSH_MS), pending: () => pending };
+}

gateway/src/main.ts

@@ -1,5 +1,6 @@
 import { createVerifier, remoteKeys } from "./auth";
 import { createGateway } from "./gateway";
+import { createReporter } from "./events";
 import { createGuard } from "./guard";
 import { Ledger, readHistory } from "./history";
 import { signIn } from "./identity";
@@ -22,6 +23,12 @@ const verify = createVerifier(issuer, remoteKeys(env.TNT_GATEWAY_JWKS_URL || `${
 let identity = false;
 let direct = false;
 const signingIn = new Set<string>();
+const eventsSecret = env.TNT_GATEWAY_EVENTS_SECRET;
+const reporter = eventsSecret ? createReporter({ issuer, secret: eventsSecret, file: `${stateDir}/gateway-events.json` }) : null;
+if (reporter) {
+  await reporter.load();
+  reporter.start();
+}
 const guard = createGuard({
   issuer,
   reserve: (handle) => {
@@ -62,6 +69,8 @@ for await (const line of console) {
     direct = supportsPrivate(event.version);
   } else if (event.type === "private.created") {
     gateway.handlePrivate(event);
+  } else if (event.type === "identity.changed") {
+    void reporter?.report({ handle: event.handle, type: `tnt.${event.action}`, detail: event.detail });
   } else if (event.type === "identity.requested") {
     const { session } = event;
     if (!identity || signingIn.has(session)) continue;
@@ -70,7 +79,10 @@ for await (const line of console) {
       issuer,
       verify,
       notice: (plain_text) => write(encodeRecord({ type: "identity.notice", session, plain_text })),
-      verified: (handle) => write(encodeRecord({ type: "identity.verified", session, handle })),
+      verified: (handle) => {
+        write(encodeRecord({ type: "identity.verified", session, handle }));
+        void reporter?.report({ handle, type: "tnt.signed_in", detail: { method: "device_code" } });
+      },
     }).finally(() => signingIn.delete(session));
   } else {
     if (identity && event.type === "presence.joined") void guard.check(event.nickname);

gateway/src/module.ts

@@ -2,11 +2,13 @@ export type TntEvent =
   | { type: "handshake"; version: string }
   | { type: "message.created"; message: { timestamp: string; sender: string; plain_text: string; verified: boolean } }
   | { type: "identity.requested"; session: string }
+  | { type: "identity.changed"; action: IdentityAction; handle: string; detail: Record<string, string | boolean> }
   | { type: "private.created"; from: string; to: string; plain_text: string; timestamp: string; verified: boolean }
   | { type: "presence.snapshot"; nicknames: string[] }
   | { type: "presence.joined" | "presence.left"; nickname: string };
 
-export type RoomEvent = Exclude<TntEvent, { type: "handshake" | "identity.requested" | "private.created" }>;
+export type IdentityAction = "key_recognized" | "key_bound" | "signed_out";
+export type RoomEvent = Exclude<TntEvent, { type: "handshake" | "identity.requested" | "identity.changed" | "private.created" }>;
 export type PrivateEvent = Extract<TntEvent, { type: "private.created" }>;
 
 export const HANDSHAKE_OK = `${JSON.stringify({
@@ -68,6 +70,17 @@ export function parseTntLine(line: string): { reply: string | null; event: TntEv
         ? { reply: null, event: { type: "private.created", from, to, plain_text, timestamp, verified: record.verified === true } }
         : NOTHING;
     }
+    case "identity.recognized":
+    case "identity.key_bound":
+    case "identity.signed_out": {
+      if (!isString(record.handle)) return NOTHING;
+      const action = ({ "identity.recognized": "key_recognized", "identity.key_bound": "key_bound", "identity.signed_out": "signed_out" } as const)[record.type];
+      const detail: Record<string, string | boolean> = {};
+      if (isString(record.fingerprint)) detail.fingerprint = record.fingerprint;
+      if (isString(record.previous_handle)) detail.previous_handle = record.previous_handle;
+      if (typeof record.forgot_key === "boolean") detail.forgot_key = record.forgot_key;
+      return { reply: null, event: { type: "identity.changed", action, handle: record.handle, detail } };
+    }
     case "identity.requested":
       return isString(record.session) ? { reply: null, event: { type: "identity.requested", session: record.session } } : NOTHING;
     case "presence.snapshot":

gateway/test/events.test.ts

@@ -0,0 +1,63 @@
+import { afterEach, expect, test } from "bun:test";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { createReporter } from "../src/events";
+
+let dir = "";
+afterEach(() => rmSync(dir, { recursive: true, force: true }));
+
+function setup(eventsStatus: () => number) {
+  dir = mkdtempSync(join(tmpdir(), "tnt-events-"));
+  const posted: unknown[] = [];
+  const lookups: string[] = [];
+  const fetch = (async (url: string, init?: RequestInit) => {
+    const path = new URL(url).pathname;
+    if (path.startsWith("/people/")) {
+      lookups.push(path);
+      return path === "/people/lune" ? Response.json({ sub: "sub-lune" }) : Response.json({ error: "not_found" }, { status: 404 });
+    }
+    posted.push({ authorization: new Headers(init?.headers).get("authorization"), body: JSON.parse(String(init?.body)) });
+    return new Response(null, { status: eventsStatus() });
+  }) as unknown as typeof globalThis.fetch;
+  const options = { issuer: "https://id.test", secret: "s3cret", file: join(dir, "events.json"), fetch };
+  return { reporter: createReporter(options), options, posted, lookups };
+}
+
+test("reports an event with the person's sub and forgets it once accepted", async () => {
+  const { reporter, posted, lookups } = setup(() => 202);
+  await reporter.report({ handle: "lune", type: "tnt.key_recognized", detail: { fingerprint: "SHA256:abc" } });
+  await reporter.flush();
+  await reporter.report({ handle: "lune", type: "tnt.signed_out", detail: { forgot_key: false } });
+  await reporter.flush();
+  expect(lookups).toEqual(["/people/lune"]);
+  expect(posted[0]).toMatchObject({
+    authorization: `Basic ${btoa("tnt:s3cret")}`,
+    body: { events: [{ sub: "sub-lune", type: "tnt.key_recognized", actor: { kind: "self" }, detail: { fingerprint: "SHA256:abc" } }] },
+  });
+  expect(reporter.pending()).toEqual([]);
+});
+
+test("keeps refused events on disk and sends them after a restart", async () => {
+  let status = 503;
+  const { reporter, options, posted } = setup(() => status);
+  await reporter.report({ handle: "lune", type: "tnt.signed_in", detail: { method: "device_code" } });
+  await reporter.flush();
+  expect(reporter.pending()).toHaveLength(1);
+  const id = reporter.pending()[0]!.id;
+
+  status = 202;
+  const restarted = createReporter(options);
+  await restarted.load();
+  await restarted.flush();
+  expect(restarted.pending()).toEqual([]);
+  expect((posted.at(-1) as { body: { events: { id: string }[] } }).body.events[0]!.id).toBe(id);
+});
+
+test("drops events for a handle m1ngsama ID does not know", async () => {
+  const { reporter, posted } = setup(() => 202);
+  await reporter.report({ handle: "ghost", type: "tnt.signed_out" });
+  await reporter.flush();
+  expect(posted).toEqual([]);
+  expect(reporter.pending()).toEqual([]);
+});

gateway/test/module.test.ts

@@ -49,6 +49,19 @@ describe("parseTntLine", () => {
     expect(supportsPrivate("1.6.0")).toBe(true);
   });
 
+  test("turns TNT's identity notifications into events", () => {
+    expect(parseTntLine('{"type":"identity.recognized","session":"s-8","handle":"lune","fingerprint":"SHA256:a"}').event).toEqual({
+      type: "identity.changed", action: "key_recognized", handle: "lune", detail: { fingerprint: "SHA256:a" },
+    });
+    expect(parseTntLine('{"type":"identity.key_bound","session":"s-9","handle":"lune","fingerprint":"SHA256:a","previous_handle":"zed"}').event).toEqual({
+      type: "identity.changed", action: "key_bound", handle: "lune", detail: { fingerprint: "SHA256:a", previous_handle: "zed" },
+    });
+    expect(parseTntLine('{"type":"identity.signed_out","session":"s-2","handle":"lune","forgot_key":false}')).toEqual({
+      reply: null, event: { type: "identity.changed", action: "signed_out", handle: "lune", detail: { forgot_key: false } },
+    });
+    expect(parseTntLine('{"type":"identity.signed_out","session":"s-2"}')).toEqual({ reply: null, event: null });
+  });
+
   test("parses presence records without answering them", () => {
     expect(parseTntLine('{"type":"presence.snapshot","nicknames":["a",1,"b"]}')).toEqual({
       reply: null,