nucleic/cloud/nucleic-edge/src/index.ts

// Nucleic edge Worker.
//
//   Phase A — DAI heartbeat (daily active installs) → Workers Analytics Engine.
//   Phase B — WebSocket relay (Durable Object `Room`) + relay-token admission + relay DAU.
//   Phase C — APNS approval push (relay-triggered, content-free).
//
// See docs/CLOUD_INFRA.md and cloud/nucleic-edge/README.md.

import type { Env } from "./env";
import { parseHeartbeat, type Heartbeat } from "./heartbeat";
import { adoptLiveActivityPayload, apnsConfig, approvalPayload, clearPayload, liveActivityPayload, reconnectPayload, sendApns, sendApnsAdopt, sendApnsClear, sendApnsReconnect, sendLiveActivityApns } from "./apns";
import { authedHostId, enrollHost, registerPushToken, wakeableRecord, type PushRegisterBody } from "./push";
import { handleApply, handleReview } from "./testers";
import {
  CONNECTION_TTL,
  MEMBERSHIP_TTL,
  mintClaims,
  signToken,
  type TokenClaims,
  verifyToken,
} from "./relayToken";
import { authedHostRoom, enrollChallenge, enrollComplete } from "./relayEnroll";

export { Room } from "./room";

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const { pathname } = url;

    if (request.method === "GET" && pathname === "/health") return new Response("ok\n");

    if (pathname === "/v1/heartbeat") return method(request, "POST", () => handleHeartbeat(request, env));

    // --- relay ---
    if (pathname === "/relay") return handleRelay(request, env, url);
    // Proof-of-possession enrollment: the host proves it holds the X25519 static private key that
    // derives its roomId (relayEnroll.ts) before it can mint tokens for that room.
    if (pathname === "/v1/relay/enroll/challenge")
      return method(request, "POST", async () => enrollChallenge(env.NUCLEIC_RELAY_TOKENS, await safeJson(request) as { staticKey?: unknown } | null));
    if (pathname === "/v1/relay/enroll")
      return method(request, "POST", async () => enrollComplete(env.NUCLEIC_RELAY_TOKENS, await safeJson(request) as { challengeId?: unknown; proof?: unknown } | null));
    if (pathname === "/v1/relay/token") return method(request, "POST", () => mintMembership(request, env));
    if (pathname === "/v1/relay/connect") return method(request, "POST", () => mintConnection(request, env));
    if (pathname === "/v1/relay/revoke") return method(request, "POST", () => revoke(request, env));

    // --- push (out-of-band wake; the in-band path is the Room "wake" control frame) ---
    if (pathname === "/v1/host/enroll")
      return method(request, "POST", async () => enrollHost(env.NUCLEIC_PUSH_TOKENS));
    if (pathname === "/v1/push/register") return method(request, "POST", () => pushRegister(request, env));
    if (pathname === "/v1/push/notify") return method(request, "POST", () => pushNotify(request, env));
    if (pathname === "/v1/push/clear") return method(request, "POST", () => pushClear(request, env));
    if (pathname === "/v1/push/liveActivity") return method(request, "POST", () => pushLiveActivity(request, env));
    if (pathname === "/v1/push/liveActivityAdopt") return method(request, "POST", () => pushLiveActivityAdopt(request, env));
    if (pathname === "/v1/push/reconnect") return method(request, "POST", () => pushReconnect(request, env));

    // --- trusted-tester program (apply is open + CORS'd for the site form; review is HMAC-gated) ---
    if (pathname === "/v1/tester/apply") return handleApply(request, env);
    if (pathname === "/v1/tester/review") return handleReview(request, env, url);

    return new Response("not found", { status: 404 });
  },
} satisfies ExportedHandler<Env>;

// MARK: - DAI — daily active installs (Phase A)

async function handleHeartbeat(request: Request, env: Env): Promise<Response> {
  let body: unknown;
  try {
    body = await request.json();
  } catch {
    return new Response("bad json", { status: 400 });
  }
  const parsed = parseHeartbeat(body);
  if (!parsed.ok) return new Response(`bad field: ${parsed.error}`, { status: 422 });
  // No IP read or stored — the heartbeat is anonymous.
  writeHeartbeat(env, parsed.value);
  return new Response(null, { status: 204 });
}

// Indexed by the anonymous install UUID → `count(DISTINCT index1)` is a daily-active-*install*
// count (one user across two devices counts twice). For a per-user count, see writeRelayDau.
function writeHeartbeat(env: Env, hb: Heartbeat): void {
  env.NUCLEIC_DAI.writeDataPoint({
    indexes: [hb.installId],
    blobs: [hb.platform, hb.osVersion, hb.channel, hb.appVersion, hb.build, hb.arch, hb.locale],
    doubles: [1],
  });
}

// MARK: - Relay (Phase B)

async function handleRelay(request: Request, env: Env, url: URL): Promise<Response> {
  if (request.headers.get("upgrade") !== "websocket") {
    return new Response("expected websocket", { status: 426 });
  }
  if (!env.RELAY_TOKEN_SECRET) return new Response("relay not configured", { status: 503 });

  const token = url.searchParams.get("t") ?? bearer(request);
  if (!token) return new Response("missing token", { status: 401 });

  const v = await verifyToken(env.RELAY_TOKEN_SECRET, token);
  if (!v.ok) return new Response(`token ${v.error}`, { status: 401 });
  if (v.claims.typ !== "c") return new Response("connection token required", { status: 401 });
  // Defense in depth for the reserved deviceId (also rejected at mint time — see
  // mintMembership): a client tagged "host" would collide with the room's role tag.
  if (v.claims.deviceId === "host" && v.claims.role !== "host") {
    return new Response("reserved deviceId", { status: 403 });
  }
  if (await env.NUCLEIC_RELAY_TOKENS.get(`revoked:${v.claims.deviceId}`)) {
    return new Response("revoked", { status: 403 });
  }

  // Admission succeeded — count this connection toward the relay DAU (indexed by roomId ≈ user).
  writeRelayDau(env, v.claims);

  // Route to the room's DO, carrying the verified identity (the DO trusts these headers because
  // only this Worker can reach it).
  const stub = env.ROOMS.get(env.ROOMS.idFromName(v.claims.roomId));
  const headers = new Headers(request.headers);
  headers.set("x-nucleic-role", v.claims.role);
  headers.set("x-nucleic-device", v.claims.deviceId);
  // The room's *name* — a DO can't recover it from its own id, and the presence→wake hook
  // (docs/CARBON_RUNNER.md §3) needs the roomId to name the runner pool's room.
  headers.set("x-nucleic-room", v.claims.roomId);
  return stub.fetch(new Request(url.toString(), { method: request.method, headers, body: request.body }));
}

// Relay DAU — a true daily-active-*user* metric. Unlike the anonymous DAI heartbeat (indexed by a
// random install UUID), the relay carries a real identity: `roomId` is derived from the host's
// static public key (docs/CLOUD_INFRA.md §2.1), so one room ≈ one user/account. Indexing on it
// makes `count(DISTINCT index1)` count users, not installs. `role`/`deviceId` ride along as blobs
// for slicing (e.g. host vs client, distinct devices per user). Emitted on every admitted
// connection; same-day reconnects collapse under the DISTINCT.
function writeRelayDau(env: Env, claims: TokenClaims): void {
  env.NUCLEIC_RELAY_DAU.writeDataPoint({
    indexes: [claims.roomId],
    blobs: [claims.role, claims.deviceId],
    doubles: [1],
  });
}

/** Host mints a long-lived membership token for a device in *its own* room. The room is the
 * PoP-proven binding from enrollment (relayEnroll.ts) — never a client-supplied value — so a host
 * can only ever mint for the room whose static private key it proved possession of. The admin
 * bearer remains an operator override that may name any room explicitly. */
async function mintMembership(request: Request, env: Env): Promise<Response> {
  if (!env.RELAY_TOKEN_SECRET) return new Response("relay not configured", { status: 503 });

  const body = (await safeJson(request)) as { roomId?: string; deviceId?: string; role?: string } | null;
  if (!body?.deviceId) return new Response("deviceId required", { status: 422 });
  // "host" is the room's role tag in the DO's hibernation-tag namespace — a *client* named
  // "host" would be matched by `getWebSockets("host")` and blackhole the room's routing.
  if (body.deviceId === "host" && body.role !== "host") {
    return new Response("reserved deviceId", { status: 422 });
  }

  let roomId: string;
  if (adminOk(request, env)) {
    // Operator override: an explicit roomId is required and trusted.
    if (!body.roomId) return new Response("roomId required", { status: 422 });
    roomId = body.roomId;
  } else {
    const host = await authedHostRoom(env.NUCLEIC_RELAY_TOKENS, bearer(request));
    if (!host) return unauthorized();
    roomId = host.roomId; // ignore any client-supplied roomId
  }

  const claims = mintClaims({
    roomId,
    deviceId: body.deviceId,
    role: body.role === "host" ? "host" : "client",
    typ: "m",
    ttlSeconds: MEMBERSHIP_TTL,
  });
  return json({ token: await signToken(env.RELAY_TOKEN_SECRET, claims), exp: claims.exp });
}

/** Trade a membership token for a short-lived connection token (presented on the WS upgrade). */
async function mintConnection(request: Request, env: Env): Promise<Response> {
  if (!env.RELAY_TOKEN_SECRET) return new Response("relay not configured", { status: 503 });

  const m = bearer(request);
  const v = m ? await verifyToken(env.RELAY_TOKEN_SECRET, m) : null;
  if (!v || !v.ok || v.claims.typ !== "m") return unauthorized();
  if (await env.NUCLEIC_RELAY_TOKENS.get(`revoked:${v.claims.deviceId}`)) return new Response("revoked", { status: 403 });

  const claims = mintClaims({
    roomId: v.claims.roomId,
    deviceId: v.claims.deviceId,
    role: v.claims.role,
    typ: "c",
    ttlSeconds: CONNECTION_TTL,
  });
  return json({ token: await signToken(env.RELAY_TOKEN_SECRET, claims), exp: claims.exp });
}

/** Revoke a device on unpair (admin). KV TTL outlives the longest membership token. */
async function revoke(request: Request, env: Env): Promise<Response> {
  if (!adminOk(request, env)) return unauthorized();
  const body = (await safeJson(request)) as { deviceId?: string } | null;
  if (!body?.deviceId) return new Response("deviceId required", { status: 422 });
  await env.NUCLEIC_RELAY_TOKENS.put(`revoked:${body.deviceId}`, "1", { expirationTtl: MEMBERSHIP_TTL });
  return json({ revoked: body.deviceId });
}

// MARK: - Push (Phase C, out-of-band)

/** The caller's push identity: a self-enrolled host, or the operator's admin bearer
 * (which may act on any device). Null ⇒ unauthorized. */
async function pushCaller(request: Request, env: Env): Promise<{ hostId: string; isAdmin: boolean } | null> {
  if (adminOk(request, env)) return { hostId: "admin", isAdmin: true };
  const hostId = await authedHostId(env.NUCLEIC_PUSH_TOKENS, bearer(request));
  return hostId ? { hostId, isAdmin: false } : null;
}

/** Register a phone's APNS token (the host uploads it after the sync Hello). */
async function pushRegister(request: Request, env: Env): Promise<Response> {
  const caller = await pushCaller(request, env);
  if (!caller) return unauthorized();
  const body = (await safeJson(request)) as PushRegisterBody | null;
  return registerPushToken(env.NUCLEIC_PUSH_TOKENS, caller.hostId, body);
}

async function pushNotify(request: Request, env: Env): Promise<Response> {
  const caller = await pushCaller(request, env);
  if (!caller) return unauthorized();
  const cfg = apnsConfig(env);
  if (!cfg) return new Response("apns not configured", { status: 503 });

  const body = (await safeJson(request)) as { deviceId?: string; kind?: string } | null;
  if (!body?.deviceId) return new Response("deviceId required", { status: 422 });
  // Scoped: a host may only wake devices it registered (404 either way, so a probe can't
  // distinguish "unknown device" from "someone else's device").
  const rec = await wakeableRecord(env.NUCLEIC_PUSH_TOKENS, caller.hostId, body.deviceId, caller.isAdmin);
  if (!rec) return new Response("no push token for device", { status: 404 });

  // The device record's environment + topic win (the host knows which build/channel its phone
  // runs); the global APNS_ENV / APNS_TOPIC secrets only back records that predate them.
  const apsEnv = rec.env ?? cfg.env;
  const topic = rec.topic ?? cfg.topic;
  // `kind:"question"` ⇒ the block is an agent question, so the alert reads "waiting for your
  // answer" rather than "…approval". Absent/any other value keeps the approval wording.
  const result = await sendApns({ ...cfg, env: apsEnv, topic }, rec.token, approvalPayload(body.kind === "question"));
  logPush("notify", body.deviceId, topic, apsEnv, result);
  return json(result, result.status === 200 ? 200 : 502);
}

/** Recall a device's approval wake tickle — the counterpart to `pushNotify`. The host calls this
 * once an approval is resolved (on any device) and nothing else is pending, so a phone still
 * showing "A session is waiting for your approval" clears it. Same ownership scoping as wake. */
async function pushClear(request: Request, env: Env): Promise<Response> {
  const caller = await pushCaller(request, env);
  if (!caller) return unauthorized();
  const cfg = apnsConfig(env);
  if (!cfg) return new Response("apns not configured", { status: 503 });

  const body = (await safeJson(request)) as { deviceId?: string } | null;
  if (!body?.deviceId) return new Response("deviceId required", { status: 422 });
  const rec = await wakeableRecord(env.NUCLEIC_PUSH_TOKENS, caller.hostId, body.deviceId, caller.isAdmin);
  if (!rec) return new Response("no push token for device", { status: 404 });

  const apsEnv = rec.env ?? cfg.env;
  const topic = rec.topic ?? cfg.topic;
  const result = await sendApnsClear({ ...cfg, env: apsEnv, topic }, rec.token, clearPayload());
  logPush("clear", body.deviceId, topic, apsEnv, result);
  return json(result, result.status === 200 ? 200 : 502);
}

/** Silently wake a phone so it adopts a push-started Live Activity — comes online and registers the
 * new activity's update token so the host can keep the glance fresh and end it cleanly (UX_IOS §5.3).
 * A content-free `content-available` background push to the device token (not the activity token).
 * Same ownership scoping as wake. */
async function pushLiveActivityAdopt(request: Request, env: Env): Promise<Response> {
  const caller = await pushCaller(request, env);
  if (!caller) return unauthorized();
  const cfg = apnsConfig(env);
  if (!cfg) return new Response("apns not configured", { status: 503 });

  const body = (await safeJson(request)) as { deviceId?: string } | null;
  if (!body?.deviceId) return new Response("deviceId required", { status: 422 });
  const rec = await wakeableRecord(env.NUCLEIC_PUSH_TOKENS, caller.hostId, body.deviceId, caller.isAdmin);
  if (!rec) return new Response("no push token for device", { status: 404 });

  const apsEnv = rec.env ?? cfg.env;
  const topic = rec.topic ?? cfg.topic;
  const result = await sendApnsAdopt({ ...cfg, env: apsEnv, topic }, rec.token, adoptLiveActivityPayload());
  logPush("liveActivityAdopt", body.deviceId, topic, apsEnv, result);
  return json(result, result.status === 200 ? 200 : 502);
}

/** Silently nudge a device to reconnect over a non-local transport (tailnet/relay) after its LAN
 * link drops (SYNC_PROTOCOL §3.2). A content-free `content-available` background push to the device
 * token — it grants the app a moment of runtime to re-run its dial loop from anywhere. Same
 * ownership scoping as wake: a host may only nudge devices it registered. */
async function pushReconnect(request: Request, env: Env): Promise<Response> {
  const caller = await pushCaller(request, env);
  if (!caller) return unauthorized();
  const cfg = apnsConfig(env);
  if (!cfg) return new Response("apns not configured", { status: 503 });

  const body = (await safeJson(request)) as { deviceId?: string } | null;
  if (!body?.deviceId) return new Response("deviceId required", { status: 422 });
  const rec = await wakeableRecord(env.NUCLEIC_PUSH_TOKENS, caller.hostId, body.deviceId, caller.isAdmin);
  if (!rec) return new Response("no push token for device", { status: 404 });

  const apsEnv = rec.env ?? cfg.env;
  const topic = rec.topic ?? cfg.topic;
  const result = await sendApnsReconnect({ ...cfg, env: apsEnv, topic }, rec.token, reconnectPayload());
  logPush("reconnect", body.deviceId, topic, apsEnv, result);
  return json(result, result.status === 200 ? 200 : 502);
}

/** Surface a push send in `wrangler tail` — previously every APNs failure (a `BadTopic` for a
 * mis-addressed per-channel build, a `BadDeviceToken` for an env mismatch) vanished silently.
 * `result.reason` carries APNs's rejection string. */
function logPush(
  kind: string,
  deviceId: string,
  topic: string,
  env: string,
  result: { status: number; reason?: string },
): void {
  const line = `push ${kind} device=${deviceId} topic=${topic} env=${env} status=${result.status}${result.reason ? ` reason=${result.reason}` : ""}`;
  if (result.status === 200) console.log(line);
  else console.warn(line);
}

/** Push a Live Activity content-state update to a phone's activity (UX_IOS §5.3). The activity's
 * push token rides inline (it's per-activity, not the device token); we still scope on device
 * ownership — a host may only push to a device it registered — so the `deviceId` gates it. */
async function pushLiveActivity(request: Request, env: Env): Promise<Response> {
  const caller = await pushCaller(request, env);
  if (!caller) return unauthorized();
  const cfg = apnsConfig(env);
  if (!cfg) return new Response("apns not configured", { status: 503 });

  const body = (await safeJson(request)) as
    | {
        deviceId?: string;
        token?: string;
        env?: string;
        topic?: string;
        event?: string;
        contentState?: unknown;
        // Present only on a push-to-start (`event:"start"`): the Activity's static attributes and
        // the type name they must decode into on the phone.
        attributes?: unknown;
        attributesType?: string;
        // Set on an `update` that should sound/vibrate through the glance (a session newly needs
        // the user while the phone was away): "approval" | "input". The host sends this instead of
        // a separate wake tickle / banner.
        alert?: string;
      }
    | null;
  if (!body?.deviceId || !body.token) return new Response("deviceId and token required", { status: 422 });
  // Same scoping as wake: a host may only reach devices it registered (404 either way).
  const rec = await wakeableRecord(env.NUCLEIC_PUSH_TOKENS, caller.hostId, body.deviceId, caller.isAdmin);
  if (!rec) return new Response("no push token for device", { status: 404 });

  const apsEnv =
    body.env === "production" ? "production" : body.env === "sandbox" ? "sandbox" : (rec.env ?? cfg.env);
  // The push body's topic wins (freshest, from this phone's hello), then the stored record, then
  // the worker's configured fallback. `sendLiveActivityApns` appends `.push-type.liveactivity`.
  const topic = body.topic ?? rec.topic ?? cfg.topic;
  // A push-to-start carries the static attributes + their type name; update/end don't.
  const start =
    body.event === "start" && body.attributesType
      ? { type: body.attributesType, attributes: body.attributes ?? {} }
      : null;
  // An alerting update sounds/vibrates through the glance (replaces the wake tickle / banner for a
  // phone that already has a Live Activity); the payload builder ignores it for start/end.
  const alert = body.alert === "approval" || body.alert === "input" ? body.alert : null;
  const event = body.event ?? "update";
  const payload = liveActivityPayload(event, body.contentState ?? null, Date.now(), start, alert);
  // Mix APNs priorities so routine traffic can't starve the ActivityKit budget (which shrinks the
  // longer the app is backgrounded): a plain content-state update rides low priority `5` (off the
  // budget), while the pushes that must land promptly — the push-to-`start`, the `end`, and an
  // alerting "needs you" update — ride `10`. Without this, a busy session's frequent priority-10
  // updates exhausted the budget while the phone was away and the next push-to-start got throttled
  // — the glance failing to appear after a long background (docs/UX_IOS.md §5.3).
  const priority = event === "update" && alert === null ? "5" : "10";
  const result = await sendLiveActivityApns({ ...cfg, env: apsEnv, topic }, body.token, payload, priority);
  logPush("liveActivity", body.deviceId, `${topic}.push-type.liveactivity`, apsEnv, result);
  return json(result, result.status === 200 ? 200 : 502);
}

// MARK: - helpers

function method(request: Request, want: string, run: () => Promise<Response>): Promise<Response> | Response {
  if (request.method !== want) return new Response("method not allowed", { status: 405, headers: { allow: want } });
  return run();
}

function bearer(request: Request): string | null {
  const h = request.headers.get("authorization");
  const m = h && /^Bearer\s+(.+)$/i.exec(h);
  return m ? m[1] : null;
}

function adminOk(request: Request, env: Env): boolean {
  const want = env.RELAY_ADMIN_SECRET;
  const got = bearer(request);
  return !!want && !!got && safeEqual(got, want);
}

function safeEqual(a: string, b: string): boolean {
  if (a.length !== b.length) return false;
  let diff = 0;
  for (let i = 0; i < a.length; i++) diff |= a.charCodeAt(i) ^ b.charCodeAt(i);
  return diff === 0;
}

async function safeJson(request: Request): Promise<unknown> {
  try {
    return await request.json();
  } catch {
    return null;
  }
}

function json(obj: unknown, status = 200): Response {
  return new Response(JSON.stringify(obj), { status, headers: { "content-type": "application/json" } });
}

function unauthorized(): Response {
  return new Response("unauthorized", { status: 401 });
}