Gateway: status route with per-route results, test push, reconnect and unpair

gateway/src/api.ts

@@ -11,19 +11,25 @@ import type {
   ProjectsResponse,
   PushRegistration,
   RespondRequest,
+  PushResult,
   SendMessageRequest,
+  StatusResponse,
 } from "./contract.ts";
 import type { Db, DeviceRow } from "./db.ts";
 import type { Hub } from "./hub.ts";
+import type { Pusher } from "./push.ts";
 import { HostError, type T3Host } from "./t3/host.ts";
 import { RpcError } from "./t3/rpc.ts";
 import type { ApprovalDecision, ModelSelection, ThreadShell } from "./t3/types.ts";
-import { buildFeed, buildSummary, buildThreadDetail, hostSummary, type SummaryCache } from "./views.ts";
+import { buildFeed, buildSummary, buildThreadDetail, hostStatusDetail, hostSummary, type SummaryCache } from "./views.ts";
 
 export interface ApiDeps {
   db: Db;
   hub: Hub;
   cache: SummaryCache;
+  pusher: Pusher;
+  /** When the gateway started, for the status screen. */
+  startedAt: string;
   log?: (message: string) => void;
 }
 
@@ -50,7 +56,7 @@ class HttpError extends Error {
 const json = (body: unknown, status = 200) => Response.json(body, { status });
 const accepted = () => new Response(null, { status: 202 });
 
-export function createRoutes({ db, hub, cache, log = () => {} }: ApiDeps) {
+export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {} }: ApiDeps) {
   let pairFailures: number[] = [];
 
   /** Wraps a handler with device auth and error mapping. */
@@ -131,6 +137,52 @@ export function createRoutes({ db, hub, cache, log = () => {} }: ApiDeps) {
       }),
     },
 
+    "/v1/device": {
+      /** Unpairs the watch asking. */
+      DELETE: authed<"/v1/device">((_req, device) => {
+        db.devices.revoke(device.id);
+        log(`unpaired device ${device.name} (${device.id})`);
+        return new Response(null, { status: 204 });
+      }),
+    },
+
+    "/v1/device/test-push": {
+      POST: authed<"/v1/device/test-push">(async (_req, device) =>
+        json(
+          (await pusher.sendTo(device, {
+            title: "WristCode",
+            body: "Test notification. Pushes reach this watch.",
+            category: "test",
+            data: {},
+            collapseId: "test",
+          })) satisfies PushResult,
+        ),
+      ),
+    },
+
+    "/v1/status": {
+      GET: authed((_req, device) => {
+        const expiries = new Map(hub.rows().map((row) => [row.id, row.token_expires_at]));
+        return json({
+          hosts: hub.list().map((host) => hostStatusDetail(host, expiries.get(host.id) ?? null)),
+          gateway: { startedAt, pushConfigured: pusher.configured, lastPush: pusher.last },
+          device: {
+            name: device.name,
+            pairedAt: device.created_at,
+            pushRegistered: device.apns_token !== null,
+            apnsEnvironment: device.apns_env,
+          },
+        } satisfies StatusResponse);
+      }),
+    },
+
+    "/v1/hosts/:hostId/reconnect": {
+      POST: authed<"/v1/hosts/:hostId/reconnect">((req) => {
+        findHost(req.params.hostId).reconnect();
+        return accepted();
+      }),
+    },
+
     "/v1/feed": {
       GET: authed(async () => json(await buildFeed(hub.list(), cache))),
     },

gateway/src/contract.ts

@@ -13,6 +13,64 @@ export interface HostSummary {
   error: string | null;
 }
 
+/**
+ * How a route reaches its host, guessed from the URL: `public` for HTTPS, `vpn` when the hostname
+ * says vpn, and `direct` for the rest.
+ */
+export type RouteKind = "direct" | "vpn" | "public";
+
+export interface Route {
+  url: string;
+  kind: RouteKind;
+  /** From the latest connect. `untried` when an earlier route answered first. */
+  result: "ok" | "failed" | "untried";
+  /** Why it failed, such as `refused` or `no answer in 5 s`. */
+  error: string | null;
+  checkedAt: string | null;
+}
+
+/** One host as the status screens show it. */
+export interface HostStatusDetail extends HostSummary {
+  /** The route in use. Null unless online. */
+  activeUrl: string | null;
+  /** Preferred first, the order the gateway tries them. */
+  routes: Route[];
+  onlineSince: string | null;
+  lastOnlineAt: string | null;
+  /** When the gateway tries again, while it waits to retry. */
+  nextRetryAt: string | null;
+  tokenExpiresAt: string | null;
+  serverVersion: string | null;
+}
+
+/** What Apple answered for one push. */
+export interface PushResult {
+  ok: boolean;
+  at: string;
+  /** Apple's HTTP status, or 0 when the push never reached Apple. */
+  status: number;
+  /** Apple's reason, such as `BadDeviceToken`, or why the gateway could not send. */
+  reason: string | null;
+}
+
+export interface StatusResponse {
+  hosts: HostStatusDetail[];
+  gateway: {
+    startedAt: string;
+    /** False when the gateway has no APNs key, so nothing can be pushed. */
+    pushConfigured: boolean;
+    lastPush: PushResult | null;
+  };
+  /** The watch asking. */
+  device: {
+    name: string;
+    pairedAt: string;
+    /** True once the watch has sent its APNs token. */
+    pushRegistered: boolean;
+    apnsEnvironment: "sandbox" | "production" | null;
+  };
+}
+
 /** Something a thread is waiting on: an approval, or questions for the user. */
 export interface PendingRequest {
   id: string;
@@ -137,11 +195,12 @@ export interface ErrorResponse {
 
 /** Custom keys on every push notification, beside `aps`. */
 export interface PushPayload {
-  hostId: string;
+  /** Absent on `test` pushes. */
+  hostId?: string;
   /** Absent on `host` pushes, which are about a host rather than a thread. */
   threadId?: string;
   requestId?: string;
 }
 
 /** Notification categories the watch registers actions for. */
-export type PushCategory = "approval" | "finished" | "failed" | "input" | "host";
+export type PushCategory = "approval" | "finished" | "failed" | "input" | "host" | "test";

gateway/src/main.ts

@@ -9,7 +9,8 @@ import { createRoutes } from "./api.ts";
 import { loadConfig } from "./config.ts";
 import { openDb } from "./db.ts";
 import { Hub } from "./hub.ts";
-import { Notifier, type SendPush } from "./notify.ts";
+import { Notifier } from "./notify.ts";
+import { Pusher } from "./push.ts";
 import { SummaryCache } from "./views.ts";
 
 const HOST_SYNC_MS = 15_000;
@@ -17,6 +18,7 @@ const TOKEN_CHECK_MS = 6 * 60 * 60_000;
 
 const log = (message: string) => console.log(`${new Date().toISOString()} ${message}`);
 
+const startedAt = new Date().toISOString();
 const config = loadConfig();
 mkdirSync(dirname(config.dbPath), { recursive: true });
 const db = openDb(config.dbPath);
@@ -24,37 +26,9 @@ const cache = new SummaryCache(log);
 const apns = config.apns ? new Apns(config.apns) : null;
 if (!apns) log("APNs is not configured. Pushes will only be logged.");
 
-const sendPush: SendPush = async (push) => {
-  const devices = db.devices.active().filter((device) => device.apns_token && device.apns_env);
-  if (!apns || devices.length === 0) {
-    log(`push (not sent): [${push.category}] ${push.title}: ${push.body}`);
-    return;
-  }
-  const payload = {
-    aps: {
-      alert: { title: push.title, ...(push.subtitle ? { subtitle: push.subtitle } : {}), body: push.body },
-      sound: "default",
-      category: push.category,
-      "thread-id": push.data.threadId ?? push.data.hostId,
-    },
-    ...push.data,
-  };
-  await Promise.all(
-    devices.map(async (device) => {
-      const result = await apns.send({
-        deviceToken: device.apns_token!,
-        environment: device.apns_env!,
-        payload,
-        collapseId: push.collapseId,
-      });
-      if (result.ok) return;
-      log(`APNs refused a push to ${device.name}: ${result.status} ${result.reason}`);
-      if (result.tokenInvalid) db.devices.clearPush(device.apns_token!);
-    }),
-  );
-};
+const pusher = new Pusher(db, apns, log);
 
-const notifier = new Notifier(sendPush, cache, log);
+const notifier = new Notifier(pusher.sendAll, cache, log);
 const hub = new Hub(db, { onThreadChange: notifier.onThreadChange, onStatus: notifier.onHostStatus, log });
 hub.sync();
 setInterval(() => hub.sync(), HOST_SYNC_MS);
@@ -63,7 +37,7 @@ setInterval(() => notifier.checkTokenExpiry(hub.rows()), TOKEN_CHECK_MS);
 
 const server = Bun.serve({
   port: config.port,
-  routes: createRoutes({ db, hub, cache, log }),
+  routes: createRoutes({ db, hub, cache, pusher, startedAt, log }),
   fetch: () => Response.json({ error: "Not found." }, { status: 404 }),
 });
 log(`WristCode gateway listening on :${server.port} with ${hub.list().length} host(s)`);

gateway/src/push.ts

@@ -0,0 +1,63 @@
+// Delivers pushes to paired watches through APNs, and remembers Apple's last answer so the watch's
+// status screen can show whether pushes are getting through. Without APNs settings it only logs.
+
+import type { Apns } from "./apns.ts";
+import type { PushResult } from "./contract.ts";
+import type { Db, DeviceRow } from "./db.ts";
+import type { Push } from "./notify.ts";
+
+export class Pusher {
+  /** Apple's answer to the most recent push sent to any watch. */
+  last: PushResult | null = null;
+
+  constructor(
+    private readonly db: Db,
+    private readonly apns: Apns | null,
+    private readonly log: (message: string) => void = () => {},
+  ) {}
+
+  get configured() {
+    return this.apns !== null;
+  }
+
+  /** Sends to every paired watch that registered for pushes. */
+  sendAll = async (push: Push): Promise<void> => {
+    const devices = this.db.devices.active().filter((device) => device.apns_token && device.apns_env);
+    if (!this.apns || devices.length === 0) {
+      this.log(`push (not sent): [${push.category}] ${push.title}: ${push.body}`);
+      return;
+    }
+    await Promise.all(devices.map((device) => this.sendTo(device, push)));
+  };
+
+  /** Sends to one watch and returns what Apple said. */
+  async sendTo(device: DeviceRow, push: Push): Promise<PushResult> {
+    const at = new Date().toISOString();
+    if (!this.apns) return { ok: false, at, status: 0, reason: "The gateway has no APNs key configured." };
+    if (!device.apns_token || !device.apns_env) {
+      return { ok: false, at, status: 0, reason: "This watch has not registered for notifications." };
+    }
+    const payload = {
+      aps: {
+        alert: { title: push.title, ...(push.subtitle ? { subtitle: push.subtitle } : {}), body: push.body },
+        sound: "default",
+        category: push.category,
+        "thread-id": push.data.threadId ?? push.data.hostId ?? push.category,
+      },
+      ...push.data,
+    };
+    const answer = await this.apns.send({
+      deviceToken: device.apns_token,
+      environment: device.apns_env,
+      payload,
+      collapseId: push.collapseId,
+    });
+    const result: PushResult = answer.ok ? { ok: true, at, status: 200, reason: null } : { ok: false, at, status: answer.status, reason: answer.reason };
+    this.last = result;
+    if (!answer.ok) {
+      this.log(`APNs refused a push to ${device.name}: ${answer.status} ${answer.reason}`);
+      if (answer.tokenInvalid) this.db.devices.clearPush(device.apns_token);
+    }
+    return result;
+  }
+}

gateway/src/t3/host.ts

@@ -35,6 +35,13 @@ export interface HostConfig {
   environmentId: string;
 }
 
+/** What happened the last time the gateway tried one route. */
+export interface RouteAttempt {
+  ok: boolean;
+  error: string | null;
+  at: string;
+}
+
 export interface ThreadChange {
   host: T3Host;
   previous: ThreadShell | undefined;
@@ -71,6 +78,15 @@ export class T3Host {
   readonly projects = new Map<string, Project>();
   readonly threads = new Map<string, ThreadShell>();
   serverConfig: ServerConfig | null = null;
+  serverVersion: string | null = null;
+  /** Results of the latest connect, by route. Routes after the one that answered were not tried. */
+  readonly routeAttempts = new Map<string, RouteAttempt>();
+  /** When the current connection came up. Null when not online. */
+  onlineSince: string | null = null;
+  /** When the last connection ended. Null if it never connected or is online now. */
+  lastOnlineAt: string | null = null;
+  /** When the next connect is scheduled, while waiting to retry. */
+  nextRetryAt: string | null = null;
 
   private socket: RpcSocket | null = null;
   private sequence: number | null = null;
@@ -78,6 +94,8 @@ export class T3Host {
   private stopped = false;
   private attempt = 0;
   private retryTimer: ReturnType<typeof setTimeout> | null = null;
+  /** Set by reconnect() so a deliberate close connects again at once, with no backoff. */
+  private reconnecting = false;
   /** The route this connection uses. Chosen afresh, preferred first, on every connect. */
   private activeUrl: string;
 
@@ -114,6 +132,22 @@ export class T3Host {
     this.socket?.close();
   }
 
+  /**
+   * Connects again now, starting from the preferred route. For when a better route may be back,
+   * such as the LAN after getting home, or to skip a backoff wait. Does nothing mid-connect.
+   */
+  reconnect() {
+    if (this.stopped || this.status === "connecting") return;
+    this.attempt = 0;
+    if (this.socket) {
+      this.reconnecting = true;
+      this.socket.close();
+      return;
+    }
+    if (this.retryTimer) clearTimeout(this.retryTimer);
+    void this.connect();
+  }
+
   /** Sends one orchestration command and resolves with its sequence. */
   async dispatch(command: Command): Promise<{ sequence: number }> {
     return this.requireSocket().call("orchestration.dispatchCommand", command);
@@ -162,6 +196,11 @@ export class T3Host {
   }
 
   private setStatus(status: HostStatus, error: string | null = null) {
+    if (status === "online") this.onlineSince = new Date().toISOString();
+    else if (this.status === "online") {
+      this.lastOnlineAt = new Date().toISOString();
+      this.onlineSince = null;
+    }
     this.status = status;
     this.lastError = error;
     this.options.onStatus?.(this);
@@ -169,6 +208,7 @@ export class T3Host {
 
   private async connect() {
     if (this.stopped) return;
+    this.nextRetryAt = null;
     this.setStatus("connecting", this.lastError);
     try {
       const descriptor = await this.pickRoute();
@@ -192,6 +232,11 @@ export class T3Host {
       this.setStatus("online");
       await socket.closed;
       this.socket = null;
+      if (this.reconnecting) {
+        this.reconnecting = false;
+        void this.connect();
+        return;
+      }
       this.scheduleRetry("offline", "connection closed");
     } catch (error) {
       this.socket?.close();
@@ -206,6 +251,7 @@ export class T3Host {
   /** Uses the first route whose well-known descriptor answers, and returns that descriptor. */
   private async pickRoute(): Promise<EnvironmentDescriptor> {
     const failures: string[] = [];
+    this.routeAttempts.clear();
     for (const url of this.config.urls) {
       try {
         const response = await fetch(new URL("/.well-known/t3/environment", url), {
@@ -215,9 +261,13 @@ export class T3Host {
         const descriptor = (await response.json()) as EnvironmentDescriptor;
         if (url !== this.activeUrl) this.options.log?.(`${this.label}: using ${url}`);
         this.activeUrl = url;
+        this.serverVersion = descriptor.serverVersion ?? null;
+        this.routeAttempts.set(url, { ok: true, error: null, at: new Date().toISOString() });
         return descriptor;
       } catch (error) {
-        failures.push(`${new URL(url).host} ${error instanceof Error ? error.message : String(error)}`);
+        const message = routeError(error);
+        this.routeAttempts.set(url, { ok: false, error: message, at: new Date().toISOString() });
+        failures.push(`${new URL(url).host} ${message}`);
       }
     }
     throw new HostError(`no route answered (${failures.join(", ")})`, 503);
@@ -246,6 +296,7 @@ export class T3Host {
       status === "auth_invalid" || status === "incompatible"
         ? AUTH_RETRY_MS
         : Math.min(BACKOFF_MAX_MS, BACKOFF_MIN_MS * 2 ** this.attempt++) * (0.5 + Math.random() / 2);
+    this.nextRetryAt = new Date(Date.now() + delay).toISOString();
     this.retryTimer = setTimeout(() => void this.connect(), delay);
   }
 
@@ -301,3 +352,12 @@ export class T3Host {
     this.options.onThreadChange?.({ host: this, previous, next, initial });
   }
 }
+
+/** A short reason a route failed, worded for a watch screen. */
+function routeError(error: unknown): string {
+  if (error instanceof DOMException && error.name === "TimeoutError") return `no answer in ${ROUTE_TIMEOUT_MS / 1000} s`;
+  const message = error instanceof Error ? error.message : String(error);
+  if (/ECONNREFUSED|Unable to connect|refused/i.test(message)) return "refused";
+  if (/ENOTFOUND|getaddrinfo|DNS/i.test(message)) return "name not found";
+  return message;
+}

gateway/src/views.ts

@@ -4,7 +4,17 @@
 // the thread's `updatedAt` moves.
 
 import { threadPhase, type Phase } from "./awareness.ts";
-import type { Feed, FeedItem, HostSummary, Message, PendingRequest, Summary, ThreadDetail } from "./contract.ts";
+import type {
+  Feed,
+  FeedItem,
+  HostStatusDetail,
+  HostSummary,
+  Message,
+  PendingRequest,
+  RouteKind,
+  Summary,
+  ThreadDetail,
+} from "./contract.ts";
 import type { T3Host } from "./t3/host.ts";
 import {
   isApprovalRequest,
@@ -141,6 +151,35 @@ export const hostSummary = (host: T3Host): HostSummary => ({
   error: host.lastError,
 });
 
+/** A host with its routes, for the status screens. `tokenExpiresAt` comes from the hosts table. */
+export function hostStatusDetail(host: T3Host, tokenExpiresAt: string | null): HostStatusDetail {
+  return {
+    ...hostSummary(host),
+    activeUrl: host.status === "online" ? host.url : null,
+    routes: host.config.urls.map((url) => {
+      const attempt = host.routeAttempts.get(url);
+      return {
+        url,
+        kind: routeKind(url),
+        result: attempt ? (attempt.ok ? "ok" : "failed") : "untried",
+        error: attempt?.error ?? null,
+        checkedAt: attempt?.at ?? null,
+      };
+    }),
+    onlineSince: host.onlineSince,
+    lastOnlineAt: host.lastOnlineAt,
+    nextRetryAt: host.nextRetryAt,
+    tokenExpiresAt,
+    serverVersion: host.serverVersion,
+  };
+}
+
+export function routeKind(url: string): RouteKind {
+  const { protocol, hostname } = new URL(url);
+  if (protocol === "https:") return "public";
+  return hostname.includes("vpn") ? "vpn" : "direct";
+}
+
 function feedItem(host: T3Host, thread: ThreadShell, phase: Phase, summary: ThreadSummary | undefined): FeedItem {
   return {
     hostId: host.id,

gateway/test/host.test.ts

@@ -5,16 +5,23 @@ import type { Server } from "bun";
 import { T3Host } from "../src/t3/host.ts";
 
 const ENVIRONMENT_ID = "env-laptop";
-let server: Server<undefined> | undefined;
-afterEach(() => server?.stop(true));
+let servers: Server<undefined>[] = [];
+afterEach(() => {
+  for (const server of servers) server.stop(true);
+  servers = [];
+});
 
-/** Just enough of a T3 server: the descriptor, a ticket, config, and an empty shell snapshot. */
-function fakeT3() {
-  server = Bun.serve({
+/**
+ * Just enough of a T3 server: the descriptor, a ticket, config, and an empty shell snapshot.
+ * While `state.up` is false its descriptor returns 503, like a route that is down.
+ */
+function fakeT3(state = { up: true }) {
+  const server = Bun.serve({
     port: 0,
     fetch(req, srv) {
       const { pathname } = new URL(req.url);
       if (pathname === "/.well-known/t3/environment") {
+        if (!state.up) return new Response("down", { status: 503 });
         return Response.json({ environmentId: ENVIRONMENT_ID, label: "framework-gib", serverVersion: "0.1.5", orchestrationProtocolVersion: 2 });
       }
       if (pathname === "/api/auth/websocket-ticket/rpc") return Response.json({ ticket: "t" });
@@ -34,25 +41,44 @@ function fakeT3() {
       },
     },
   });
+  servers.push(server);
   return `http://localhost:${server.port}`;
 }
 
+/** Resolves once `check` passes, polling every few milliseconds. */
+async function until(check: () => boolean) {
+  while (!check()) await Bun.sleep(5);
+}
+
 describe("T3Host routes", () => {
   it("falls back to the next route when the preferred one does not answer", async () => {
     const working = fakeT3();
     // Port 9 (discard) refuses connections, standing in for a LAN route that is down.
-    const host = new T3Host({ id: ENVIRONMENT_ID, label: "Laptop", urls: ["http://127.0.0.1:9", working], token: "x", environmentId: ENVIRONMENT_ID });
-    const online = new Promise<void>((resolve) => {
-      const check = setInterval(() => {
-        if (host.status === "online") {
-          clearInterval(check);
-          resolve();
-        }
-      }, 5);
-    });
+    const refused = "http://127.0.0.1:9";
+    const spare = "https://t3laptop.example.org";
+    const host = new T3Host({ id: ENVIRONMENT_ID, label: "Laptop", urls: [refused, working, spare], token: "x", environmentId: ENVIRONMENT_ID });
     host.start();
-    await online;
+    await until(() => host.status === "online");
     expect(host.url).toBe(working);
+    expect(host.routeAttempts.get(refused)).toMatchObject({ ok: false, error: "refused" });
+    expect(host.routeAttempts.get(working)).toMatchObject({ ok: true, error: null });
+    expect(host.routeAttempts.has(spare)).toBe(false);
+    host.stop();
+  });
+
+  it("moves back to the preferred route on reconnect once it answers", async () => {
+    const preferredState = { up: false };
+    const preferred = fakeT3(preferredState);
+    const fallback = fakeT3();
+    const host = new T3Host({ id: ENVIRONMENT_ID, label: "Laptop", urls: [preferred, fallback], token: "x", environmentId: ENVIRONMENT_ID });
+    host.start();
+    await until(() => host.status === "online");
+    expect(host.url).toBe(fallback);
+
+    preferredState.up = true;
+    host.reconnect();
+    await until(() => host.status === "online" && host.url === preferred);
+    expect(host.lastOnlineAt).not.toBeNull();
     host.stop();
   });
 });

gateway/test/views.test.ts

@@ -2,7 +2,7 @@ import { describe, expect, it } from "bun:test";
 
 import type { BoundedThread, ThreadShell } from "../src/t3/types.ts";
 import type { T3Host } from "../src/t3/host.ts";
-import { buildSummary, summarize, trimText } from "../src/views.ts";
+import { buildSummary, routeKind, summarize, trimText } from "../src/views.ts";
 
 const thread = (overrides: Partial<ThreadShell> = {}): ThreadShell => ({
   id: "thread-1",
@@ -129,3 +129,12 @@ describe("buildSummary", () => {
     });
   });
 });
+
+describe("routeKind", () => {
+  it("tells direct, VPN and public routes apart", () => {
+    expect(routeKind("http://desktop.gib:3773")).toBe("direct");
+    expect(routeKind("http://192.168.2.2:3773")).toBe("direct");
+    expect(routeKind("http://laptopvpn.gib:3773")).toBe("vpn");
+    expect(routeKind("https://t3laptopvpn.gbrown.org")).toBe("public");
+  });
+});