feat: email a recipient about messages that arrive while they are away

.dev.vars.example

@@ -14,7 +14,10 @@ EDGE_CHAT_JWT_SECRET=
 
 # ---- Optional (feature degrades gracefully if unset, not startup-blocking) ----
 # Webhook called on new/offline messages to UniBooks-BE; local default
-# http://127.0.0.1:8000/api/v1/messaging/webhook/edge-chat/
+# http://127.0.0.1:8000/api/v1/messaging/webhook/edge-chat/. Carries two kinds
+# of call: the inbox-preview mirror, and (event=offline_email) the request to
+# email a recipient who has no connection open to this Worker. Unset = no
+# preview sync and no email notification for messages missed while away.
 DJANGO_WEBHOOK_URL=
 # X-Webhook-Secret sent with DJANGO_WEBHOOK_URL calls, must exactly match
 # UniBooks-BE's EDGE_CHAT_WEBHOOK_SECRET. If unset, Django rejects the

src/ChatRoom.ts

@@ -1,5 +1,6 @@
 import { DurableObject } from "cloudflare:workers";
 import { isImageUrlAllowed } from "./imageUrlPolicy";
+import { isWebhookUrlAllowed } from "./webhookUrlPolicy";
 
 export interface Env {
   CHAT_ROOM: DurableObjectNamespace;
@@ -486,7 +487,7 @@ export class ChatRoom extends DurableObject<Env> {
         const roomId = await this.ctx.storage.get<string>("roomId");
 
         // Trigger Webhook for Django to update Inbox Preview and optionally send push notifications
-        if (roomId && this.env.DJANGO_WEBHOOK_URL && this.isWebhookUrlAllowed(this.env.DJANGO_WEBHOOK_URL)) {
+        if (roomId && isWebhookUrlAllowed(this.env.DJANGO_WEBHOOK_URL)) {
           const webhookContent = messageType === "image" ? IMAGE_PREVIEW_TOKEN : data.content;
           this.triggerOfflineWebhook(roomId, userId, webhookContent, !otherUsersConnected);
         }
@@ -562,6 +563,10 @@ export class ChatRoom extends DurableObject<Env> {
         body: JSON.stringify({
           room_id: roomId,
           sender_id: senderId,
+          // Which participant this hub belongs to. The hub needs it to name a
+          // recipient for the offline email even when that user has never
+          // connected to their hub (so it holds no userId of its own).
+          recipient_id: participantId,
           preview,
           timestamp,
           self: isSelf,
@@ -570,22 +575,6 @@ export class ChatRoom extends DurableObject<Env> {
     }));
   }
 
-  // DJANGO_WEBHOOK_URL is operator-set config, not attacker-controlled input,
-  // but this still guards against a typo'd/misconfigured value turning into
-  // an SSRF vector — only allow https, or http to localhost for local dev.
-  private isWebhookUrlAllowed(rawUrl: string): boolean {
-    try {
-      const parsed = new URL(rawUrl);
-      if (parsed.protocol === "https:") return true;
-      if (parsed.protocol === "http:" && (parsed.hostname === "localhost" || parsed.hostname === "127.0.0.1")) {
-        return true;
-      }
-      return false;
-    } catch {
-      return false;
-    }
-  }
-
   private deleteMessageForUser(ws: WebSocket, userId: string, messageId: string) {
     const exists = [...this.ctx.storage.sql.exec(
       `SELECT id FROM messages WHERE id = ?`, messageId

src/UserHub.ts

@@ -1,10 +1,35 @@
 import { DurableObject } from "cloudflare:workers";
+import { shouldSendOfflineEmail } from "./offlineEmailPolicy";
+import { isWebhookUrlAllowed } from "./webhookUrlPolicy";
 
 export interface Env {
   CHAT_ROOM: DurableObjectNamespace;
   USER_HUB: DurableObjectNamespace;
   EDGE_CHAT_JWT_SECRET: string;
   DJANGO_WEBHOOK_URL?: string;
+  // Shared secret Django requires on every webhook call (X-Webhook-Secret);
+  // without it the offline-email callback is rejected with 403 and the
+  // recipient silently never hears about the message.
+  DJANGO_WEBHOOK_SECRET?: string;
+}
+
+// Rooms this user has already been emailed about and has not opened since.
+// Persisted (not in-memory) because the whole point is to still be true
+// after this DO hibernates — which it will, since nobody is connected to it
+// while its user is away.
+const EMAIL_NOTIFIED_KEY = "emailNotifiedRooms";
+
+// Body of a /push call from ChatRoom.
+interface PushPayload {
+  room_id: string;
+  sender_id: string;
+  // The hub's own user, passed explicitly by ChatRoom (which knows the
+  // participant list from the sender's signed token) so the email callback
+  // can name a recipient even on a hub that has never been connected to.
+  recipient_id?: string;
+  preview: string;
+  timestamp: number;
+  self?: boolean;
 }
 
 // Minimum time between accepted WebSocket connections for the same user's
@@ -52,6 +77,9 @@ export class UserHub extends DurableObject<Env> {
 
       this.ctx.acceptWebSocket(server);
       server.serializeAttachment({ userId });
+      // Remembered for the offline-email callback, which runs when there is
+      // no socket left to read the id off.
+      await this.ctx.storage.put("userId", userId);
 
       const requestedProtocol = request.headers.get("Sec-WebSocket-Protocol");
       const responseHeaders: HeadersInit | undefined = requestedProtocol
@@ -77,13 +105,7 @@ export class UserHub extends DurableObject<Env> {
     // their own hub from increment (their own send doesn't make their inbox
     // unread), and we increment /every other/ participant's unread set.
     if (request.method === "POST" && url.pathname.endsWith("/push")) {
-      const data = await request.json() as {
-        room_id: string;
-        sender_id: string;
-        preview: string;
-        timestamp: number;
-        self?: boolean;
-      };
+      const data = await request.json() as PushPayload;
       const roomUpdateMsg = JSON.stringify({ type: "room_update", ...data });
       for (const ws of this.ctx.getWebSockets()) {
         try {
@@ -98,6 +120,10 @@ export class UserHub extends DurableObject<Env> {
       if (!data.self) {
         await this.markUnread(data.room_id);
       }
+      // Separate from the unread mark on purpose: unread is per *message*
+      // state the badge reflects, this is a one-shot-per-conversation email
+      // for a user who has no way to see that badge right now.
+      await this.maybeSendOfflineEmail(data);
       return new Response("ok");
     }
 
@@ -148,6 +174,14 @@ export class UserHub extends DurableObject<Env> {
   }
 
   private async markRead(roomId: string): Promise<void> {
+    // Opening the conversation is what re-arms the email for it: from here on
+    // the next message that arrives while this user is away may notify again.
+    // Done before the early return below, because a room can be marked
+    // notified and then read from another device that had already cleared the
+    // unread flag — the mark would otherwise stick forever and silence every
+    // future notification for that conversation.
+    await this.clearEmailNotified(roomId);
+
     const unread = new Set((await this.ctx.storage.get<string[]>("unread")) || []);
     if (!unread.delete(roomId)) {
       // already read, no need to re-broadcast
@@ -174,6 +208,68 @@ export class UserHub extends DurableObject<Env> {
     }
   }
 
+  // Emails the recipient about a message that arrived while they had the site
+  // closed — see shouldSendOfflineEmail for the two conditions. The Worker
+  // doesn't send mail itself: it calls the same Django webhook ChatRoom uses
+  // for the inbox mirror, with an `event` discriminator, and Django owns the
+  // templates, the address, and every "should this person be mailed at all"
+  // rule (deactivated account, conversation they deleted, ...).
+  private async maybeSendOfflineEmail(data: PushPayload): Promise<void> {
+    const notifiedRooms = (await this.ctx.storage.get<string[]>(EMAIL_NOTIFIED_KEY)) || [];
+    const decision = shouldSendOfflineEmail({
+      roomId: data.room_id,
+      isSelf: !!data.self,
+      // Counts every device/tab this user has open, hibernated ones included.
+      activeSocketCount: this.ctx.getWebSockets().length,
+      notifiedRooms,
+    });
+    if (!decision) return;
+
+    const webhookUrl = this.env.DJANGO_WEBHOOK_URL;
+    if (!isWebhookUrlAllowed(webhookUrl)) return;
+
+    const recipientId = data.recipient_id || (await this.ctx.storage.get<string>("userId"));
+    if (!recipientId) {
+      // Nothing to address the mail to. Leaving the room unmarked means a
+      // later message (by then carrying recipient_id) can still notify.
+      console.error("Offline email skipped: no recipient id for this hub");
+      return;
+    }
+
+    // Marked before the request goes out, not after: the send is a
+    // fire-and-forget waitUntil whose result nobody waits for, and one missed
+    // email (Django down) is a better failure than a burst of duplicates from
+    // every message that arrives while it is down. The mark clears the next
+    // time the recipient opens the conversation either way.
+    await this.ctx.storage.put(EMAIL_NOTIFIED_KEY, [...notifiedRooms, data.room_id]);
+
+    this.ctx.waitUntil(
+      fetch(webhookUrl!, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+          ...(this.env.DJANGO_WEBHOOK_SECRET ? { "X-Webhook-Secret": this.env.DJANGO_WEBHOOK_SECRET } : {}),
+        },
+        body: JSON.stringify({
+          // Distinguishes this from the inbox-preview mirror call ChatRoom
+          // makes to the same endpoint; Django dispatches on it.
+          event: "offline_email",
+          room_id: data.room_id,
+          recipient_id: recipientId,
+          sender_id: data.sender_id,
+          preview: data.preview,
+          timestamp: data.timestamp,
+        }),
+      }).catch(e => console.error("Offline email webhook failed", e))
+    );
+  }
+
+  private async clearEmailNotified(roomId: string): Promise<void> {
+    const notifiedRooms = (await this.ctx.storage.get<string[]>(EMAIL_NOTIFIED_KEY)) || [];
+    if (!notifiedRooms.includes(roomId)) return;
+    await this.ctx.storage.put(EMAIL_NOTIFIED_KEY, notifiedRooms.filter(id => id !== roomId));
+  }
+
   // Client -> server messages aren't part of this protocol (the hub is
   // notify-only); anything received is ignored rather than acted on.
   async webSocketMessage(_ws: WebSocket, _message: string | ArrayBuffer) {}

src/offlineEmailPolicy.test.ts

@@ -0,0 +1,59 @@
+import { describe, it, expect } from "vitest";
+import { shouldSendOfflineEmail, OfflineEmailInput } from "./offlineEmailPolicy";
+
+const ROOM = "3f0b2a1e-0000-4000-8000-000000000001";
+
+function input(overrides: Partial<OfflineEmailInput> = {}): OfflineEmailInput {
+  return {
+    roomId: ROOM,
+    isSelf: false,
+    activeSocketCount: 0,
+    notifiedRooms: [],
+    ...overrides,
+  };
+}
+
+describe("shouldSendOfflineEmail", () => {
+  it("notifies a recipient who has no connection to the chat service", () => {
+    expect(shouldSendOfflineEmail(input())).toBe(true);
+  });
+
+  it("stays silent while the recipient has the site open", () => {
+    // The unread badge is the notification in that case — the requirement is
+    // to mail people who are *away*, not people whose chat panel is closed.
+    expect(shouldSendOfflineEmail(input({ activeSocketCount: 1 }))).toBe(false);
+  });
+
+  it("stays silent when only one of several devices is connected", () => {
+    expect(shouldSendOfflineEmail(input({ activeSocketCount: 3 }))).toBe(false);
+  });
+
+  it("never notifies the sender about their own message", () => {
+    // ChatRoom pushes to the sender's hub too, so their inbox preview updates.
+    expect(shouldSendOfflineEmail(input({ isSelf: true }))).toBe(false);
+    // Even offline: sending a message from another device is not news.
+    expect(shouldSendOfflineEmail(input({ isSelf: true, activeSocketCount: 0 }))).toBe(false);
+  });
+
+  it("sends only once per conversation until it is opened", () => {
+    // The second and third message of a burst must not each become an email.
+    expect(shouldSendOfflineEmail(input({ notifiedRooms: [ROOM] }))).toBe(false);
+  });
+
+  it("still notifies about a different conversation", () => {
+    // The one-email rule is per conversation, not per user: a message from
+    // someone else, about another listing, is genuinely new information.
+    expect(shouldSendOfflineEmail(input({ notifiedRooms: ["some-other-room"] }))).toBe(true);
+  });
+
+  it("notifies again once the conversation has been opened", () => {
+    // Opening the room clears it from notifiedRooms (UserHub.markRead), which
+    // is the whole state this rule reads.
+    expect(shouldSendOfflineEmail(input({ notifiedRooms: [] }))).toBe(true);
+  });
+
+  it("does nothing without a room to notify about", () => {
+    expect(shouldSendOfflineEmail(input({ roomId: undefined }))).toBe(false);
+    expect(shouldSendOfflineEmail(input({ roomId: "" }))).toBe(false);
+  });
+});

src/offlineEmailPolicy.ts

@@ -0,0 +1,42 @@
+/**
+ * Whether a new message should send the recipient an email.
+ *
+ * Two rules, both from the product requirement:
+ *
+ *   1. Only when the recipient is not reachable in the app at all — no
+ *      WebSocket on their UserHub, which the frontend opens once per visit
+ *      for the whole site (not per conversation). Having a conversation
+ *      *closed* is not being away; having the site closed is.
+ *   2. At most one email per conversation per "unread streak": once a room
+ *      has been emailed about, later messages in that same room stay silent
+ *      until the recipient actually opens it (which clears the mark via
+ *      UserHub's mark-read path). Otherwise a chatty sender turns into a
+ *      dozen identical emails about one conversation.
+ *
+ * Kept out of UserHub so it can be tested as what it is: a pure decision over
+ * a handful of values, with no Durable Object or workerd runtime involved.
+ */
+
+export interface OfflineEmailInput {
+  /** Room the message landed in. */
+  roomId: string | undefined;
+  /** True when this hub belongs to the message's own sender. */
+  isSelf: boolean;
+  /** Live sockets on this user's hub, across all their devices/tabs. */
+  activeSocketCount: number;
+  /** Rooms already emailed about and not yet opened since. */
+  notifiedRooms: readonly string[];
+}
+
+export function shouldSendOfflineEmail(input: OfflineEmailInput): boolean {
+  const { roomId, isSelf, activeSocketCount, notifiedRooms } = input;
+  // Nothing addressable to link to, and nothing to dedupe on.
+  if (!roomId) return false;
+  // You are never notified about your own message.
+  if (isSelf) return false;
+  // The site is open somewhere — the in-app badge is the notification.
+  if (activeSocketCount > 0) return false;
+  // Already told them about this conversation; wait until they open it.
+  if (notifiedRooms.includes(roomId)) return false;
+  return true;
+}

src/webhookUrlPolicy.test.ts

@@ -0,0 +1,30 @@
+import { describe, it, expect } from "vitest";
+import { isWebhookUrlAllowed } from "./webhookUrlPolicy";
+
+describe("isWebhookUrlAllowed", () => {
+  it("allows https", () => {
+    expect(isWebhookUrlAllowed("https://api.unibooks.example/api/v1/messaging/webhook/edge-chat/")).toBe(true);
+  });
+
+  it("allows plain http only to the local dev server", () => {
+    expect(isWebhookUrlAllowed("http://localhost:8000/api/v1/messaging/webhook/edge-chat/")).toBe(true);
+    expect(isWebhookUrlAllowed("http://127.0.0.1:8000/api/v1/messaging/webhook/edge-chat/")).toBe(true);
+  });
+
+  it("rejects plain http to anywhere else", () => {
+    // A webhook body carries the room id and a message preview; over http it
+    // is readable by anyone on the path.
+    expect(isWebhookUrlAllowed("http://api.unibooks.example/webhook/")).toBe(false);
+    expect(isWebhookUrlAllowed("http://169.254.169.254/latest/meta-data/")).toBe(false);
+  });
+
+  it("rejects non-http schemes and unparseable values", () => {
+    expect(isWebhookUrlAllowed("file:///etc/passwd")).toBe(false);
+    expect(isWebhookUrlAllowed("not a url")).toBe(false);
+  });
+
+  it("treats an unset or empty variable as 'no webhook configured'", () => {
+    expect(isWebhookUrlAllowed(undefined)).toBe(false);
+    expect(isWebhookUrlAllowed("")).toBe(false);
+  });
+});

src/webhookUrlPolicy.ts

@@ -0,0 +1,27 @@
+/**
+ * Which URLs this Worker is willing to call back into Django on.
+ *
+ * DJANGO_WEBHOOK_URL is operator-set config, not attacker-controlled input,
+ * but a typo'd or stale value would otherwise turn every message into an
+ * outbound request to wherever that value happens to point — so only https,
+ * or http to localhost for `wrangler dev` against a local Django.
+ *
+ * Shared by ChatRoom (inbox-preview mirror) and UserHub (offline email
+ * notification), which both read the same variable: a guard that only one of
+ * them applied would be a guard the other quietly skipped.
+ */
+
+/** Hosts http:// is tolerated for, so local dev works without TLS. */
+const LOCAL_HTTP_HOSTNAMES = new Set(["localhost", "127.0.0.1"]);
+
+export function isWebhookUrlAllowed(rawUrl: string | undefined): boolean {
+  if (!rawUrl) return false;
+  let parsed: URL;
+  try {
+    parsed = new URL(rawUrl);
+  } catch {
+    return false;
+  }
+  if (parsed.protocol === "https:") return true;
+  return parsed.protocol === "http:" && LOCAL_HTTP_HOSTNAMES.has(parsed.hostname);
+}