crossmate

engagement-worker.js (8692B)

---

 // Coarse debounce for the durable `lastSeenAt` write on the message hot path.
 // The room TTL only needs second-ish granularity, so a write + alarm reschedule
 // on every keystroke broadcast is pure churn. In-memory, so it resets to a
 // guaranteed write on each hibernation wake — which is fine.
 const TOUCH_DEBOUNCE_MS = 30 * 1000;
 
 export class EngagementRoom {
   constructor(state, env) {
     this.state = state;
     this.env = env;
     // Answer keepalive pings without waking the Durable Object, so an otherwise
     // idle foreground socket stays warm through NAT / edge idle timeouts. Pairs
     // with the client's periodic "ping" (EngagementHost.startPing). Auto-answered
     // frames never reach `webSocketMessage`, so they cost nothing here.
     this.state.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong"));
   }
 
   async fetch(request) {
     const upgrade = request.headers.get("Upgrade");
     if (upgrade !== "websocket") {
       return new Response("Expected WebSocket upgrade", { status: 426 });
     }
 
     const url = new URL(request.url);
     const roomID = roomIDFromPath(url.pathname);
     if (!roomID) {
       return new Response("Missing room ID", { status: 400 });
     }
 
     const auth = await this.authenticate(roomID, url.searchParams);
     if (!auth.ok) {
       return new Response(auth.message, { status: auth.status });
     }
 
     const pair = new WebSocketPair();
     const [client, server] = Object.values(pair);
     server.serializeAttachment({
       authorID: auth.authorID,
       deviceID: auth.deviceID,
       connectedAt: Date.now()
     });
     // Supersede any earlier socket from the same device. A reconnect would
     // otherwise leave the stale one lingering until its TCP dies, so the room
     // fans every broadcast at a zombie and can echo the device's own prior
     // frames back to it. `server` is not accepted yet, so it is not in the set.
     for (const existing of this.state.getWebSockets()) {
       const attachment = existing.deserializeAttachment();
       if (attachment && attachment.authorID === auth.authorID && attachment.deviceID === auth.deviceID) {
         try {
           existing.close(1000, "Superseded by a newer connection");
         } catch {
           // Already closing; the runtime will reap it.
         }
       }
     }
     this.state.acceptWebSocket(server);
 
     return new Response(null, {
       status: 101,
       webSocket: client
     });
   }
 
   async authenticate(roomID, params) {
     const authorID = params.get("authorID") || "";
     const deviceID = params.get("deviceID") || "";
     const timestamp = params.get("timestamp") || "";
     const nonce = params.get("nonce") || "";
     const secret = params.get("secret") || "";
     const signature = params.get("signature") || "";
 
     if (!authorID || !deviceID || !timestamp || !nonce || !secret || !signature) {
       return { ok: false, status: 401, message: "Missing auth parameters" };
     }
 
     const nowSeconds = Math.floor(Date.now() / 1000);
     const timestampSeconds = Number(timestamp);
     const maxSkewSeconds = Number(this.env.MAX_AUTH_SKEW_SECONDS || "120");
     if (!Number.isFinite(timestampSeconds) || Math.abs(nowSeconds - timestampSeconds) > maxSkewSeconds) {
       return { ok: false, status: 401, message: "Stale auth timestamp" };
     }
 
     const nonceKey = `nonce:${nonce}`;
     if (await this.state.storage.get(nonceKey)) {
       return { ok: false, status: 401, message: "Nonce already used" };
     }
 
     const payload = [roomID, authorID, deviceID, timestamp, nonce].join("|");
     const expectedSignature = await hmacSHA256(secret, payload);
     if (!timingSafeEqual(signature, expectedSignature)) {
       return { ok: false, status: 401, message: "Invalid signature" };
     }
 
     const secretHash = await sha256(secret);
     const storedSecretHash = await this.state.storage.get("secretHash");
     if (storedSecretHash && storedSecretHash !== secretHash) {
       return { ok: false, status: 403, message: "Wrong room secret" };
     }
     if (!storedSecretHash) {
       await this.state.storage.put("secretHash", secretHash);
       await this.state.storage.put("createdAt", Date.now());
     }
 
     await this.state.storage.put("lastSeenAt", Date.now());
     await this.state.storage.put(nonceKey, Date.now());
     await this.pruneNonces();
     await this.scheduleExpiry();
     this.lastTouchAt = Date.now();
 
     return { ok: true, authorID, deviceID };
   }
 
   async webSocketMessage(ws, message) {
     const data = typeof message === "string" ? message : message.slice(0);
     await this.touch();
     for (const peer of this.state.getWebSockets()) {
       if (peer === ws) continue;
       try {
         peer.send(data);
       } catch {
         // A peer caught mid-close throws here; skip it so the rest of the room
         // still receives this frame. The dead socket is reaped via
         // `webSocketClose` / runtime cleanup.
       }
     }
   }
 
   async webSocketClose(ws, code, reason) {
     await this.state.storage.put("lastSeenAt", Date.now());
     await this.scheduleExpiry();
     ws.close(code, reason);
   }
 
   async alarm() {
     await this.pruneNonces();
     const ttlMs = Number(this.env.ROOM_TTL_SECONDS || "600") * 1000;
     const sockets = this.state.getWebSockets();
     if (sockets.length > 0) {
       // Still connected — the room is alive regardless of how long since the
       // last broadcast (keepalive pings are auto-answered and don't advance
       // `lastSeenAt`). Re-arm from now, not the stale deadline, or a long idle
       // session would refire the alarm in a tight loop.
       await this.state.storage.setAlarm(Date.now() + ttlMs);
       return;
     }
 
     const lastSeenAt = await this.state.storage.get("lastSeenAt");
     const idleMs = Date.now() - (lastSeenAt || Date.now());
     if (idleMs > ttlMs) {
       await this.state.storage.deleteAll();
       return;
     }
     await this.scheduleExpiry();
   }
 
   async touch() {
     const now = Date.now();
     // Debounce the durable write + alarm reschedule off the message hot path.
     if (this.lastTouchAt && now - this.lastTouchAt < TOUCH_DEBOUNCE_MS) return;
     this.lastTouchAt = now;
     await this.state.storage.put("lastSeenAt", now);
     await this.scheduleExpiry();
   }
 
   async pruneNonces() {
     const maxAgeMs = Number(this.env.NONCE_TTL_SECONDS || "300") * 1000;
     const cutoff = Date.now() - maxAgeMs;
     const nonces = await this.state.storage.list({ prefix: "nonce:" });
     for (const [key, createdAt] of nonces) {
       if (createdAt < cutoff) {
         await this.state.storage.delete(key);
       }
     }
   }
 
   async scheduleExpiry() {
     const ttlMs = Number(this.env.ROOM_TTL_SECONDS || "600") * 1000;
     const lastSeenAt = (await this.state.storage.get("lastSeenAt")) || Date.now();
     await this.state.storage.setAlarm(lastSeenAt + ttlMs);
   }
 }
 
 export default {
   async fetch(request, env) {
     const url = new URL(request.url);
     if (url.pathname === "/health") {
       return new Response("ok");
     }
     const roomID = roomIDFromPath(url.pathname);
     if (!roomID) {
       return new Response("Not found", { status: 404 });
     }
     const id = env.ENGAGEMENT_ROOMS.idFromName(roomID);
     return env.ENGAGEMENT_ROOMS.get(id).fetch(request);
   }
 };
 
 function roomIDFromPath(pathname) {
   const match = pathname.match(/^\/rooms\/([^/]+)\/socket$/);
   return match ? match[1] : null;
 }
 
 async function hmacSHA256(secret, payload) {
   const key = await crypto.subtle.importKey(
     "raw",
     base64URLDecode(secret),
     { name: "HMAC", hash: "SHA-256" },
     false,
     ["sign"]
   );
   const signature = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(payload));
   return base64URLEncode(new Uint8Array(signature));
 }
 
 async function sha256(value) {
   const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(value));
   return base64URLEncode(new Uint8Array(digest));
 }
 
 function base64URLDecode(value) {
   const base64 = value.replace(/-/g, "+").replace(/_/g, "/").padEnd(Math.ceil(value.length / 4) * 4, "=");
   const binary = atob(base64);
   return Uint8Array.from(binary, (char) => char.charCodeAt(0));
 }
 
 function base64URLEncode(bytes) {
   let binary = "";
   for (const byte of bytes) {
     binary += String.fromCharCode(byte);
   }
   return btoa(binary).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/g, "");
 }
 
 function timingSafeEqual(a, b) {
   const left = new TextEncoder().encode(a);
   const right = new TextEncoder().encode(b);
   if (left.length !== right.length) return false;
   let diff = 0;
   for (let index = 0; index < left.length; index += 1) {
     diff |= left[index] ^ right[index];
   }
   return diff === 0;
 }