Gateway: watch API, pairing, notifications, APNs sender, CLI, tests

gateway/src/notify.ts

@@ -0,0 +1,133 @@
+// Decides when to push. A thread notifies when its phase becomes one the user cares about
+// (waiting for them, finished, failed), and again only after it moves on. A new approval on a
+// thread already waiting counts as a change, so each request notifies once. Nothing notifies while
+// a host loads its first snapshot, so a restart never replays old news.
+
+import { threadPhase, type Phase } from "./awareness.ts";
+import type { PushCategory, PushPayload } from "./contract.ts";
+import type { HostRow } from "./db.ts";
+import type { T3Host, ThreadChange } from "./t3/host.ts";
+import { trimText, type SummaryCache, type ThreadSummary } from "./views.ts";
+
+export interface Push {
+  title: string;
+  subtitle?: string;
+  body: string;
+  category: PushCategory;
+  data: PushPayload;
+  /** Replaces an earlier push about the same thing on the watch. */
+  collapseId: string;
+}
+
+export type SendPush = (push: Push) => Promise<void>;
+
+const NOTIFYING: ReadonlySet<Phase> = new Set(["waiting_for_approval", "waiting_for_input", "completed", "failed"]);
+const BODY_CHARS = 180;
+const SUMMARY_WAIT_MS = 3_000;
+const TOKEN_WARNING_MS = 7 * 24 * 60 * 60_000;
+
+export class Notifier {
+  /** The last phase key seen per thread, `phase:requestId`. */
+  private readonly keys = new Map<string, string>();
+  /** Hosts already warned about, so each warning goes out once a day at most. */
+  private readonly warned = new Map<string, number>();
+
+  constructor(
+    private readonly send: SendPush,
+    private readonly cache: SummaryCache,
+    private readonly log: (message: string) => void = () => {},
+  ) {}
+
+  onThreadChange = ({ host, previous, next, initial }: ThreadChange) => {
+    const id = `${host.id}:${(next ?? previous)?.id}`;
+    if (!next) {
+      this.keys.delete(id);
+      return;
+    }
+    const phase = threadPhase(next);
+    const key = `${phase}:${next.pendingRuntimeRequest?.id ?? ""}`;
+    const before = this.keys.get(id);
+    this.keys.set(id, key);
+    if (initial || phase === null || !NOTIFYING.has(phase) || key === before) return;
+    void this.notify(host, next, phase).catch((error: unknown) =>
+      this.log(`push for ${id} failed: ${error instanceof Error ? error.message : String(error)}`),
+    );
+  };
+
+  /** Warns once when a host starts rejecting the gateway's token. */
+  onHostStatus = (host: T3Host) => {
+    if (host.status !== "auth_invalid") return;
+    this.warnHost(host.id, host.label, `T3 rejected the gateway's token. Pair ${host.label} again.`);
+  };
+
+  /** Warns about tokens that expire within a week. Run it daily. */
+  checkTokenExpiry(rows: HostRow[], now = Date.now()) {
+    for (const row of rows) {
+      if (!row.token_expires_at) continue;
+      const left = Date.parse(row.token_expires_at) - now;
+      if (left > TOKEN_WARNING_MS) continue;
+      const days = Math.max(0, Math.ceil(left / (24 * 60 * 60_000)));
+      this.warnHost(row.id, row.label, `The gateway's T3 token expires in ${days} day${days === 1 ? "" : "s"}. Pair ${row.label} again.`);
+    }
+  }
+
+  private warnHost(hostId: string, label: string, body: string) {
+    const last = this.warned.get(hostId) ?? 0;
+    if (Date.now() - last < 24 * 60 * 60_000) return;
+    this.warned.set(hostId, Date.now());
+    void this.send({
+      title: label,
+      body,
+      category: "host",
+      data: { hostId, threadId: "" },
+      collapseId: `host:${hostId}`,
+    }).catch(() => {});
+  }
+
+  private async notify(host: T3Host, thread: NonNullable<ThreadChange["next"]>, phase: Phase) {
+    const summary = await Promise.race([
+      this.cache.get(host, thread).catch(() => undefined),
+      Bun.sleep(SUMMARY_WAIT_MS).then(() => undefined),
+    ]);
+    const project = host.projects.get(thread.projectId)?.title;
+    const requestId = thread.pendingRuntimeRequest?.id;
+    await this.send({
+      title: thread.title,
+      subtitle: project ? `${host.label} · ${project}` : host.label,
+      body: body(phase, thread.lastError ?? null, summary),
+      category: category(phase),
+      data: { hostId: host.id, threadId: thread.id, ...(requestId ? { requestId } : {}) },
+      collapseId: `${host.id}:${thread.id}`.slice(0, 64),
+    });
+  }
+}
+
+function category(phase: Phase): PushCategory {
+  switch (phase) {
+    case "waiting_for_approval":
+      return "approval";
+    case "waiting_for_input":
+      return "input";
+    case "failed":
+      return "failed";
+    default:
+      return "finished";
+  }
+}
+
+function body(phase: Phase, lastError: string | null, summary: ThreadSummary | undefined): string {
+  switch (phase) {
+    case "waiting_for_approval": {
+      const prompt = summary?.request?.prompt;
+      return prompt ? `Wants to run: ${trimText(prompt, BODY_CHARS)}` : "Needs your approval.";
+    }
+    case "waiting_for_input": {
+      const question = summary?.request?.questions[0]?.question;
+      return question ? trimText(question, BODY_CHARS) : "Has a question for you.";
+    }
+    case "failed":
+      return lastError ? trimText(lastError, BODY_CHARS) : "The run failed.";
+    default:
+      return summary?.lastAssistantMessage ? trimText(summary.lastAssistantMessage, BODY_CHARS) : "Finished.";
+  }
+}