Wake phones through the push relay (GRYT-1656)

dist-selfhosted/config.env

@@ -41,6 +41,9 @@ METRICS_PORT=0
 # IMAGE_WORKER_POLL_MS=1000
 
 # ── Optional settings ─────────────────────────────────────────────
+# Phone notifications go through push.gryt.chat, which sees a random ID and "new message",
+# never who or what. Set to off to turn them off, or to your own relay's https URL.
+# GRYT_PUSH_RELAY_URL=https://push.gryt.chat
 # CORS_ORIGIN=*
 # JWT_SECRET=change-me-to-something-random
 # REFRESH_TOKEN_TTL_DAYS=7

src/db/sqlite/connection.ts

@@ -247,6 +247,17 @@ function createSchema(d: DatabaseSync): void {
       updated_at TEXT NOT NULL
     );
 
+    -- Phones to wake through the push relay (GRYT-1656). The capability is all the
+    -- relay needs; the server never sees the Apple or Google token behind it.
+    CREATE TABLE IF NOT EXISTS push_devices (
+      server_user_id TEXT NOT NULL,
+      install_id TEXT NOT NULL,
+      capability TEXT NOT NULL,
+      created_at TEXT NOT NULL,
+      updated_at TEXT NOT NULL,
+      PRIMARY KEY (server_user_id, install_id)
+    );
+
     -- Friends on this server (GRYT-1471), one row per pair with the smaller id
     -- first. Each person's whole list is on their own devices.
     CREATE TABLE IF NOT EXISTS friendships (

src/db/sqlite/index.ts

@@ -19,6 +19,7 @@ export * from "./reports";
 export * from "./userReports";
 export * from "./blocks";
 export * from "./contactPrefs";
+export * from "./pushDevices";
 export * from "./friends";
 export * from "./mentions";
 export * from "./webhooks";

src/db/sqlite/mergeGuest.test.ts

@@ -217,7 +217,7 @@ describe("merging a guest into an account that is already a member", () => {
       "users.server_user_id", "roles.server_user_id", "conversation_members.server_user_id",
       "mentions.server_user_id", "refresh_tokens.server_user_id",
       "mls_devices.server_user_id", "mls_key_packages.server_user_id", "mls_welcomes.server_user_id",
-      "mls_removed_devices.server_user_id",
+      "mls_removed_devices.server_user_id", "push_devices.server_user_id",
     ]);
     const listed = new Set(AUTHOR_COLUMNS.map(([t, c]) => `${t}.${c}`));
     const tables = db.prepare(`SELECT name FROM sqlite_master WHERE type = 'table'`).all() as { name: string }[];

src/db/sqlite/mergeGuest.ts

@@ -147,6 +147,8 @@ function mergeInTransaction(
   );
 
   db.prepare(`UPDATE refresh_tokens SET revoked = 1 WHERE gryt_user_id = ? AND revoked = 0`).run(guestGrytUserId);
+  // The phone registers again as the account when it reconnects.
+  db.prepare(`DELETE FROM push_devices WHERE server_user_id = ?`).run(from);
   db.prepare(`DELETE FROM users WHERE server_user_id = ?`).run(from);
 
   return { guestServerUserId: from, accountServerUserId: to, ownerMoved, conversationIds };

src/db/sqlite/pushDevices.test.ts

@@ -0,0 +1,39 @@
+import assert from "node:assert/strict";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { after, before, describe, it } from "node:test";
+
+import { initSqlite } from "./connection";
+import { listPushDevices, MAX_PUSH_DEVICES_PER_USER, removePushCapability, removePushDevice, savePushDevice } from "./pushDevices";
+
+const cap = (c: string) => `p_${c.repeat(43)}`;
+
+describe("push devices", () => {
+  const dir = mkdtempSync(join(tmpdir(), "gryt-push-"));
+  before(async () => {
+    process.env.DATA_DIR = dir;
+    await initSqlite();
+  });
+  after(() => rmSync(dir, { recursive: true, force: true }));
+
+  it("keeps one row per install and replaces its capability", () => {
+    savePushDevice("u1", "install-1", cap("a"));
+    savePushDevice("u1", "install-1", cap("b"));
+    assert.deepEqual(listPushDevices("u1"), [{ installId: "install-1", capability: cap("b") }]);
+    removePushDevice("u1", "install-1");
+    assert.deepEqual(listPushDevices("u1"), []);
+  });
+
+  it("drops a dead capability from every account that had it", () => {
+    savePushDevice("u2", "install-2", cap("c"));
+    savePushDevice("u3", "install-2", cap("c"));
+    removePushCapability(cap("c"));
+    assert.deepEqual([...listPushDevices("u2"), ...listPushDevices("u3")], []);
+  });
+
+  it("holds at most ten phones per account", () => {
+    for (let i = 0; i < MAX_PUSH_DEVICES_PER_USER + 3; i++) savePushDevice("u4", `install-${String(i).padStart(2, "0")}`, cap("d"));
+    assert.equal(listPushDevices("u4").length, MAX_PUSH_DEVICES_PER_USER);
+  });
+});

src/db/sqlite/pushDevices.ts

@@ -0,0 +1,39 @@
+import { getSqliteDb, toIso } from "./connection";
+
+/** A phone that asked to be woken for this account (GRYT-1656). */
+export interface PushDevice {
+  installId: string;
+  capability: string;
+}
+
+/** More than this and the oldest goes. Nobody has ten phones; a reinstall loop might. */
+export const MAX_PUSH_DEVICES_PER_USER = 10;
+
+export function savePushDevice(serverUserId: string, installId: string, capability: string): void {
+  const db = getSqliteDb();
+  const now = toIso(new Date());
+  db.prepare(
+    `INSERT INTO push_devices (server_user_id, install_id, capability, created_at, updated_at) VALUES (?, ?, ?, ?, ?)
+     ON CONFLICT(server_user_id, install_id) DO UPDATE SET capability = excluded.capability, updated_at = excluded.updated_at`,
+  ).run(serverUserId, installId, capability, now, now);
+  db.prepare(
+    `DELETE FROM push_devices WHERE server_user_id = ? AND install_id NOT IN (
+       SELECT install_id FROM push_devices WHERE server_user_id = ? ORDER BY updated_at DESC, install_id LIMIT ?)`,
+  ).run(serverUserId, serverUserId, MAX_PUSH_DEVICES_PER_USER);
+}
+
+export function listPushDevices(serverUserId: string): PushDevice[] {
+  const rows = getSqliteDb()
+    .prepare(`SELECT install_id, capability FROM push_devices WHERE server_user_id = ?`)
+    .all(serverUserId) as { install_id: string; capability: string }[];
+  return rows.map((r) => ({ installId: r.install_id, capability: r.capability }));
+}
+
+export function removePushDevice(serverUserId: string, installId: string): void {
+  getSqliteDb().prepare(`DELETE FROM push_devices WHERE server_user_id = ? AND install_id = ?`).run(serverUserId, installId);
+}
+
+/** The relay said the phone is gone, so every account on it here stops trying. */
+export function removePushCapability(capability: string): void {
+  getSqliteDb().prepare(`DELETE FROM push_devices WHERE capability = ?`).run(capability);
+}

src/services/push.test.ts

@@ -0,0 +1,104 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+
+import type { PushDevice } from "../db";
+import type { Clients } from "../types";
+import { createPusher, isPresent, readPushRelay } from "./push";
+
+const CAP_A = `p_${"a".repeat(43)}`;
+const CAP_B = `p_${"b".repeat(43)}`;
+
+function client(serverUserId: string, extra: Partial<Clients[string]> = {}): Clients[string] {
+  return { serverUserId, isAFK: false, ...extra } as Clients[string];
+}
+
+function harness(devices: Record<string, PushDevice[]>, status = 202) {
+  const calls: { url: string; auth: string; body: string }[] = [];
+  const forgotten: string[] = [];
+  let now = 1_000_000;
+  const pusher = createPusher({
+    relay: "https://push.test",
+    listDevices: (id) => devices[id] ?? [],
+    forget: (cap) => forgotten.push(cap),
+    fetch: (async (url: string, init: RequestInit) => {
+      calls.push({ url, auth: (init.headers as Record<string, string>).authorization, body: String(init.body) });
+      return new Response(null, { status });
+    }) as unknown as typeof fetch,
+    now: () => now,
+  });
+  return { pusher, calls, forgotten, tick: (ms: number) => (now += ms) };
+}
+
+const flush = () => new Promise((r) => setImmediate(r));
+
+describe("push relay URL", () => {
+  it("defaults to the public relay and can be turned off", () => {
+    assert.equal(readPushRelay({}), "https://push.gryt.chat");
+    assert.equal(readPushRelay({ GRYT_PUSH_RELAY_URL: "off" }), null);
+    assert.equal(readPushRelay({ GRYT_PUSH_RELAY_URL: "" }), null);
+    assert.equal(readPushRelay({ GRYT_PUSH_RELAY_URL: "https://push.example.org/" }), "https://push.example.org");
+  });
+
+  it("refuses plain http except to this machine", () => {
+    assert.equal(readPushRelay({ GRYT_PUSH_RELAY_URL: "http://push.example.org" }), null);
+    assert.equal(readPushRelay({ GRYT_PUSH_RELAY_URL: "http://localhost:8080" }), "http://localhost:8080");
+  });
+});
+
+describe("who is at a screen", () => {
+  it("counts a socket that is neither idle nor a backgrounded phone", () => {
+    assert.equal(isPresent({ a: client("u1") }, "u1"), true);
+    assert.equal(isPresent({ a: client("u1", { isAFK: true }) }, "u1"), false);
+    assert.equal(isPresent({ a: client("u1", { appInBackground: true }) }, "u1"), false);
+    assert.equal(isPresent({ a: client("u1", { appInBackground: true }), b: client("u1") }, "u1"), true);
+    assert.equal(isPresent({ a: client("u2") }, "u1"), false);
+  });
+});
+
+describe("pushing", () => {
+  it("sends the capability in the header and only the kind in the body", async () => {
+    const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }] });
+    h.pusher.notify({}, ["u1"], "dm", "conv");
+    await flush();
+    assert.deepEqual(h.calls, [{ url: "https://push.test/v1/push", auth: `Bearer ${CAP_A}`, body: JSON.stringify({ kind: "dm" }) }]);
+  });
+
+  it("skips somebody who is at a screen, and wakes them once they put the phone away", async () => {
+    const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }] });
+    h.pusher.notify({ s: client("u1") }, ["u1"], "mention", "conv");
+    await flush();
+    assert.equal(h.calls.length, 0);
+    h.pusher.notify({ s: client("u1", { appInBackground: true }) }, ["u1"], "mention", "conv");
+    await flush();
+    assert.equal(h.calls.length, 1);
+  });
+
+  it("buzzes once per phone per conversation in fifteen seconds", async () => {
+    const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }, { installId: "tablet-12", capability: CAP_B }] });
+    h.pusher.notify({}, ["u1"], "dm", "conv");
+    h.pusher.notify({}, ["u1", "u1"], "dm", "conv");
+    h.pusher.notify({}, ["u1"], "dm", "other");
+    h.tick(15_000);
+    h.pusher.notify({}, ["u1"], "dm", "conv");
+    await flush();
+    assert.equal(h.calls.length, 6);
+  });
+
+  it("forgets a capability the relay calls gone or unknown", async () => {
+    for (const status of [404, 410]) {
+      const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }] }, status);
+      h.pusher.notify({}, ["u1"], "dm", "conv");
+      await flush();
+      await flush();
+      assert.deepEqual(h.forgotten, [CAP_A]);
+    }
+  });
+
+  it("keeps it when the relay is only busy", async () => {
+    const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }] }, 502);
+    h.pusher.notify({}, ["u1"], "dm", "conv");
+    await flush();
+    await flush();
+    assert.deepEqual(h.forgotten, []);
+  });
+});

src/services/push.ts

@@ -0,0 +1,123 @@
+import consola from "consola";
+
+import { listPushDevices, removePushCapability, type PushDevice } from "../db";
+import type { Clients } from "../types";
+
+/**
+ * Waking phones through the push relay (GRYT-1656). The relay gets a capability and
+ * a kind, nothing else: it writes the notification text itself.
+ */
+
+export type PushKind = "mention" | "dm";
+
+export const DEFAULT_PUSH_RELAY = "https://push.gryt.chat";
+
+/** Same shape the relay hands out. Anything else is refused before it is stored. */
+export const CAPABILITY_SHAPE = /^p_[A-Za-z0-9_-]{43}$/;
+
+/** One buzz per phone per conversation in this window. A busy DM is one notification, not twenty. */
+const QUIET_MS = 15_000;
+const TIMEOUT_MS = 5_000;
+
+/** Unset means the public relay. "off" or empty turns pushes off for this server. */
+export function readPushRelay(env: NodeJS.ProcessEnv = process.env): string | null {
+  const raw = env.GRYT_PUSH_RELAY_URL;
+  if (raw === undefined) return DEFAULT_PUSH_RELAY;
+  const value = raw.trim();
+  if (value === "" || value.toLowerCase() === "off") return null;
+  try {
+    const url = new URL(value);
+    const local = url.hostname === "localhost" || url.hostname === "127.0.0.1";
+    if (url.protocol !== "https:" && !(local && url.protocol === "http:")) throw new Error("not https");
+    return url.origin + url.pathname.replace(/\/+$/, "");
+  } catch {
+    consola.warn(`GRYT_PUSH_RELAY_URL "${value}" is not an https URL, so pushes are off`);
+    return null;
+  }
+}
+
+/** Somebody at a screen: a socket that is neither idle on the desktop nor a phone in a pocket. */
+export function isPresent(clientsInfo: Clients, serverUserId: string): boolean {
+  return Object.values(clientsInfo).some(
+    (c) => c?.serverUserId === serverUserId && !c.isAFK && !c.appInBackground,
+  );
+}
+
+export interface PusherDeps {
+  relay: string | null;
+  listDevices: (serverUserId: string) => PushDevice[];
+  forget: (capability: string) => void;
+  fetch: typeof fetch;
+  now: () => number;
+}
+
+export function createPusher(deps: PusherDeps) {
+  const lastSent = new Map<string, number>();
+
+  async function send(device: PushDevice, kind: PushKind): Promise<void> {
+    try {
+      const res = await deps.fetch(`${deps.relay}/v1/push`, {
+        method: "POST",
+        headers: { authorization: `Bearer ${device.capability}`, "content-type": "application/json" },
+        body: JSON.stringify({ kind }),
+        signal: AbortSignal.timeout(TIMEOUT_MS),
+      });
+      // 404: the relay forgot it. 410: Apple or Google said the phone is gone.
+      if (res.status === 404 || res.status === 410) deps.forget(device.capability);
+      else if (!res.ok && res.status !== 429) consola.debug(`push relay answered ${res.status}`);
+    } catch (err) {
+      consola.debug("push relay unreachable", (err as Error).message);
+    }
+  }
+
+  /** Fire and forget: called after delivery, and never awaited by a send. */
+  function notify(clientsInfo: Clients, serverUserIds: Iterable<string>, kind: PushKind, conversationId: string): void {
+    if (!deps.relay) return;
+    const now = deps.now();
+    for (const serverUserId of new Set(serverUserIds)) {
+      if (isPresent(clientsInfo, serverUserId)) continue;
+      let devices: PushDevice[];
+      try {
+        devices = deps.listDevices(serverUserId);
+      } catch (err) {
+        consola.warn("listing push devices failed", err);
+        continue;
+      }
+      for (const device of devices) {
+        const key = `${device.capability}:${conversationId}`;
+        if (now - (lastSent.get(key) ?? 0) < QUIET_MS) continue;
+        lastSent.set(key, now);
+        void send(device, kind);
+      }
+    }
+    if (lastSent.size > 5_000) {
+      for (const [key, at] of lastSent) if (now - at >= QUIET_MS) lastSent.delete(key);
+    }
+  }
+
+  return { notify };
+}
+
+let shared: ReturnType<typeof createPusher> | null = null;
+let relay: string | null | undefined;
+
+/** Read once, so a bad value warns once rather than on every message. */
+function configuredRelay(): string | null {
+  if (relay === undefined) relay = readPushRelay();
+  return relay;
+}
+
+export function pushNotify(clientsInfo: Clients, serverUserIds: Iterable<string>, kind: PushKind, conversationId: string): void {
+  shared ??= createPusher({
+    relay: configuredRelay(),
+    listDevices: listPushDevices,
+    forget: removePushCapability,
+    fetch,
+    now: Date.now,
+  });
+  shared.notify(clientsInfo, serverUserIds, kind, conversationId);
+}
+
+export function pushEnabled(): boolean {
+  return configuredRelay() !== null;
+}