From b2238dbd0cd0960a57051034515bc639119a8f7a Mon Sep 17 00:00:00 2001 From: olegberman Date: Sat, 22 Aug 2026 14:50:25 -0400 Subject: [PATCH 1/6] feat: add real-time Mail API events --- api/hqbase-mail-api-v1.openapi.json | 86 ++++++++++ ...hqbase-mail-api-v1.postman_collection.json | 25 +++ app/app.tsx | 18 ++ app/features/drafts/use-drafts.ts | 2 +- app/features/events/use-mail-events.ts | 154 ++++++++++++++++++ app/features/messages/use-mail-sync.ts | 2 +- package.json | 3 + scripts/generate-mail-api-artifacts.mjs | 11 +- scripts/hqbase/config.mjs | 4 + test/integration/worker/mail-api.test.ts | 74 +++++++++ test/unit/app/events/use-mail-events.test.tsx | 145 +++++++++++++++++ test/unit/cloudflare-workers.ts | 9 + test/unit/scripts/install.test.mjs | 6 + test/unit/scripts/mail-api-artifacts.test.mjs | 5 + .../worker/features/send/send-service.test.ts | 6 + test/unit/worker/index-notifications.test.ts | 17 +- vitest.config.ts | 5 +- worker-configuration.d.ts | 4 +- worker/auth/mail-api.ts | 36 +++- worker/features/drafts/routes.ts | 18 +- worker/features/events/durable-object.ts | 130 +++++++++++++++ worker/features/events/route.ts | 89 ++++++++++ worker/features/events/service.ts | 111 +++++++++++++ worker/features/events/types.ts | 19 +++ worker/features/mail-api/discovery.ts | 1 + worker/features/mailbox-access/routes.ts | 7 + worker/features/mailboxes/routes.ts | 9 + worker/features/mcp/draft-tools.ts | 10 ++ worker/features/mcp/mail-tools.ts | 14 +- .../features/messages/conversation-queries.ts | 18 +- .../features/messages/conversation-routes.ts | 21 ++- worker/features/messages/routes.ts | 13 +- worker/features/send/service.ts | 13 ++ worker/features/users/routes.ts | 4 + worker/index.ts | 14 ++ worker/jobs/consumer.ts | 19 ++- wrangler.jsonc | 9 + 37 files changed, 1101 insertions(+), 30 deletions(-) create mode 100644 app/features/events/use-mail-events.ts create mode 100644 test/unit/app/events/use-mail-events.test.tsx create mode 100644 test/unit/cloudflare-workers.ts create mode 100644 worker/features/events/durable-object.ts create mode 100644 worker/features/events/route.ts create mode 100644 worker/features/events/service.ts create mode 100644 worker/features/events/types.ts diff --git a/api/hqbase-mail-api-v1.openapi.json b/api/hqbase-mail-api-v1.openapi.json index 09bc1625..2d83f33d 100644 --- a/api/hqbase-mail-api-v1.openapi.json +++ b/api/hqbase-mail-api-v1.openapi.json @@ -34,6 +34,9 @@ }, { "name": "Sending" + }, + { + "name": "Events" } ], "paths": { @@ -117,6 +120,89 @@ "operationId": "listMailboxes" } }, + "/api/v1/events": { + "get": { + "summary": "Open change event WebSocket", + "description": "Opens an authenticated wake-only WebSocket. The server sends `changed` frames for messages, mailboxes, and permitted drafts. Frames contain no mail data or cursors. After a frame or reconnect, read the authoritative REST resources and change journals. OAuth tokens require `mail:read`; draft frames also require `mail:send`.", + "security": [ + { + "oauth2": ["mail:read"] + }, + { + "cookieSession": [] + } + ], + "parameters": [ + { + "name": "Upgrade", + "in": "header", + "required": true, + "description": "WebSocket upgrade request.", + "schema": { + "type": "string", + "const": "websocket" + } + } + ], + "responses": { + "101": { + "description": "WebSocket established. Text frames use `{\"type\":\"changed\",\"topic\":\"messages|drafts|mailboxes\"}`." + }, + "401": { + "description": "Missing or invalid authentication", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + }, + "403": { + "description": "Insufficient OAuth scope or invalid browser origin", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + }, + "405": { + "description": "Method not allowed", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + }, + "426": { + "description": "WebSocket upgrade required", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + }, + "503": { + "description": "Event connection unavailable", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + } + }, + "tags": ["Events"], + "operationId": "openMailEvents" + } + }, "/api/v1/changes": { "get": { "summary": "Read message changes", diff --git a/api/hqbase-mail-api-v1.postman_collection.json b/api/hqbase-mail-api-v1.postman_collection.json index c1c9496e..44547b36 100644 --- a/api/hqbase-mail-api-v1.postman_collection.json +++ b/api/hqbase-mail-api-v1.postman_collection.json @@ -150,6 +150,31 @@ } ] }, + { + "name": "Events", + "item": [ + { + "name": "Open change event WebSocket", + "request": { + "method": "GET", + "header": [ + { + "key": "Upgrade", + "value": "websocket", + "disabled": false, + "description": "WebSocket upgrade request." + } + ], + "url": { + "raw": "{{base_url}}/api/v1/events", + "variable": [] + }, + "description": "Opens an authenticated wake-only WebSocket. The server sends `changed` frames for messages, mailboxes, and permitted drafts. Frames contain no mail data or cursors. After a frame or reconnect, read the authoritative REST resources and change journals. OAuth tokens require `mail:read`; draft frames also require `mail:send`." + }, + "response": [] + } + ] + }, { "name": "Changes", "item": [ diff --git a/app/app.tsx b/app/app.tsx index 7979b658..75853336 100644 --- a/app/app.tsx +++ b/app/app.tsx @@ -14,6 +14,7 @@ import { import type { CurrentUser } from "@/features/auth/types"; import { DraftsPage } from "@/features/drafts/drafts-page"; import { useDrafts } from "@/features/drafts/use-drafts"; +import { useMailEvents } from "@/features/events/use-mail-events"; import { InboxPage } from "@/features/inbox/inbox-page"; import { listMailboxes } from "@/features/mailboxes/api"; import type { Mailbox } from "@/features/mailboxes/types"; @@ -114,6 +115,23 @@ export function App(): React.ReactElement { } }, [loadWorkspace]); + const refreshWorkspace = React.useCallback(async () => { + if (!currentUserId) return; + const currentUser = await getCurrentUser(); + setUser(currentUser); + if (!currentUser.passwordSetupRequired) await loadWorkspace(currentUser); + }, [currentUserId, loadWorkspace]); + const refreshRealtimeState = React.useCallback(async () => { + await Promise.allSettled([refreshWorkspace(), mailSync.refresh(), draftState.refresh()]); + }, [draftState.refresh, mailSync.refresh, refreshWorkspace]); + + useMailEvents(currentUserId, { + onDrafts: draftState.refresh, + onMailboxes: refreshRealtimeState, + onMessages: mailSync.refresh, + onReconnect: refreshRealtimeState + }); + React.useEffect(() => { if (publicAuthenticationPath) return; void reload(); diff --git a/app/features/drafts/use-drafts.ts b/app/features/drafts/use-drafts.ts index 8f8003fe..9680c9ed 100644 --- a/app/features/drafts/use-drafts.ts +++ b/app/features/drafts/use-drafts.ts @@ -3,7 +3,7 @@ import * as React from "react"; import { listDrafts } from "./api"; import type { Draft } from "./types"; -const refreshIntervalMs = 10_000; +const refreshIntervalMs = 60_000; export function useDrafts(userId: string | null): { drafts: Draft[]; diff --git a/app/features/events/use-mail-events.ts b/app/features/events/use-mail-events.ts new file mode 100644 index 00000000..7f216e0d --- /dev/null +++ b/app/features/events/use-mail-events.ts @@ -0,0 +1,154 @@ +import * as React from "react"; + +const reconnectBaseDelayMs = 1_000; +const reconnectMaxDelayMs = 30_000; + +type MailEventTopic = "drafts" | "mailboxes" | "messages"; + +type MailEventHandlers = { + onDrafts: () => unknown; + onMailboxes: () => unknown; + onMessages: () => unknown; + onReconnect: () => unknown; +}; + +type MailEvent = { + topic: MailEventTopic; + type: "changed"; +}; + +export function useMailEvents(userId: string | null, handlers: MailEventHandlers): void { + const currentHandlers = React.useRef(handlers); + currentHandlers.current = handlers; + + React.useEffect(() => { + if (!userId) return; + + let active = true; + let attempt = 0; + let reconnectTimer: number | null = null; + let socket: WebSocket | null = null; + const pendingTopics = new Set(); + let flushScheduled = false; + + const invoke = (callback: () => unknown): void => { + void Promise.resolve() + .then(callback) + .catch(() => undefined); + }; + const flush = (): void => { + flushScheduled = false; + if (!active) return; + for (const topic of pendingTopics) { + const callback = + topic === "messages" + ? currentHandlers.current.onMessages + : topic === "drafts" + ? currentHandlers.current.onDrafts + : currentHandlers.current.onMailboxes; + invoke(callback); + } + pendingTopics.clear(); + }; + const queueTopic = (topic: MailEventTopic): void => { + pendingTopics.add(topic); + if (flushScheduled) return; + flushScheduled = true; + queueMicrotask(flush); + }; + const canConnect = (): boolean => + active && document.visibilityState === "visible" && navigator.onLine !== false; + const closeSocket = (): void => { + const current = socket; + socket = null; + if (current && current.readyState < WebSocket.CLOSING) { + current.close(1000, "Connection paused."); + } + }; + + const connect = (): void => { + if (!canConnect() || socket !== null) return; + + const url = new URL("/api/v1/events", window.location.href); + url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; + const next = new WebSocket(url); + socket = next; + next.addEventListener("open", () => { + if (!active || socket !== next) return; + attempt = 0; + invoke(currentHandlers.current.onReconnect); + }); + next.addEventListener("message", (event) => { + const parsed = parseMailEvent(event.data); + if (parsed) queueTopic(parsed.topic); + }); + next.addEventListener("error", () => { + if (socket === next && next.readyState < WebSocket.CLOSING) next.close(); + }); + next.addEventListener("close", () => { + if (socket !== next) return; + socket = null; + if (!canConnect() || reconnectTimer !== null) return; + const delay = Math.min( + reconnectMaxDelayMs, + reconnectBaseDelayMs * 2 ** Math.min(attempt, 5) + ); + attempt += 1; + reconnectTimer = window.setTimeout(() => { + reconnectTimer = null; + connect(); + }, delay + reconnectJitterMs()); + }); + }; + const resume = (): void => { + if (reconnectTimer !== null) { + window.clearTimeout(reconnectTimer); + reconnectTimer = null; + } + connect(); + }; + const handleVisibilityChange = (): void => { + if (document.visibilityState === "visible") resume(); + else closeSocket(); + }; + const handleOnline = (): void => resume(); + const handleOffline = (): void => closeSocket(); + + document.addEventListener("visibilitychange", handleVisibilityChange); + window.addEventListener("online", handleOnline); + window.addEventListener("offline", handleOffline); + connect(); + return () => { + active = false; + if (reconnectTimer !== null) window.clearTimeout(reconnectTimer); + document.removeEventListener("visibilitychange", handleVisibilityChange); + window.removeEventListener("online", handleOnline); + window.removeEventListener("offline", handleOffline); + closeSocket(); + }; + }, [userId]); +} + +function parseMailEvent(value: unknown): MailEvent | null { + if (typeof value !== "string") return null; + try { + const parsed: unknown = JSON.parse(value); + if (!parsed || typeof parsed !== "object") return null; + const candidate = parsed as Partial; + if ( + candidate.type !== "changed" || + !["drafts", "mailboxes", "messages"].includes(candidate.topic ?? "") + ) { + return null; + } + return candidate as MailEvent; + } catch { + return null; + } +} + +function reconnectJitterMs(): number { + const values = new Uint32Array(1); + crypto.getRandomValues(values); + return (values[0] ?? 0) % 500; +} diff --git a/app/features/messages/use-mail-sync.ts b/app/features/messages/use-mail-sync.ts index 2d8cca2a..1da57150 100644 --- a/app/features/messages/use-mail-sync.ts +++ b/app/features/messages/use-mail-sync.ts @@ -8,7 +8,7 @@ import type { FolderId } from "@/lib/routes"; import { listConversations } from "./api"; import type { ConversationAction, ConversationSummary } from "./types"; -const refreshIntervalMs = 10_000; +const refreshIntervalMs = 60_000; type MailSyncOptions = { activeFolder: FolderId; diff --git a/package.json b/package.json index ddee6a90..2a8019a4 100644 --- a/package.json +++ b/package.json @@ -54,6 +54,9 @@ "MAIL_OBJECTS": { "description": "Create a new R2 bucket for this HQBase workspace." }, + "MAIL_EVENTS": { + "description": "Hibernating Durable Object used for access-scoped mail synchronization wake-ups." + }, "HQBASE_JOBS": { "description": "Queue used for bounded maintenance and integrity jobs with a dead-letter queue." }, diff --git a/scripts/generate-mail-api-artifacts.mjs b/scripts/generate-mail-api-artifacts.mjs index 65235660..68ad67e1 100644 --- a/scripts/generate-mail-api-artifacts.mjs +++ b/scripts/generate-mail-api-artifacts.mjs @@ -83,6 +83,7 @@ function validateOpenApi(document) { "/api/v1/mailboxes", "/api/v1/messages", "/api/v1/changes", + "/api/v1/events", "/api/v1/conversations", "/api/v1/drafts", "/api/v1/drafts/changes", @@ -176,9 +177,17 @@ function postmanRequest(route, method, operation) { disabled: parameter.name !== "folder", description: parameter.description })); + const headers = (operation.parameters ?? []) + .filter((parameter) => parameter.in === "header") + .map((parameter) => ({ + key: parameter.name, + value: parameter.schema?.const ?? "", + disabled: parameter.required !== true, + description: parameter.description + })); const request = { method: method.toUpperCase(), - header: [], + header: headers, url: { raw: `{{base_url}}${postmanRoute}`, variable: variables.map((name) => ({ key: name, value: `{{${name}}}` })), diff --git a/scripts/hqbase/config.mjs b/scripts/hqbase/config.mjs index 54ece3a7..fa6ae9f5 100644 --- a/scripts/hqbase/config.mjs +++ b/scripts/hqbase/config.mjs @@ -48,6 +48,10 @@ export function createWranglerConfig(manifest) { invocation_logs: false } }, + durable_objects: { + bindings: [{ name: "MAIL_EVENTS", class_name: "MailEvents" }] + }, + migrations: [{ tag: "mail-events-v1", new_sqlite_classes: ["MailEvents"] }], secrets: { required: ["BETTER_AUTH_SECRET"] }, diff --git a/test/integration/worker/mail-api.test.ts b/test/integration/worker/mail-api.test.ts index 539e537b..8358c8b9 100644 --- a/test/integration/worker/mail-api.test.ts +++ b/test/integration/worker/mail-api.test.ts @@ -329,6 +329,47 @@ describe("HQBase Mail API v1", () => { expect(legacyBearer.status).toBe(401); }); + it("opens an access-scoped wake-only WebSocket", async () => { + const missingUpgrade = await apiFetch("/api/v1/events", readToken); + expect(missingUpgrade.status).toBe(426); + expect(missingUpgrade.headers.get("upgrade")).toBe("websocket"); + + const insufficientScope = await apiFetch("/api/v1/events", writeToken, { + headers: { upgrade: "websocket" } + }); + expect(insufficientScope.status).toBe(403); + expect(insufficientScope.headers.get("www-authenticate")).toContain('scope="mail:read"'); + + const invalidOrigin = await SELF.fetch(`${origin}/api/v1/events`, { + headers: { cookie, origin: "https://other.example", upgrade: "websocket" } + }); + expect(invalidOrigin.status).toBe(403); + + const messageSocket = await openEventSocket({ + authorization: `Bearer ${readToken}`, + upgrade: "websocket" + }); + const messageFrame = nextSocketFrame(messageSocket); + await env.MAIL_EVENTS.getByName("workspace").publish({ + topic: "messages", + userIds: [userId] + }); + await expect(messageFrame).resolves.toEqual({ type: "changed", topic: "messages" }); + messageSocket.close(1000, "Test complete."); + + const draftSocket = await openEventSocket({ + authorization: `Bearer ${fullToken}`, + upgrade: "websocket" + }); + const draftFrame = nextSocketFrame(draftSocket); + await env.MAIL_EVENTS.getByName("workspace").publish({ + topic: "drafts", + userIds: [userId] + }); + await expect(draftFrame).resolves.toEqual({ type: "changed", topic: "drafts" }); + draftSocket.close(1000, "Test complete."); + }); + it("reads mail with an audience-bound bearer token without exposing storage keys", async () => { const list = await apiFetch("/api/v1/messages", readToken); expect(list.status, await list.clone().text()).toBe(200); @@ -363,11 +404,18 @@ describe("HQBase Mail API v1", () => { }); it("unarchives and restores mail through the versioned action route", async () => { + const socket = await openEventSocket({ + authorization: `Bearer ${readToken}`, + upgrade: "websocket" + }); + const changed = nextSocketFrame(socket); const archived = await apiFetch("/api/v1/messages/msg_api/archive", writeToken, { method: "POST" }); expect(archived.status, await archived.clone().text()).toBe(200); await expect(archived.json()).resolves.toMatchObject({ folder: "archived" }); + await expect(changed).resolves.toEqual({ type: "changed", topic: "messages" }); + socket.close(1000, "Test complete."); const unarchived = await apiFetch("/api/v1/messages/msg_api/unarchive", writeToken, { method: "POST" @@ -882,3 +930,29 @@ function extractSessionCookie(response: Response): string { if (!match?.[1]) throw new Error("Session cookie was not returned."); return match[1]; } + +async function openEventSocket(headers: HeadersInit): Promise { + const response = await SELF.fetch(`${origin}/api/v1/events`, { headers }); + if (response.status !== 101) { + throw new Error(`WebSocket upgrade failed (${response.status}): ${await response.text()}`); + } + if (!response.webSocket) throw new Error("WebSocket upgrade did not return a socket."); + response.webSocket.accept(); + return response.webSocket; +} + +function nextSocketFrame(socket: WebSocket): Promise { + return new Promise((resolve, reject) => { + socket.addEventListener( + "message", + (event) => { + try { + resolve(JSON.parse(String(event.data))); + } catch (error) { + reject(error); + } + }, + { once: true } + ); + }); +} diff --git a/test/unit/app/events/use-mail-events.test.tsx b/test/unit/app/events/use-mail-events.test.tsx new file mode 100644 index 00000000..75cb1b17 --- /dev/null +++ b/test/unit/app/events/use-mail-events.test.tsx @@ -0,0 +1,145 @@ +// @vitest-environment happy-dom +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { useMailEvents } from "@/features/events/use-mail-events"; +import { flushHookEffects, renderHook } from "../render-hook"; + +class FakeWebSocket extends EventTarget { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSING = 2; + static readonly CLOSED = 3; + static readonly instances: FakeWebSocket[] = []; + + readonly url: string; + readyState = FakeWebSocket.CONNECTING; + + constructor(url: string | URL) { + super(); + this.url = url.toString(); + FakeWebSocket.instances.push(this); + } + + close = vi.fn(() => { + this.readyState = FakeWebSocket.CLOSED; + this.dispatchEvent(new CloseEvent("close")); + }); + + open(): void { + this.readyState = FakeWebSocket.OPEN; + this.dispatchEvent(new Event("open")); + } + + message(data: unknown): void { + this.dispatchEvent(new MessageEvent("message", { data })); + } +} + +function handlers() { + return { + onDrafts: vi.fn(), + onMailboxes: vi.fn(), + onMessages: vi.fn(), + onReconnect: vi.fn() + }; +} + +describe("useMailEvents", () => { + beforeEach(() => { + FakeWebSocket.instances.length = 0; + Object.defineProperty(document, "visibilityState", { + configurable: true, + value: "visible" + }); + Object.defineProperty(navigator, "onLine", { configurable: true, value: true }); + vi.stubGlobal("WebSocket", FakeWebSocket); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + }); + + it("opens one same-origin socket and coalesces wake events by topic", async () => { + const callbacks = handlers(); + const hook = await renderHook(({ userId }) => useMailEvents(userId, callbacks), { + userId: "user-1" + }); + const socket = FakeWebSocket.instances[0]; + + expect(socket?.url).toBe("ws://localhost:3000/api/v1/events"); + await flushHookEffects(() => socket?.open()); + expect(callbacks.onReconnect).toHaveBeenCalledOnce(); + + await flushHookEffects(() => { + socket?.message('{"type":"changed","topic":"messages"}'); + socket?.message('{"type":"changed","topic":"messages"}'); + socket?.message('{"type":"changed","topic":"drafts"}'); + socket?.message('{"type":"unknown","topic":"mailboxes"}'); + socket?.message("not-json"); + }); + + expect(callbacks.onMessages).toHaveBeenCalledOnce(); + expect(callbacks.onDrafts).toHaveBeenCalledOnce(); + expect(callbacks.onMailboxes).not.toHaveBeenCalled(); + await hook.unmount(); + }); + + it("pauses while hidden and reconnects when the page becomes visible", async () => { + const callbacks = handlers(); + const hook = await renderHook(({ userId }) => useMailEvents(userId, callbacks), { + userId: "user-1" + }); + const first = FakeWebSocket.instances[0]; + await flushHookEffects(() => first?.open()); + + Object.defineProperty(document, "visibilityState", { + configurable: true, + value: "hidden" + }); + await flushHookEffects(() => document.dispatchEvent(new Event("visibilitychange"))); + expect(first?.close).toHaveBeenCalledOnce(); + + Object.defineProperty(document, "visibilityState", { + configurable: true, + value: "visible" + }); + await flushHookEffects(() => document.dispatchEvent(new Event("visibilitychange"))); + expect(FakeWebSocket.instances).toHaveLength(2); + await hook.unmount(); + }); + + it("uses the latest handlers without reopening the socket", async () => { + const firstHandlers = handlers(); + const secondHandlers = handlers(); + const hook = await renderHook(({ callbacks }) => useMailEvents("user-1", callbacks), { + callbacks: firstHandlers + }); + const socket = FakeWebSocket.instances[0]; + await hook.rerender({ callbacks: secondHandlers }); + await flushHookEffects(() => socket?.message('{"type":"changed","topic":"mailboxes"}')); + + expect(FakeWebSocket.instances).toHaveLength(1); + expect(firstHandlers.onMailboxes).not.toHaveBeenCalled(); + expect(secondHandlers.onMailboxes).toHaveBeenCalledOnce(); + await hook.unmount(); + }); + + it("backs off after an unexpected close", async () => { + vi.useFakeTimers(); + vi.spyOn(crypto, "getRandomValues").mockImplementation((values) => { + if (values instanceof Uint32Array) values[0] = 0; + return values; + }); + const hook = await renderHook(() => useMailEvents("user-1", handlers()), undefined); + const first = FakeWebSocket.instances[0]; + await flushHookEffects(() => first?.close()); + + await flushHookEffects(() => vi.advanceTimersByTime(999)); + expect(FakeWebSocket.instances).toHaveLength(1); + await flushHookEffects(() => vi.advanceTimersByTime(1)); + expect(FakeWebSocket.instances).toHaveLength(2); + await hook.unmount(); + }); +}); diff --git a/test/unit/cloudflare-workers.ts b/test/unit/cloudflare-workers.ts new file mode 100644 index 00000000..d8cfca1f --- /dev/null +++ b/test/unit/cloudflare-workers.ts @@ -0,0 +1,9 @@ +export abstract class DurableObject { + protected readonly ctx: DurableObjectState; + protected readonly env: Env; + + constructor(ctx: DurableObjectState, env: Env) { + this.ctx = ctx; + this.env = env; + } +} diff --git a/test/unit/scripts/install.test.mjs b/test/unit/scripts/install.test.mjs index 54f14198..d2db53ce 100644 --- a/test/unit/scripts/install.test.mjs +++ b/test/unit/scripts/install.test.mjs @@ -87,6 +87,12 @@ describe("HQBase installation resources", () => { } ] }); + expect(config.durable_objects).toEqual({ + bindings: [{ name: "MAIL_EVENTS", class_name: "MailEvents" }] + }); + expect(config.migrations).toEqual([ + { tag: "mail-events-v1", new_sqlite_classes: ["MailEvents"] } + ]); }); it("records customer-managed OAuth as non-secret deployment configuration", () => { diff --git a/test/unit/scripts/mail-api-artifacts.test.mjs b/test/unit/scripts/mail-api-artifacts.test.mjs index 129ed087..8e67ea13 100644 --- a/test/unit/scripts/mail-api-artifacts.test.mjs +++ b/test/unit/scripts/mail-api-artifacts.test.mjs @@ -15,6 +15,10 @@ describe("Mail API public artifacts", () => { expect(openApi.paths["/api/v1/changes"].get.security).toContainEqual({ oauth2: ["mail:read"] }); + expect(openApi.paths["/api/v1/events"].get.security).toContainEqual({ + oauth2: ["mail:read"] + }); + expect(openApi.paths["/api/v1/events"].get.responses["101"]).toBeDefined(); expect(openApi.paths["/api/v1/drafts/changes"].get.security).toContainEqual({ oauth2: ["mail:send"] }); @@ -55,6 +59,7 @@ describe("Mail API public artifacts", () => { const serialized = JSON.stringify(postman); expect(serialized).toContain("/.well-known/oauth-protected-resource/api/v1"); expect(serialized).toContain("/api/auth/oauth2/register"); + expect(serialized).toContain("Open change event WebSocket"); const oauthSetup = postman.item.find((folder) => folder.name === "OAuth setup"); const registrationRequest = oauthSetup.item.find( (request) => request.name === "Register public client" diff --git a/test/unit/worker/features/send/send-service.test.ts b/test/unit/worker/features/send/send-service.test.ts index 4da7d814..14f9ed62 100644 --- a/test/unit/worker/features/send/send-service.test.ts +++ b/test/unit/worker/features/send/send-service.test.ts @@ -11,6 +11,11 @@ vi.mock("@worker/features/mailboxes/queries", () => ({ vi.mock("@worker/features/mailboxes/address-queries", () => ({ findAddressIdentity: vi.fn().mockResolvedValue(null) })); +vi.mock("@worker/features/events/service", () => ({ + ignoreMailEventFailure: (promise: Promise) => promise.catch(() => undefined), + publishMessageMailEvent: () => Promise.resolve(), + publishUserMailEvent: () => Promise.resolve() +})); vi.mock("@worker/features/messages/queries", () => ({ getMessageDetail: vi.fn(), @@ -76,6 +81,7 @@ describe("send service", () => { HQBASE_RELEASE_MANIFEST_URL: "https://github.com/HQBase/hqbase/releases/latest/download/stable.json", HQBASE_WORKER_NAME: "hqbase", + MAIL_EVENTS: {} as WorkerEnv["MAIL_EVENTS"], MAIL_OBJECTS: { get, put } as unknown as R2Bucket, MAIL_SENDER: { send } as unknown as SendEmail, HQBASE_JOBS: {} as Queue diff --git a/test/unit/worker/index-notifications.test.ts b/test/unit/worker/index-notifications.test.ts index 16cb30aa..f2507aa1 100644 --- a/test/unit/worker/index-notifications.test.ts +++ b/test/unit/worker/index-notifications.test.ts @@ -2,13 +2,18 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ handleInboundEmail: vi.fn(), - notifyInboundMessage: vi.fn() + notifyInboundMessage: vi.fn(), + publishMessageMailEvent: vi.fn() })); vi.mock("@worker/email/inbound", () => ({ handleInboundEmail: mocks.handleInboundEmail })); vi.mock("@worker/features/notifications/delivery", () => ({ notifyInboundMessage: mocks.notifyInboundMessage })); +vi.mock("@worker/features/events/service", () => ({ + ignoreMailEventFailure: (promise: Promise) => promise.catch(() => undefined), + publishMessageMailEvent: mocks.publishMessageMailEvent +})); import worker from "@worker/index"; import type { WorkerEnv } from "@worker/lib/env"; @@ -23,6 +28,7 @@ describe("inbound notification scheduling", () => { beforeEach(() => { vi.clearAllMocks(); mocks.notifyInboundMessage.mockResolvedValue(undefined); + mocks.publishMessageMailEvent.mockResolvedValue(undefined); }); it("does not schedule push for a duplicate inbound message", async () => { @@ -54,7 +60,12 @@ describe("inbound notification scheduling", () => { ); expect(mocks.notifyInboundMessage).toHaveBeenCalledWith({}, storedMessage, true); - expect(waitUntil).toHaveBeenCalledOnce(); - await expect(waitUntil.mock.calls[0]?.[0]).resolves.toBeUndefined(); + expect(mocks.publishMessageMailEvent).toHaveBeenCalledWith({}, [ + { isUnassigned: true, mailboxId: "mbx_1" } + ]); + expect(waitUntil).toHaveBeenCalledTimes(2); + await Promise.all( + waitUntil.mock.calls.map(([promise]) => expect(promise).resolves.toBeUndefined()) + ); }); }); diff --git a/vitest.config.ts b/vitest.config.ts index 7405f303..66ff0676 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -8,7 +8,10 @@ export default defineConfig({ resolve: { alias: { "@": fileURLToPath(new URL("./app", import.meta.url)), - "@worker": fileURLToPath(new URL("./worker", import.meta.url)) + "@worker": fileURLToPath(new URL("./worker", import.meta.url)), + "cloudflare:workers": fileURLToPath( + new URL("./test/unit/cloudflare-workers.ts", import.meta.url) + ) } }, test: { diff --git a/worker-configuration.d.ts b/worker-configuration.d.ts index 90e5c41f..75a0ff8f 100644 --- a/worker-configuration.d.ts +++ b/worker-configuration.d.ts @@ -1,5 +1,5 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: 24b30229e48a27be5a956f19fccf9d08) +// Generated by Wrangler by running `wrangler types` (hash: 393d712f7f958bfd1757b2ac2ae4d932) // Runtime types generated with workerd@1.20260722.1 2026-07-28 nodejs_compat interface __BaseEnv_Env { MAIL_OBJECTS: R2Bucket; @@ -8,10 +8,12 @@ interface __BaseEnv_Env { HQBASE_JOBS: Queue; ASSETS: Fetcher; BETTER_AUTH_SECRET: string; + MAIL_EVENTS: DurableObjectNamespace; } declare namespace Cloudflare { interface GlobalProps { mainModule: typeof import("./worker/index"); + durableNamespaces: "MailEvents"; } interface Env extends __BaseEnv_Env {} } diff --git a/worker/auth/mail-api.ts b/worker/auth/mail-api.ts index 99ec1bc8..0ade5a06 100644 --- a/worker/auth/mail-api.ts +++ b/worker/auth/mail-api.ts @@ -10,6 +10,12 @@ export type MailApiScope = (typeof mailApiScopes)[number]; export const mailApiMetadataPath = "/.well-known/oauth-protected-resource/api/v1"; const agentSkillPath = "/skills/hqbase-mail/SKILL.md"; +export type MailApiPrincipal = { + auth: AuthContext; + authentication: "bearer" | "session"; + scopes: ReadonlySet; +}; + export class MailApiAuthError extends AppError { readonly authError: "invalid_token" | "insufficient_scope" | null; readonly requiredScope: MailApiScope; @@ -33,13 +39,29 @@ export async function requireMailApiContext( request: Request, requiredScope: MailApiScope ): Promise { + return (await requireMailApiPrincipal(env, request, requiredScope)).auth; +} + +export async function requireMailApiPrincipal( + env: WorkerEnv, + request: Request, + requiredScope: MailApiScope +): Promise { if (!isVersionedMailApiRequest(request)) { - return requireAuthContext(env, request); + return { + auth: await requireAuthContext(env, request), + authentication: "session", + scopes: new Set(mailApiScopes) + }; } if (!request.headers.has("authorization")) { try { - return await requireAuthContext(env, request); + return { + auth: await requireAuthContext(env, request), + authentication: "session", + scopes: new Set(mailApiScopes) + }; } catch (error) { if (error instanceof AppError && error.status === 401) { throw new MailApiAuthError( @@ -68,7 +90,15 @@ export async function requireMailApiContext( "insufficient_scope" ); } - return { session: principal.session, user: principal.user }; + return { + auth: { session: principal.session, user: principal.user }, + authentication: "bearer", + scopes: new Set( + [...principal.scopes].filter((scope): scope is MailApiScope => + mailApiScopes.includes(scope as MailApiScope) + ) + ) + }; } catch (error) { if (error instanceof MailApiAuthError) throw error; if (error instanceof OAuthBearerError) { diff --git a/worker/features/drafts/routes.ts b/worker/features/drafts/routes.ts index 090ee3a5..52389182 100644 --- a/worker/features/drafts/routes.ts +++ b/worker/features/drafts/routes.ts @@ -1,9 +1,10 @@ -import { Hono } from "hono"; +import { type Context, Hono } from "hono"; import { requireMailApiContext } from "../../auth/mail-api"; import type { HonoApp } from "../../lib/env"; import { AppError } from "../../lib/errors"; import { readJson } from "../../lib/json"; import { parseWith } from "../../lib/validation"; +import { ignoreMailEventFailure, publishUserMailEvent } from "../events/service"; import { getAccessibleDraft, listAccessibleDraftPage, requireDraftAccess } from "./access"; import { defaultDraftChangeLimit, listDraftChanges, maxDraftChangeLimit } from "./change-queries"; import { defaultDraftLimit, maxDraftLimit } from "./list-queries"; @@ -49,20 +50,25 @@ draftRoutes.post("/", async (c) => { const auth = await requireMailApiContext(c.env, c.req.raw, "mail:send"); const input = parseWith(draftSchema, await readJson(c.req.raw)); await requireDraftAccess(c.env, principal(auth), input); - return c.json(await saveDraft(c.env.DB, auth.user.id, input), 201); + const draft = await saveDraft(c.env.DB, auth.user.id, input); + scheduleDraftEvent(c, auth.user.id); + return c.json(draft, 201); }); draftRoutes.patch("/:id", async (c) => { const auth = await requireMailApiContext(c.env, c.req.raw, "mail:send"); await getAccessibleDraft(c.env, principal(auth), c.req.param("id")); const input = parseWith(draftSchema, await readJson(c.req.raw)); await requireDraftAccess(c.env, principal(auth), input); - return c.json(await saveDraft(c.env.DB, auth.user.id, { ...input, id: c.req.param("id") })); + const draft = await saveDraft(c.env.DB, auth.user.id, { ...input, id: c.req.param("id") }); + scheduleDraftEvent(c, auth.user.id); + return c.json(draft); }); draftRoutes.delete("/:id", async (c) => { const auth = await requireMailApiContext(c.env, c.req.raw, "mail:send"); await getAccessibleDraft(c.env, principal(auth), c.req.param("id")); if (!(await deleteDraft(c.env.DB, c.env.MAIL_OBJECTS, auth.user.id, c.req.param("id")))) throw new AppError("DRAFT_NOT_FOUND", "Draft not found.", 404); + scheduleDraftEvent(c, auth.user.id); return c.body(null, 204); }); draftRoutes.post("/:id/attachments", async (c) => { @@ -75,6 +81,7 @@ draftRoutes.post("/:id/attachments", async (c) => { await c.env.MAIL_OBJECTS.put(added.r2Key, file.stream(), { httpMetadata: { contentType: added.attachment.contentType } }); + scheduleDraftEvent(c, auth.user.id); return c.json(added.attachment, 201); }); draftRoutes.delete("/:draftId/attachments/:id", async (c) => { @@ -90,9 +97,14 @@ draftRoutes.delete("/:draftId/attachments/:id", async (c) => { )) ) throw new AppError("ATTACHMENT_NOT_FOUND", "Attachment not found.", 404); + scheduleDraftEvent(c, auth.user.id); return c.body(null, 204); }); +function scheduleDraftEvent(c: Context, userId: string): void { + c.executionCtx.waitUntil(ignoreMailEventFailure(publishUserMailEvent(c.env, userId, "drafts"))); +} + function principal(auth: Awaited>) { return { role: auth.user.role, userId: auth.user.id }; } diff --git a/worker/features/events/durable-object.ts b/worker/features/events/durable-object.ts new file mode 100644 index 00000000..80384caa --- /dev/null +++ b/worker/features/events/durable-object.ts @@ -0,0 +1,130 @@ +import { DurableObject } from "cloudflare:workers"; + +import type { WorkerEnv } from "../../lib/env"; + +import { + type MailEventConnection, + type MailEventPublish, + type MailEventTopic, + mailEventTopics +} from "./types"; + +const connectionLifetimeMs = 10 * 60 * 1000; +const maxConnectionsPerUser = 3; +const maxWorkspaceConnections = 1_000; +const internalUserHeader = "x-hqbase-event-user"; +const internalTopicsHeader = "x-hqbase-event-topics"; +const internalRequestIdHeader = "x-hqbase-event-request-id"; + +export class MailEvents extends DurableObject { + override async fetch(request: Request): Promise { + if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") { + return new Response("WebSocket upgrade required.", { status: 426 }); + } + + const userId = request.headers.get(internalUserHeader); + const topics = parseTopics(request.headers.get(internalTopicsHeader)); + if (!userId || topics.length === 0) { + return new Response("Authenticated event context required.", { status: 403 }); + } + if (this.ctx.getWebSockets().length >= maxWorkspaceConnections) { + return new Response("Event connection capacity reached.", { status: 503 }); + } + + const userTag = `user:${userId}`; + const existing = this.ctx.getWebSockets(userTag); + if (existing.length >= maxConnectionsPerUser) { + existing + .sort( + (left, right) => + (readConnection(left)?.expiresAt ?? 0) - (readConnection(right)?.expiresAt ?? 0) + )[0] + ?.close(1008, "A newer connection replaced this one."); + } + + const pair = new WebSocketPair(); + const client = pair[0]; + const server = pair[1]; + const connection: MailEventConnection = { + expiresAt: Date.now() + connectionLifetimeMs, + topics, + userId + }; + this.ctx.acceptWebSocket(server, [...topics, userTag]); + server.serializeAttachment(connection); + + return new Response(null, { + status: 101, + webSocket: client, + headers: { + "cache-control": "no-store", + "referrer-policy": "no-referrer", + "x-content-type-options": "nosniff", + "x-request-id": request.headers.get(internalRequestIdHeader) ?? crypto.randomUUID() + } + }); + } + + async publish(input: MailEventPublish): Promise { + if (!mailEventTopics.includes(input.topic) || input.userIds.length === 0) return; + + const recipients = new Set(input.userIds); + const payload = JSON.stringify({ type: "changed", topic: input.topic }); + const now = Date.now(); + for (const socket of this.ctx.getWebSockets(input.topic)) { + const connection = readConnection(socket); + if (!connection || !recipients.has(connection.userId)) continue; + if (connection.expiresAt <= now) { + socket.close(1008, "Reconnect to renew authentication."); + continue; + } + try { + socket.send(payload); + } catch { + socket.close(1011, "Event delivery failed."); + } + } + } + + override async webSocketMessage(socket: WebSocket): Promise { + socket.close(1008, "Client messages are not supported."); + } +} + +export const mailEventInternalHeaders = { + requestId: internalRequestIdHeader, + topics: internalTopicsHeader, + user: internalUserHeader +} as const; + +function parseTopics(value: string | null): MailEventTopic[] { + if (!value) return []; + return [ + ...new Set( + value + .split(",") + .map((topic) => topic.trim()) + .filter((topic): topic is MailEventTopic => + mailEventTopics.includes(topic as MailEventTopic) + ) + ) + ]; +} + +function readConnection(socket: WebSocket): MailEventConnection | null { + const value: unknown = socket.deserializeAttachment(); + if (!value || typeof value !== "object") return null; + const candidate = value as Partial; + if ( + typeof candidate.userId !== "string" || + typeof candidate.expiresAt !== "number" || + !Array.isArray(candidate.topics) + ) { + return null; + } + const topics = candidate.topics.filter( + (topic): topic is MailEventTopic => + typeof topic === "string" && mailEventTopics.includes(topic as MailEventTopic) + ); + return { expiresAt: candidate.expiresAt, topics, userId: candidate.userId }; +} diff --git a/worker/features/events/route.ts b/worker/features/events/route.ts new file mode 100644 index 00000000..c73ac44e --- /dev/null +++ b/worker/features/events/route.ts @@ -0,0 +1,89 @@ +import { MailApiAuthError, mailApiChallenge, requireMailApiPrincipal } from "../../auth/mail-api"; +import type { WorkerEnv } from "../../lib/env"; +import { AppError, errorBody, toAppError } from "../../lib/errors"; +import { jsonResponse } from "../../lib/json"; + +import { mailEventInternalHeaders } from "./durable-object"; +import type { MailEventTopic } from "./types"; + +const eventPath = "/api/v1/events"; +const workspaceHubName = "workspace"; + +export async function handleMailEventRoute( + request: Request, + env: WorkerEnv +): Promise { + if (new URL(request.url).pathname !== eventPath) return null; + + const requestId = requestIdFor(request); + try { + if (request.method !== "GET") { + return eventError("METHOD_NOT_ALLOWED", "Use GET to open the event WebSocket.", 405, { + allow: "GET", + "x-request-id": requestId + }); + } + if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") { + return eventError("WEBSOCKET_UPGRADE_REQUIRED", "Request a WebSocket upgrade.", 426, { + upgrade: "websocket", + "x-request-id": requestId + }); + } + + const principal = await requireMailApiPrincipal(env, request, "mail:read"); + validateSessionOrigin(request, principal.authentication); + const topics: MailEventTopic[] = ["messages", "mailboxes"]; + if (principal.scopes.has("mail:send")) topics.push("drafts"); + + const headers = new Headers({ upgrade: "websocket" }); + headers.set(mailEventInternalHeaders.requestId, requestId); + headers.set(mailEventInternalHeaders.topics, topics.join(",")); + headers.set(mailEventInternalHeaders.user, principal.auth.user.id); + const response = await env.MAIL_EVENTS.getByName(workspaceHubName).fetch( + new Request(request.url, { headers }) + ); + if (response.status !== 101) { + return eventError("EVENT_CONNECTION_FAILED", "The event connection failed.", 503, { + "x-request-id": requestId + }); + } + return response; + } catch (error) { + const appError = toAppError(error); + const headers: Record = { "x-request-id": requestId }; + if (error instanceof MailApiAuthError) { + headers["www-authenticate"] = mailApiChallenge(env, request, error); + } + return eventError(appError.code, appError.message, appError.status, headers); + } +} + +function validateSessionOrigin(request: Request, authentication: "bearer" | "session"): void { + if (authentication !== "session") return; + const origin = request.headers.get("origin"); + if (origin !== null && origin !== new URL(request.url).origin) { + throw new AppError("ORIGIN_FORBIDDEN", "WebSocket origin is not allowed.", 403); + } +} + +function requestIdFor(request: Request): string { + const provided = request.headers.get("x-request-id") ?? ""; + return /^[A-Za-z0-9_-]{8,100}$/.test(provided) ? provided : crypto.randomUUID(); +} + +function eventError( + code: string, + message: string, + status: number, + headers: HeadersInit = {} +): Response { + return jsonResponse(errorBody(code, message), { + status, + headers: { + "cache-control": "no-store", + "referrer-policy": "no-referrer", + "x-content-type-options": "nosniff", + ...Object.fromEntries(new Headers(headers)) + } + }); +} diff --git a/worker/features/events/service.ts b/worker/features/events/service.ts new file mode 100644 index 00000000..10a0d809 --- /dev/null +++ b/worker/features/events/service.ts @@ -0,0 +1,111 @@ +import { type SQL, sql } from "drizzle-orm"; + +import { getRow, getRows } from "../../db/drizzle"; +import type { WorkerEnv } from "../../lib/env"; + +import type { MailEventTopic } from "./types"; + +const workspaceHubName = "workspace"; + +export type MessageEventTarget = { + isUnassigned: boolean; + mailboxId: string | null; +}; + +export async function publishUserMailEvent( + env: WorkerEnv, + userId: string, + topic: MailEventTopic +): Promise { + await publishMailEvent(env, [userId], topic); +} + +export async function publishMessageMailEvent( + env: WorkerEnv, + targets: readonly MessageEventTarget[] +): Promise { + const userIds = await messageEventUserIds(env.DB, targets); + await publishMailEvent(env, userIds, "messages"); +} + +export async function publishMailboxMailEvent(env: WorkerEnv, mailboxId: string): Promise { + const rows = await getRows<{ id: string }>( + env.DB, + sql`SELECT DISTINCT user_row.id + FROM "user" user_row + WHERE COALESCE(user_row.banned, 0) = 0 + AND (user_row.role IN ('owner', 'admin') OR EXISTS ( + SELECT 1 FROM mailbox_grants grant_row + WHERE grant_row.user_id = user_row.id + AND grant_row.mailbox_id = ${mailboxId} + AND grant_row.access_level IN ('read', 'agent', 'manager') + ))` + ); + await publishMailEvent( + env, + rows.map((row) => row.id), + "mailboxes" + ); +} + +export async function messageEventTarget( + db: D1Database, + messageId: string +): Promise { + const row = await getRow<{ is_unassigned: number; mailbox_id: string | null }>( + db, + sql`SELECT mailbox_id, is_unassigned FROM messages WHERE id = ${messageId}` + ); + return row ? { isUnassigned: row.is_unassigned === 1, mailboxId: row.mailbox_id } : null; +} + +export function ignoreMailEventFailure(promise: Promise): Promise { + return promise.catch(() => undefined); +} + +async function publishMailEvent( + env: WorkerEnv, + userIds: readonly string[], + topic: MailEventTopic +): Promise { + const recipients = [...new Set(userIds)]; + if (recipients.length === 0) return; + await env.MAIL_EVENTS.getByName(workspaceHubName).publish({ topic, userIds: recipients }); +} + +async function messageEventUserIds( + db: D1Database, + targets: readonly MessageEventTarget[] +): Promise { + const mailboxIds = [ + ...new Set( + targets.flatMap((target) => + target.isUnassigned || target.mailboxId === null ? [] : [target.mailboxId] + ) + ) + ]; + const includeUnassigned = targets.some((target) => target.isUnassigned); + const visibility: SQL[] = []; + if (includeUnassigned) visibility.push(sql`user_row.role = 'owner'`); + if (mailboxIds.length > 0) { + visibility.push(sql`user_row.role = 'owner' OR EXISTS ( + SELECT 1 FROM mailbox_grants grant_row + WHERE grant_row.user_id = user_row.id + AND grant_row.mailbox_id IN (${sql.join( + mailboxIds.map((mailboxId) => sql`${mailboxId}`), + sql`, ` + )}) + AND grant_row.access_level IN ('read', 'agent', 'manager') + )`); + } + if (visibility.length === 0) return []; + + const rows = await getRows<{ id: string }>( + db, + sql`SELECT DISTINCT user_row.id + FROM "user" user_row + WHERE COALESCE(user_row.banned, 0) = 0 + AND (${sql.join(visibility, sql` OR `)})` + ); + return rows.map((row) => row.id); +} diff --git a/worker/features/events/types.ts b/worker/features/events/types.ts new file mode 100644 index 00000000..a4c4a5f8 --- /dev/null +++ b/worker/features/events/types.ts @@ -0,0 +1,19 @@ +export const mailEventTopics = ["messages", "drafts", "mailboxes"] as const; + +export type MailEventTopic = (typeof mailEventTopics)[number]; + +export type MailEvent = { + type: "changed"; + topic: MailEventTopic; +}; + +export type MailEventPublish = { + topic: MailEventTopic; + userIds: string[]; +}; + +export type MailEventConnection = { + expiresAt: number; + topics: MailEventTopic[]; + userId: string; +}; diff --git a/worker/features/mail-api/discovery.ts b/worker/features/mail-api/discovery.ts index 02cb144d..30146bb5 100644 --- a/worker/features/mail-api/discovery.ts +++ b/worker/features/mail-api/discovery.ts @@ -137,6 +137,7 @@ The method index is an orientation aid. Consult ${openApiUrl} for exact paramete - \`GET ${apiBase}/messages\` and \`GET ${apiBase}/drafts\` return one page. Follow the \`Link: ; rel="next"\` response header for the next page. No \`Link\` header means the last page. - To start message synchronization, get a checkpoint from \`GET ${apiBase}/changes\` without a cursor, paginate the full message list, then read changes after the checkpoint until \`hasMore\` is false. - To start draft synchronization, get a checkpoint from \`GET ${apiBase}/drafts/changes\` without a cursor, paginate the full draft list, then read draft changes after the checkpoint until \`hasMore\` is false. +- Open \`${apiBase}/events\` as a WebSocket when low-latency updates are useful. Each frame only identifies a changed topic. After a frame or reconnect, use the REST resources and change journals to reconcile state. Reconnect with bounded exponential backoff. Keep periodic synchronization as a fallback. - List mailboxes before each change cycle. Remove cached mail for mailboxes that are no longer readable, and bootstrap each newly readable mailbox. - Repeat a full draft bootstrap when mailbox access changes so newly hidden or visible drafts are reconciled. - Ignore response fields you do not recognize. diff --git a/worker/features/mailbox-access/routes.ts b/worker/features/mailbox-access/routes.ts index 11145649..25825d80 100644 --- a/worker/features/mailbox-access/routes.ts +++ b/worker/features/mailbox-access/routes.ts @@ -10,6 +10,7 @@ import { AppError } from "../../lib/errors"; import { readJson } from "../../lib/json"; import { parseWith } from "../../lib/validation"; import { recordAudit } from "../audit/service"; +import { ignoreMailEventFailure, publishUserMailEvent } from "../events/service"; import { listMailboxGrants, revokeMailboxGrant, setMailboxGrant } from "./queries"; const grantSchema = z.object({ @@ -49,6 +50,9 @@ mailboxAccessRoutes.put("/", async (c) => { outcome: "success", metadata: { accessLevel: input.accessLevel } }); + c.executionCtx.waitUntil( + ignoreMailEventFailure(publishUserMailEvent(c.env, input.userId, "mailboxes")) + ); return c.body(null, 204); }); @@ -65,5 +69,8 @@ mailboxAccessRoutes.delete("/:mailboxId/:userId", async (c) => { resourceId: `${c.req.param("mailboxId")}:${c.req.param("userId")}`, outcome: "success" }); + c.executionCtx.waitUntil( + ignoreMailEventFailure(publishUserMailEvent(c.env, c.req.param("userId"), "mailboxes")) + ); return c.body(null, 204); }); diff --git a/worker/features/mailboxes/routes.ts b/worker/features/mailboxes/routes.ts index 6c0a5665..ce9313a7 100644 --- a/worker/features/mailboxes/routes.ts +++ b/worker/features/mailboxes/routes.ts @@ -6,6 +6,7 @@ import type { HonoApp } from "../../lib/env"; import { readJson } from "../../lib/json"; import { parseWith } from "../../lib/validation"; import { recordAudit } from "../audit/service"; +import { ignoreMailEventFailure, publishMailboxMailEvent } from "../events/service"; import { listMailboxesForUser } from "./queries"; import { @@ -42,6 +43,7 @@ mailboxRoutes.post("/", async (c) => { resourceId: mailbox.id, outcome: "success" }); + scheduleMailboxEvent(c, mailbox.id); return c.json(mailbox, 201); }); @@ -60,6 +62,7 @@ mailboxRoutes.patch("/:id", async (c) => { resourceId: c.req.param("id"), outcome: "success" }); + scheduleMailboxEvent(c, c.req.param("id")); return c.json(updated); }); @@ -80,6 +83,7 @@ mailboxRoutes.post("/:id/addresses", async (c) => { resourceId: c.req.param("id"), outcome: "success" }); + scheduleMailboxEvent(c, c.req.param("id")); return c.json(address, 201); }); @@ -96,5 +100,10 @@ mailboxRoutes.delete("/:id/addresses/:addressId", async (c) => { resourceId: c.req.param("id"), outcome: "success" }); + scheduleMailboxEvent(c, c.req.param("id")); return c.body(null, 204); }); + +function scheduleMailboxEvent(c: Context, mailboxId: string): void { + c.executionCtx.waitUntil(ignoreMailEventFailure(publishMailboxMailEvent(c.env, mailboxId))); +} diff --git a/worker/features/mcp/draft-tools.ts b/worker/features/mcp/draft-tools.ts index 6a4d02cc..d94e0799 100644 --- a/worker/features/mcp/draft-tools.ts +++ b/worker/features/mcp/draft-tools.ts @@ -13,6 +13,7 @@ import { saveDraft } from "../drafts/queries"; import { draftSchema } from "../drafts/validation"; +import { ignoreMailEventFailure, publishUserMailEvent } from "../events/service"; import type { McpPrincipal } from "./route"; import { base64File, maxMcpAttachmentBase64Length, toolResult } from "./tool-result"; @@ -73,6 +74,7 @@ export function registerDraftTools( await requireDraftAccess(env, principal, parsed); const draft = await saveDraft(env.DB, principal.userId, parsed); await recordDraftMutation(env, principal, "mcp.draft.create", draft.id); + await notifyDraftChange(env, principal.userId); return draft; }) ); @@ -110,6 +112,7 @@ export function registerDraftTools( await requireDraftAccess(env, principal, parsed); const draft = await saveDraft(env.DB, principal.userId, parsed); await recordDraftMutation(env, principal, "mcp.draft.update", draft.id); + await notifyDraftChange(env, principal.userId); return draft; }) ); @@ -126,6 +129,7 @@ export function registerDraftTools( await getAccessibleDraft(env, principal, draftId); await deleteDraft(env.DB, env.MAIL_OBJECTS, principal.userId, draftId); await recordDraftMutation(env, principal, "mcp.draft.delete", draftId); + await notifyDraftChange(env, principal.userId); return { deleted: true, draftId }; }) ); @@ -173,6 +177,7 @@ function registerDraftAttachmentTools( httpMetadata: { contentType: added.attachment.contentType } }); await recordDraftMutation(env, principal, "mcp.draft.attachment.add", added.attachment.id); + await notifyDraftChange(env, principal.userId); return added.attachment; }) ); @@ -202,6 +207,7 @@ function registerDraftAttachmentTools( throw new AppError("ATTACHMENT_NOT_FOUND", "Attachment not found.", 404); } await recordDraftMutation(env, principal, "mcp.draft.attachment.remove", attachmentId); + await notifyDraftChange(env, principal.userId); return { deleted: true, attachmentId, draftId }; }) ); @@ -223,3 +229,7 @@ function recordDraftMutation( outcome: "success" }); } + +function notifyDraftChange(env: WorkerEnv, userId: string): Promise { + return ignoreMailEventFailure(publishUserMailEvent(env, userId, "drafts")); +} diff --git a/worker/features/mcp/mail-tools.ts b/worker/features/mcp/mail-tools.ts index f93e7acf..8737d5b9 100644 --- a/worker/features/mcp/mail-tools.ts +++ b/worker/features/mcp/mail-tools.ts @@ -5,6 +5,11 @@ import { accessibleMessageScope } from "../../auth/mailbox-access"; import type { WorkerEnv } from "../../lib/env"; import { AppError } from "../../lib/errors"; import { recordAudit } from "../audit/service"; +import { + ignoreMailEventFailure, + messageEventTarget, + publishMessageMailEvent +} from "../events/service"; import { listMailboxesForUser } from "../mailboxes/queries"; import { requireAttachmentAccess, requireMessageAccess } from "../messages/access"; import { listConversations, updateConversationAction } from "../messages/conversation-queries"; @@ -193,6 +198,10 @@ function registerWriteTools(server: McpServer, env: WorkerEnv, principal: McpPri toolResult(async () => { await requireMessageAccess(env.DB, principal.userId, principal.role, messageId, "agent"); const message = await updateMessageAction(env.DB, messageId, action); + const target = await messageEventTarget(env.DB, message.id); + if (target) { + await ignoreMailEventFailure(publishMessageMailEvent(env, [target])); + } await recordMutation(env, principal, `mcp.message.${action}`, "message", messageId); return message; }) @@ -219,12 +228,15 @@ function registerWriteTools(server: McpServer, env: WorkerEnv, principal: McpPri principal.role, "agent" ); - const result = await updateConversationAction(env.DB, { + const { eventTargets, ...result } = await updateConversationAction(env.DB, { action, activeFolder, messageId, scope }); + if (eventTargets.length > 0) { + await ignoreMailEventFailure(publishMessageMailEvent(env, eventTargets)); + } await recordMutation( env, principal, diff --git a/worker/features/messages/conversation-queries.ts b/worker/features/messages/conversation-queries.ts index 77caaaab..fb40b9d8 100644 --- a/worker/features/messages/conversation-queries.ts +++ b/worker/features/messages/conversation-queries.ts @@ -6,6 +6,7 @@ import { nowIso } from "../../db/client"; import { createDatabase, getRow, getRows } from "../../db/drizzle"; import { messages } from "../../db/schema"; import { AppError } from "../../lib/errors"; +import type { MessageEventTarget } from "../events/service"; import type { MessageAction } from "./actions"; import { decodeKeysetCursor, encodeKeysetCursor, type KeysetCursor } from "./keyset-cursor"; @@ -154,7 +155,11 @@ export async function updateConversationAction( messageId: string; scope: MessageScope; } -): Promise<{ affected: number; threadId: string }> { +): Promise<{ + affected: number; + eventTargets: MessageEventTarget[]; + threadId: string; +}> { const scope = messageScopeCondition(input.scope, "mailbox_id", "is_unassigned"); if (!scope) { throw new AppError("MAILBOX_FORBIDDEN", "You do not have access to this mailbox.", 403); @@ -232,8 +237,15 @@ export async function updateConversationAction( .update(messages) .set(set) .where(and(...conditions)) - .returning({ id: messages.id }); - return { affected: result.length, threadId: selected.thread_id }; + .returning({ isUnassigned: messages.isUnassigned, mailboxId: messages.mailboxId }); + return { + affected: result.length, + eventTargets: result.map((row) => ({ + isUnassigned: row.isUnassigned, + mailboxId: row.mailboxId + })), + threadId: selected.thread_id + }; } function mapConversationSummary(row: ConversationRow): ConversationSummary { diff --git a/worker/features/messages/conversation-routes.ts b/worker/features/messages/conversation-routes.ts index 477f5192..9c2ae9d8 100644 --- a/worker/features/messages/conversation-routes.ts +++ b/worker/features/messages/conversation-routes.ts @@ -4,6 +4,7 @@ import { requireMailApiContext } from "../../auth/mail-api"; import { accessibleMessageScope } from "../../auth/mailbox-access"; import type { HonoApp } from "../../lib/env"; import { parseWith } from "../../lib/validation"; +import { ignoreMailEventFailure, publishMessageMailEvent } from "../events/service"; import { requireMessageAccess } from "./access"; import type { MessageAction } from "./actions"; import { listConversationPage, updateConversationAction } from "./conversation-queries"; @@ -58,13 +59,17 @@ for (const action of actions) { requiredAccess ); const body = parseWith(actionBodySchema, await c.req.json().catch(() => ({}))); - return c.json( - await updateConversationAction(c.env.DB, { - action, - activeFolder: body.folder, - messageId: c.req.param("id"), - scope - }) - ); + const { eventTargets, ...result } = await updateConversationAction(c.env.DB, { + action, + activeFolder: body.folder, + messageId: c.req.param("id"), + scope + }); + if (eventTargets.length > 0) { + c.executionCtx.waitUntil( + ignoreMailEventFailure(publishMessageMailEvent(c.env, eventTargets)) + ); + } + return c.json(result); }); } diff --git a/worker/features/messages/routes.ts b/worker/features/messages/routes.ts index 801c8efe..d4d5aa1b 100644 --- a/worker/features/messages/routes.ts +++ b/worker/features/messages/routes.ts @@ -3,7 +3,11 @@ import { isVersionedMailApiRequest, requireMailApiContext } from "../../auth/mai import { accessibleMessageScope } from "../../auth/mailbox-access"; import type { HonoApp } from "../../lib/env"; import { AppError } from "../../lib/errors"; - +import { + ignoreMailEventFailure, + messageEventTarget, + publishMessageMailEvent +} from "../events/service"; import { requireAttachmentAccess, requireMessageAccess } from "./access"; import type { MessageAction } from "./actions"; import { sanitizeMessageHtml } from "./html-sanitizer"; @@ -158,7 +162,12 @@ for (const action of actions) { c.req.param("id"), action === "read" || action === "unread" ? "read" : "agent" ); - return c.json(await updateMessageAction(c.env.DB, c.req.param("id"), action)); + const message = await updateMessageAction(c.env.DB, c.req.param("id"), action); + const target = await messageEventTarget(c.env.DB, message.id); + if (target) { + c.executionCtx.waitUntil(ignoreMailEventFailure(publishMessageMailEvent(c.env, [target]))); + } + return c.json(message); }); } diff --git a/worker/features/send/service.ts b/worker/features/send/service.ts index b8b605d3..08442d10 100644 --- a/worker/features/send/service.ts +++ b/worker/features/send/service.ts @@ -6,6 +6,11 @@ import { drafts } from "../../db/schema"; import type { WorkerEnv } from "../../lib/env"; import { AppError } from "../../lib/errors"; import { draftAttachmentObjects } from "../drafts/queries"; +import { + ignoreMailEventFailure, + publishMessageMailEvent, + publishUserMailEvent +} from "../events/service"; import { findAddressIdentity } from "../mailboxes/address-queries"; import { findMailboxForSending } from "../mailboxes/queries"; import { ensureReplySubject } from "../messages/headers"; @@ -211,6 +216,14 @@ async function storeSentMessage( .where(and(eq(drafts.id, input.draftId), eq(drafts.userId, input.userId))) .run(); } + await Promise.all([ + ignoreMailEventFailure( + publishMessageMailEvent(env, [{ isUnassigned: false, mailboxId: mailbox.id }]) + ), + ...(input.draftId && input.userId + ? [ignoreMailEventFailure(publishUserMailEvent(env, input.userId, "drafts"))] + : []) + ]); return message; } diff --git a/worker/features/users/routes.ts b/worker/features/users/routes.ts index a98e5ea2..5729b7d1 100644 --- a/worker/features/users/routes.ts +++ b/worker/features/users/routes.ts @@ -8,6 +8,7 @@ import { AppError } from "../../lib/errors"; import { readJson } from "../../lib/json"; import { parseWith } from "../../lib/validation"; import { recordAudit } from "../audit/service"; +import { ignoreMailEventFailure, publishUserMailEvent } from "../events/service"; import { listUsers, setWorkspaceUserRole } from "./queries"; import { @@ -125,6 +126,9 @@ userRoutes.patch("/:id", async (c) => { outcome: "success", metadata: { role: input.role } }); + c.executionCtx.waitUntil( + ignoreMailEventFailure(publishUserMailEvent(c.env, c.req.param("id"), "mailboxes")) + ); return c.json({ ok: true }); }); diff --git a/worker/index.ts b/worker/index.ts index efdcfbcb..0f67ceb0 100644 --- a/worker/index.ts +++ b/worker/index.ts @@ -3,6 +3,9 @@ import { sql } from "drizzle-orm"; import { handleMailApiMetadata } from "./auth/mail-api"; import { getRow } from "./db/drizzle"; import { handleInboundEmail } from "./email/inbound"; +import { MailEvents } from "./features/events/durable-object"; +import { handleMailEventRoute } from "./features/events/route"; +import { ignoreMailEventFailure, publishMessageMailEvent } from "./features/events/service"; import { handleMailApiDiscovery } from "./features/mail-api/discovery"; import { handleMcpRoute } from "./features/mcp/route"; import { notifyInboundMessage } from "./features/notifications/delivery"; @@ -10,6 +13,8 @@ import { consumeJobs } from "./jobs/consumer"; import type { WorkerEnv } from "./lib/env"; import { apiRoutes } from "./routes"; +export { MailEvents }; + export default { async fetch(request: Request, env: WorkerEnv, ctx: ExecutionContext): Promise { const url = new URL(request.url); @@ -17,6 +22,8 @@ export default { if (mailApiDiscovery) return mailApiDiscovery; const mailApiMetadata = handleMailApiMetadata(request, env); if (mailApiMetadata) return mailApiMetadata; + const mailEventResponse = await handleMailEventRoute(request, env); + if (mailEventResponse) return mailEventResponse; const mcpResponse = await handleMcpRoute(request, env, ctx); if (mcpResponse) return mcpResponse; if (url.pathname.startsWith("/api/")) { @@ -52,6 +59,13 @@ export default { // Push delivery is additive and never changes accepted inbound mail. }) ); + ctx.waitUntil( + ignoreMailEventFailure( + publishMessageMailEvent(env, [ + { isUnassigned: stored.isUnassigned, mailboxId: stored.message.mailboxId } + ]) + ) + ); } }, diff --git a/worker/jobs/consumer.ts b/worker/jobs/consumer.ts index 8dc3fd2a..e2232e81 100644 --- a/worker/jobs/consumer.ts +++ b/worker/jobs/consumer.ts @@ -3,6 +3,11 @@ import { eq, lt, sql } from "drizzle-orm"; import { nowIso } from "../db/client"; import { createDatabase, getRow, getRows } from "../db/drizzle"; import { messages, operationRuns, rateLimits } from "../db/schema"; +import { + ignoreMailEventFailure, + type MessageEventTarget, + publishMessageMailEvent +} from "../features/events/service"; import type { WorkerEnv } from "../lib/env"; import { operationalLog } from "../observability/log"; import { isJob, type Job } from "./types"; @@ -21,9 +26,14 @@ async function deleteExpiredRows(env: WorkerEnv): Promise } async function applyRetention(env: WorkerEnv): Promise { - const expired = await getRows<{ id: string; raw_r2_key: string | null }>( + const expired = await getRows<{ + id: string; + is_unassigned: number; + mailbox_id: string | null; + raw_r2_key: string | null; + }>( env.DB, - sql`SELECT m.id, m.raw_r2_key FROM messages m + sql`SELECT m.id, m.mailbox_id, m.is_unassigned, m.raw_r2_key FROM messages m JOIN retention_policies p ON p.mailbox_id = m.mailbox_id WHERE (m.folder = 'trash' AND COALESCE(m.trashed_at, m.updated_at) < datetime('now', '-' || p.trash_days || ' days')) @@ -48,6 +58,11 @@ async function applyRetention(env: WorkerEnv): Promise { } if (keys.length) await env.MAIL_OBJECTS.delete(keys); } + const targets: MessageEventTarget[] = expired.map((message) => ({ + isUnassigned: message.is_unassigned === 1, + mailboxId: message.mailbox_id + })); + await ignoreMailEventFailure(publishMessageMailEvent(env, targets)); return expired.length; } diff --git a/wrangler.jsonc b/wrangler.jsonc index 7d98bea3..a6a5374f 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -43,6 +43,15 @@ "bucket_name": "hqbase-mail" } ], + "durable_objects": { + "bindings": [{ "name": "MAIL_EVENTS", "class_name": "MailEvents" }] + }, + "migrations": [ + { + "tag": "mail-events-v1", + "new_sqlite_classes": ["MailEvents"] + } + ], "queues": { "producers": [{ "binding": "HQBASE_JOBS", "queue": "hqbase-jobs" }], "consumers": [ From dd80aa018ba2eccf6944eec845fa28d542a83a6d Mon Sep 17 00:00:00 2001 From: olegberman Date: Sat, 22 Aug 2026 15:07:04 -0400 Subject: [PATCH 2/6] fix: harden real-time event delivery --- api/hqbase-mail-api-v1.openapi.json | 2 +- ...hqbase-mail-api-v1.postman_collection.json | 32 +++--------- ...qbase-mail-api-v1.postman_environment.json | 6 +++ app/features/events/use-mail-events.ts | 5 +- scripts/generate-mail-api-artifacts.mjs | 7 ++- test/integration/worker/mail-api.test.ts | 42 +++++++++++++++- test/unit/scripts/install.test.mjs | 27 +++++++--- test/unit/scripts/mail-api-artifacts.test.mjs | 15 +++++- test/unit/worker/features/send/routes.test.ts | 14 ++++++ .../worker/features/send/send-service.test.ts | 6 --- worker/features/events/durable-object.ts | 50 +++++++++++++++---- worker/features/events/service.ts | 21 ++++++++ worker/features/mcp/draft-tools.ts | 28 +++++++---- worker/features/mcp/mail-tools.ts | 21 +++++--- worker/features/mcp/send-tools.ts | 15 +++++- worker/features/mcp/server.ts | 19 ++++--- worker/features/send/routes.ts | 15 ++++++ worker/features/send/service.ts | 13 ----- 18 files changed, 244 insertions(+), 94 deletions(-) diff --git a/api/hqbase-mail-api-v1.openapi.json b/api/hqbase-mail-api-v1.openapi.json index 2d83f33d..45b66067 100644 --- a/api/hqbase-mail-api-v1.openapi.json +++ b/api/hqbase-mail-api-v1.openapi.json @@ -123,7 +123,7 @@ "/api/v1/events": { "get": { "summary": "Open change event WebSocket", - "description": "Opens an authenticated wake-only WebSocket. The server sends `changed` frames for messages, mailboxes, and permitted drafts. Frames contain no mail data or cursors. After a frame or reconnect, read the authoritative REST resources and change journals. OAuth tokens require `mail:read`; draft frames also require `mail:send`.", + "description": "Opens an authenticated wake-only WebSocket. The server sends `changed` frames for messages, mailboxes, and permitted drafts. Frames contain no mail data or cursors. After a frame or reconnect, read the authoritative REST resources and change journals. OAuth tokens require `mail:read`; draft frames also require `mail:send`. The authorization decision at upgrade is an event-delivery lease for 10 minutes. Credential or session revocation does not close the current socket immediately. The server closes the socket when the lease ends, and reconnection must use current credentials and permissions.", "security": [ { "oauth2": ["mail:read"] diff --git a/api/hqbase-mail-api-v1.postman_collection.json b/api/hqbase-mail-api-v1.postman_collection.json index 44547b36..f8dc9241 100644 --- a/api/hqbase-mail-api-v1.postman_collection.json +++ b/api/hqbase-mail-api-v1.postman_collection.json @@ -2,7 +2,7 @@ "info": { "_postman_id": "62c6dbf4-835d-4a3f-87df-77b7ddcf2db1", "name": "HQBase Mail API v1", - "description": "Generated from api/hqbase-mail-api-v1.openapi.json. Set base_url, run Register public client, and use Postman's OAuth 2.0 Authorization Code flow with PKCE (S256). Auth URL: {{base_url}}/api/auth/oauth2/authorize. Token URL: {{base_url}}/api/auth/oauth2/token. Client ID: {{client_id}}. Scope: mail:read mail:write mail:send offline_access. Add authorization request parameter resource={{api_resource}}, then store the resulting token only in your local environment as access_token. Sending, replying, and forwarding are not idempotent.", + "description": "Generated from api/hqbase-mail-api-v1.openapi.json. Set base_url, run Register public client, and use Postman's OAuth 2.0 Authorization Code flow with PKCE (S256). Auth URL: {{base_url}}/api/auth/oauth2/authorize. Token URL: {{base_url}}/api/auth/oauth2/token. Client ID: {{client_id}}. Scope: mail:read mail:write mail:send offline_access. Add authorization request parameter resource={{api_resource}}, then store the resulting token only in your local environment as access_token. Postman v2.1 HTTP collections cannot contain WebSocket requests. To receive change wakes, create a separate WebSocket request to {{ws_base_url}}/api/v1/events and add Authorization: Bearer {{access_token}}. Sending, replying, and forwarding are not idempotent.", "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json" }, "auth": { @@ -21,6 +21,11 @@ "value": "https://mail.example.com", "type": "string" }, + { + "key": "ws_base_url", + "value": "wss://mail.example.com", + "type": "string" + }, { "key": "api_resource", "value": "{{base_url}}/api/v1", @@ -150,31 +155,6 @@ } ] }, - { - "name": "Events", - "item": [ - { - "name": "Open change event WebSocket", - "request": { - "method": "GET", - "header": [ - { - "key": "Upgrade", - "value": "websocket", - "disabled": false, - "description": "WebSocket upgrade request." - } - ], - "url": { - "raw": "{{base_url}}/api/v1/events", - "variable": [] - }, - "description": "Opens an authenticated wake-only WebSocket. The server sends `changed` frames for messages, mailboxes, and permitted drafts. Frames contain no mail data or cursors. After a frame or reconnect, read the authoritative REST resources and change journals. OAuth tokens require `mail:read`; draft frames also require `mail:send`." - }, - "response": [] - } - ] - }, { "name": "Changes", "item": [ diff --git a/api/hqbase-mail-api-v1.postman_environment.json b/api/hqbase-mail-api-v1.postman_environment.json index 4870cddc..5f1e160a 100644 --- a/api/hqbase-mail-api-v1.postman_environment.json +++ b/api/hqbase-mail-api-v1.postman_environment.json @@ -8,6 +8,12 @@ "enabled": true, "type": "default" }, + { + "key": "ws_base_url", + "value": "wss://mail.example.com", + "enabled": true, + "type": "default" + }, { "key": "api_resource", "value": "{{base_url}}/api/v1", diff --git a/app/features/events/use-mail-events.ts b/app/features/events/use-mail-events.ts index 7f216e0d..348ed696 100644 --- a/app/features/events/use-mail-events.ts +++ b/app/features/events/use-mail-events.ts @@ -19,7 +19,10 @@ type MailEvent = { export function useMailEvents(userId: string | null, handlers: MailEventHandlers): void { const currentHandlers = React.useRef(handlers); - currentHandlers.current = handlers; + + React.useLayoutEffect(() => { + currentHandlers.current = handlers; + }, [handlers]); React.useEffect(() => { if (!userId) return; diff --git a/scripts/generate-mail-api-artifacts.mjs b/scripts/generate-mail-api-artifacts.mjs index 68ad67e1..615b87e4 100644 --- a/scripts/generate-mail-api-artifacts.mjs +++ b/scripts/generate-mail-api-artifacts.mjs @@ -39,6 +39,9 @@ if (process.argv.includes("--write")) { function buildCollection(document) { const folders = new Map(); for (const [route, pathItem] of Object.entries(document.paths)) { + // Postman v2.1 HTTP collections cannot contain a real WebSocket request. + // Keep the socket in OpenAPI and provide manual connection details below. + if (route === "/api/v1/events") continue; for (const method of ["get", "post", "patch", "delete"]) { const operation = pathItem[method]; if (!operation) continue; @@ -54,7 +57,7 @@ function buildCollection(document) { _postman_id: "62c6dbf4-835d-4a3f-87df-77b7ddcf2db1", name: "HQBase Mail API v1", description: - "Generated from api/hqbase-mail-api-v1.openapi.json. Set base_url, run Register public client, and use Postman's OAuth 2.0 Authorization Code flow with PKCE (S256). Auth URL: {{base_url}}/api/auth/oauth2/authorize. Token URL: {{base_url}}/api/auth/oauth2/token. Client ID: {{client_id}}. Scope: mail:read mail:write mail:send offline_access. Add authorization request parameter resource={{api_resource}}, then store the resulting token only in your local environment as access_token. Sending, replying, and forwarding are not idempotent.", + "Generated from api/hqbase-mail-api-v1.openapi.json. Set base_url, run Register public client, and use Postman's OAuth 2.0 Authorization Code flow with PKCE (S256). Auth URL: {{base_url}}/api/auth/oauth2/authorize. Token URL: {{base_url}}/api/auth/oauth2/token. Client ID: {{client_id}}. Scope: mail:read mail:write mail:send offline_access. Add authorization request parameter resource={{api_resource}}, then store the resulting token only in your local environment as access_token. Postman v2.1 HTTP collections cannot contain WebSocket requests. To receive change wakes, create a separate WebSocket request to {{ws_base_url}}/api/v1/events and add Authorization: Bearer {{access_token}}. Sending, replying, and forwarding are not idempotent.", schema: "https://schema.getpostman.com/json/collection/v2.1.0/collection.json" }, auth: { @@ -63,6 +66,7 @@ function buildCollection(document) { }, variable: [ { key: "base_url", value: "https://mail.example.com", type: "string" }, + { key: "ws_base_url", value: "wss://mail.example.com", type: "string" }, { key: "api_resource", value: "{{base_url}}/api/v1", type: "string" }, { key: "client_id", value: "", type: "string" }, { key: "access_token", value: "", type: "string" }, @@ -218,6 +222,7 @@ function buildEnvironment() { name: "HQBase Mail API v1 - local secrets", values: [ { key: "base_url", value: "https://mail.example.com", enabled: true, type: "default" }, + { key: "ws_base_url", value: "wss://mail.example.com", enabled: true, type: "default" }, { key: "api_resource", value: "{{base_url}}/api/v1", enabled: true, type: "default" }, { key: "client_id", value: "", enabled: true, type: "default" }, { key: "access_token", value: "", enabled: true, type: "secret" } diff --git a/test/integration/worker/mail-api.test.ts b/test/integration/worker/mail-api.test.ts index 8358c8b9..3b4f5b6a 100644 --- a/test/integration/worker/mail-api.test.ts +++ b/test/integration/worker/mail-api.test.ts @@ -1,5 +1,5 @@ -import { env, SELF } from "cloudflare:test"; -import { beforeAll, describe, expect, it } from "vitest"; +import { env, runDurableObjectAlarm, runInDurableObject, SELF } from "cloudflare:test"; +import { beforeAll, describe, expect, it, vi } from "vitest"; import mailApiOpenApi from "../../../api/hqbase-mail-api-v1.openapi.json"; import { createAuth } from "../../../worker/auth/auth"; import { applyCurrentMigrations } from "./current-migrations"; @@ -370,6 +370,34 @@ describe("HQBase Mail API v1", () => { draftSocket.close(1000, "Test complete."); }); + it("closes an event socket when its authorization lease expires", async () => { + const socket = await openEventSocket({ + authorization: `Bearer ${readToken}`, + upgrade: "websocket" + }); + const stub = env.MAIL_EVENTS.getByName("workspace"); + + // Consume any older scheduled alarm and let the object schedule this live socket's expiry. + expect(await runDurableObjectAlarm(stub)).toBe(true); + const expiresAt = await runInDurableObject(stub, (_instance, state) => + state.storage.getAlarm() + ); + expect(expiresAt).not.toBeNull(); + + const closed = nextSocketClose(socket); + const now = vi.spyOn(Date, "now").mockReturnValue((expiresAt ?? 0) + 1); + try { + expect(await runDurableObjectAlarm(stub)).toBe(true); + await expect(closed).resolves.toMatchObject({ + code: 1008, + reason: "Reconnect to renew authentication." + }); + } finally { + now.mockRestore(); + socket.close(1000, "Test complete."); + } + }); + it("reads mail with an audience-bound bearer token without exposing storage keys", async () => { const list = await apiFetch("/api/v1/messages", readToken); expect(list.status, await list.clone().text()).toBe(200); @@ -956,3 +984,13 @@ function nextSocketFrame(socket: WebSocket): Promise { ); }); } + +function nextSocketClose(socket: WebSocket): Promise<{ code: number; reason: string }> { + return new Promise((resolve) => { + socket.addEventListener( + "close", + (event) => resolve({ code: event.code, reason: event.reason }), + { once: true } + ); + }); +} diff --git a/test/unit/scripts/install.test.mjs b/test/unit/scripts/install.test.mjs index d2db53ce..b9565aa8 100644 --- a/test/unit/scripts/install.test.mjs +++ b/test/unit/scripts/install.test.mjs @@ -13,6 +13,14 @@ import { updateOAuthManifest } from "../../../scripts/hqbase/oauth.mjs"; const repositoryWranglerConfig = JSON.parse( readFileSync(resolve(import.meta.dirname, "../../../wrangler.jsonc"), "utf8") ); +const mailEventsMigration = [{ tag: "mail-events-v1", new_sqlite_classes: ["MailEvents"] }]; + +function expectMailEventsConfiguration(config) { + expect(config.durable_objects).toEqual({ + bindings: [{ name: "MAIL_EVENTS", class_name: "MailEvents" }] + }); + expect(config.migrations).toEqual(mailEventsMigration); +} describe("HQBase installation resources", () => { it("creates a fresh manifest with independent unclaimed resources", () => { @@ -68,7 +76,7 @@ describe("HQBase installation resources", () => { } }); - it("pins generated Wrangler configuration to the recorded Cloudflare account", () => { + it("configures the MailEvents migration for a fresh installation", () => { const manifest = createManifest("qa", {}); manifest.accountId = "a".repeat(32); @@ -87,12 +95,17 @@ describe("HQBase installation resources", () => { } ] }); - expect(config.durable_objects).toEqual({ - bindings: [{ name: "MAIL_EVENTS", class_name: "MailEvents" }] - }); - expect(config.migrations).toEqual([ - { tag: "mail-events-v1", new_sqlite_classes: ["MailEvents"] } - ]); + expectMailEventsConfiguration(config); + }); + + it("keeps the MailEvents migration when an installed deployment is updated", () => { + const manifest = createManifest("existing", {}); + manifest.accountId = "b".repeat(32); + manifest.d1.id = "d1-existing"; + manifest.d1.ownership = "owned"; + manifest.worker.deployed = true; + + expectMailEventsConfiguration(createWranglerConfig(manifest)); }); it("records customer-managed OAuth as non-secret deployment configuration", () => { diff --git a/test/unit/scripts/mail-api-artifacts.test.mjs b/test/unit/scripts/mail-api-artifacts.test.mjs index 8e67ea13..168036bc 100644 --- a/test/unit/scripts/mail-api-artifacts.test.mjs +++ b/test/unit/scripts/mail-api-artifacts.test.mjs @@ -59,7 +59,17 @@ describe("Mail API public artifacts", () => { const serialized = JSON.stringify(postman); expect(serialized).toContain("/.well-known/oauth-protected-resource/api/v1"); expect(serialized).toContain("/api/auth/oauth2/register"); - expect(serialized).toContain("Open change event WebSocket"); + expect(serialized).toContain("{{ws_base_url}}/api/v1/events"); + expect(postman.variable).toContainEqual({ + key: "ws_base_url", + value: "wss://mail.example.com", + type: "string" + }); + expect( + postman.item + .flatMap((folder) => folder.item) + .some((item) => item.name.includes("Open change event WebSocket")) + ).toBe(false); const oauthSetup = postman.item.find((folder) => folder.name === "OAuth setup"); const registrationRequest = oauthSetup.item.find( (request) => request.name === "Register public client" @@ -67,7 +77,8 @@ describe("Mail API public artifacts", () => { expect(JSON.parse(registrationRequest.request.body.raw).resources).toEqual([ "{{api_resource}}" ]); - for (const pathItem of Object.values(openApi.paths)) { + for (const [route, pathItem] of Object.entries(openApi.paths)) { + if (route === "/api/v1/events") continue; for (const operation of Object.values(pathItem)) { expect(serialized).toContain(operation.summary); } diff --git a/test/unit/worker/features/send/routes.test.ts b/test/unit/worker/features/send/routes.test.ts index a6668a2c..47c52b86 100644 --- a/test/unit/worker/features/send/routes.test.ts +++ b/test/unit/worker/features/send/routes.test.ts @@ -10,6 +10,7 @@ const mocks = vi.hoisted(() => ({ requireDraftIdAccess: vi.fn(), requireMailApiContext: vi.fn(), requireMailboxAccess: vi.fn(), + scheduleSentMailEvents: vi.fn(), sendForwardDraft: vi.fn(), sendNewMessage: vi.fn() })); @@ -31,6 +32,9 @@ vi.mock("@worker/features/drafts/access", () => ({ requireDraftAttachmentIdsAccess: mocks.requireDraftAttachmentIdsAccess, requireDraftIdAccess: mocks.requireDraftIdAccess })); +vi.mock("@worker/features/events/service", () => ({ + scheduleSentMailEvents: mocks.scheduleSentMailEvents +})); vi.mock("@worker/features/mailboxes/queries", () => ({ findMailboxForSending: mocks.findMailboxForSending })); @@ -94,6 +98,11 @@ describe("send routes", () => { "agent" ); expect(mocks.sendNewMessage).toHaveBeenCalledOnce(); + expect(mocks.scheduleSentMailEvents).toHaveBeenCalledWith( + expect.objectContaining({ DB: db }), + expect.any(Function), + { draftId: "draft-forward", mailboxId: "mailbox-1", userId: "user-1" } + ); }); it("includes original attachments when sending a web forward draft", async () => { @@ -133,5 +142,10 @@ describe("send routes", () => { "user-1" ); expect(mocks.sendNewMessage).not.toHaveBeenCalled(); + expect(mocks.scheduleSentMailEvents).toHaveBeenCalledWith( + expect.objectContaining({ DB: db }), + expect.any(Function), + { draftId: "draft-forward", mailboxId: "mailbox-1", userId: "user-1" } + ); }); }); diff --git a/test/unit/worker/features/send/send-service.test.ts b/test/unit/worker/features/send/send-service.test.ts index 14f9ed62..84b67a7a 100644 --- a/test/unit/worker/features/send/send-service.test.ts +++ b/test/unit/worker/features/send/send-service.test.ts @@ -11,12 +11,6 @@ vi.mock("@worker/features/mailboxes/queries", () => ({ vi.mock("@worker/features/mailboxes/address-queries", () => ({ findAddressIdentity: vi.fn().mockResolvedValue(null) })); -vi.mock("@worker/features/events/service", () => ({ - ignoreMailEventFailure: (promise: Promise) => promise.catch(() => undefined), - publishMessageMailEvent: () => Promise.resolve(), - publishUserMailEvent: () => Promise.resolve() -})); - vi.mock("@worker/features/messages/queries", () => ({ getMessageDetail: vi.fn(), getMessageHtmlKey: vi.fn(), diff --git a/worker/features/events/durable-object.ts b/worker/features/events/durable-object.ts index 80384caa..d66baebc 100644 --- a/worker/features/events/durable-object.ts +++ b/worker/features/events/durable-object.ts @@ -27,19 +27,20 @@ export class MailEvents extends DurableObject { if (!userId || topics.length === 0) { return new Response("Authenticated event context required.", { status: 403 }); } - if (this.ctx.getWebSockets().length >= maxWorkspaceConnections) { - return new Response("Event connection capacity reached.", { status: 503 }); - } + let connections = this.liveConnections(Date.now()); const userTag = `user:${userId}`; - const existing = this.ctx.getWebSockets(userTag); + const existing = connections.filter((socket) => readConnection(socket)?.userId === userId); if (existing.length >= maxConnectionsPerUser) { - existing - .sort( - (left, right) => - (readConnection(left)?.expiresAt ?? 0) - (readConnection(right)?.expiresAt ?? 0) - )[0] - ?.close(1008, "A newer connection replaced this one."); + const replaced = existing.sort( + (left, right) => + (readConnection(left)?.expiresAt ?? 0) - (readConnection(right)?.expiresAt ?? 0) + )[0]; + replaced?.close(1008, "A newer connection replaced this one."); + if (replaced) connections = connections.filter((socket) => socket !== replaced); + } + if (connections.length >= maxWorkspaceConnections) { + return new Response("Event connection capacity reached.", { status: 503 }); } const pair = new WebSocketPair(); @@ -52,6 +53,7 @@ export class MailEvents extends DurableObject { }; this.ctx.acceptWebSocket(server, [...topics, userTag]); server.serializeAttachment(connection); + await this.scheduleExpiryAlarm(connection.expiresAt); return new Response(null, { status: 101, @@ -89,6 +91,34 @@ export class MailEvents extends DurableObject { override async webSocketMessage(socket: WebSocket): Promise { socket.close(1008, "Client messages are not supported."); } + + override async alarm(): Promise { + const connections = this.liveConnections(Date.now()); + const nextExpiry = connections.reduce((earliest, socket) => { + const expiresAt = readConnection(socket)?.expiresAt; + if (expiresAt === undefined) return earliest; + return earliest === null ? expiresAt : Math.min(earliest, expiresAt); + }, null); + if (nextExpiry !== null) await this.ctx.storage.setAlarm(nextExpiry); + } + + private liveConnections(now: number): WebSocket[] { + const live: WebSocket[] = []; + for (const socket of this.ctx.getWebSockets()) { + const connection = readConnection(socket); + if (!connection || connection.expiresAt <= now) { + socket.close(1008, "Reconnect to renew authentication."); + continue; + } + live.push(socket); + } + return live; + } + + private async scheduleExpiryAlarm(expiresAt: number): Promise { + const current = await this.ctx.storage.getAlarm(); + if (current === null || expiresAt < current) await this.ctx.storage.setAlarm(expiresAt); + } } export const mailEventInternalHeaders = { diff --git a/worker/features/events/service.ts b/worker/features/events/service.ts index 10a0d809..c721afb0 100644 --- a/worker/features/events/service.ts +++ b/worker/features/events/service.ts @@ -7,6 +7,8 @@ import type { MailEventTopic } from "./types"; const workspaceHubName = "workspace"; +export type MailEventScheduler = (promise: Promise) => void; + export type MessageEventTarget = { isUnassigned: boolean; mailboxId: string | null; @@ -63,6 +65,24 @@ export function ignoreMailEventFailure(promise: Promise): Promise { return promise.catch(() => undefined); } +export function scheduleMailEvent(schedule: MailEventScheduler, promise: Promise): void { + schedule(ignoreMailEventFailure(promise)); +} + +export function scheduleSentMailEvents( + env: WorkerEnv, + schedule: MailEventScheduler, + input: { draftId?: string | null | undefined; mailboxId: string; userId: string } +): void { + scheduleMailEvent( + schedule, + publishMessageMailEvent(env, [{ isUnassigned: false, mailboxId: input.mailboxId }]) + ); + if (input.draftId) { + scheduleMailEvent(schedule, publishUserMailEvent(env, input.userId, "drafts")); + } +} + async function publishMailEvent( env: WorkerEnv, userIds: readonly string[], @@ -88,6 +108,7 @@ async function messageEventUserIds( const visibility: SQL[] = []; if (includeUnassigned) visibility.push(sql`user_row.role = 'owner'`); if (mailboxIds.length > 0) { + // Admins can manage mailbox metadata, but mail content still requires a mailbox grant. visibility.push(sql`user_row.role = 'owner' OR EXISTS ( SELECT 1 FROM mailbox_grants grant_row WHERE grant_row.user_id = user_row.id diff --git a/worker/features/mcp/draft-tools.ts b/worker/features/mcp/draft-tools.ts index d94e0799..b1f27cff 100644 --- a/worker/features/mcp/draft-tools.ts +++ b/worker/features/mcp/draft-tools.ts @@ -13,7 +13,11 @@ import { saveDraft } from "../drafts/queries"; import { draftSchema } from "../drafts/validation"; -import { ignoreMailEventFailure, publishUserMailEvent } from "../events/service"; +import { + type MailEventScheduler, + publishUserMailEvent, + scheduleMailEvent +} from "../events/service"; import type { McpPrincipal } from "./route"; import { base64File, maxMcpAttachmentBase64Length, toolResult } from "./tool-result"; @@ -35,7 +39,8 @@ const createDraftShape = { export function registerDraftTools( server: McpServer, env: WorkerEnv, - principal: McpPrincipal + principal: McpPrincipal, + schedule: MailEventScheduler ): void { if (!principal.scopes.has("mail:send")) return; @@ -74,7 +79,7 @@ export function registerDraftTools( await requireDraftAccess(env, principal, parsed); const draft = await saveDraft(env.DB, principal.userId, parsed); await recordDraftMutation(env, principal, "mcp.draft.create", draft.id); - await notifyDraftChange(env, principal.userId); + notifyDraftChange(env, principal.userId, schedule); return draft; }) ); @@ -112,7 +117,7 @@ export function registerDraftTools( await requireDraftAccess(env, principal, parsed); const draft = await saveDraft(env.DB, principal.userId, parsed); await recordDraftMutation(env, principal, "mcp.draft.update", draft.id); - await notifyDraftChange(env, principal.userId); + notifyDraftChange(env, principal.userId, schedule); return draft; }) ); @@ -129,18 +134,19 @@ export function registerDraftTools( await getAccessibleDraft(env, principal, draftId); await deleteDraft(env.DB, env.MAIL_OBJECTS, principal.userId, draftId); await recordDraftMutation(env, principal, "mcp.draft.delete", draftId); - await notifyDraftChange(env, principal.userId); + notifyDraftChange(env, principal.userId, schedule); return { deleted: true, draftId }; }) ); - registerDraftAttachmentTools(server, env, principal); + registerDraftAttachmentTools(server, env, principal, schedule); } function registerDraftAttachmentTools( server: McpServer, env: WorkerEnv, - principal: McpPrincipal + principal: McpPrincipal, + schedule: MailEventScheduler ): void { server.registerTool( "add_draft_attachment", @@ -177,7 +183,7 @@ function registerDraftAttachmentTools( httpMetadata: { contentType: added.attachment.contentType } }); await recordDraftMutation(env, principal, "mcp.draft.attachment.add", added.attachment.id); - await notifyDraftChange(env, principal.userId); + notifyDraftChange(env, principal.userId, schedule); return added.attachment; }) ); @@ -207,7 +213,7 @@ function registerDraftAttachmentTools( throw new AppError("ATTACHMENT_NOT_FOUND", "Attachment not found.", 404); } await recordDraftMutation(env, principal, "mcp.draft.attachment.remove", attachmentId); - await notifyDraftChange(env, principal.userId); + notifyDraftChange(env, principal.userId, schedule); return { deleted: true, attachmentId, draftId }; }) ); @@ -230,6 +236,6 @@ function recordDraftMutation( }); } -function notifyDraftChange(env: WorkerEnv, userId: string): Promise { - return ignoreMailEventFailure(publishUserMailEvent(env, userId, "drafts")); +function notifyDraftChange(env: WorkerEnv, userId: string, schedule: MailEventScheduler): void { + scheduleMailEvent(schedule, publishUserMailEvent(env, userId, "drafts")); } diff --git a/worker/features/mcp/mail-tools.ts b/worker/features/mcp/mail-tools.ts index 8737d5b9..fbce732f 100644 --- a/worker/features/mcp/mail-tools.ts +++ b/worker/features/mcp/mail-tools.ts @@ -6,9 +6,10 @@ import type { WorkerEnv } from "../../lib/env"; import { AppError } from "../../lib/errors"; import { recordAudit } from "../audit/service"; import { - ignoreMailEventFailure, + type MailEventScheduler, messageEventTarget, - publishMessageMailEvent + publishMessageMailEvent, + scheduleMailEvent } from "../events/service"; import { listMailboxesForUser } from "../mailboxes/queries"; import { requireAttachmentAccess, requireMessageAccess } from "../messages/access"; @@ -41,10 +42,11 @@ const conversationFolderSchema = z.enum(conversationFolders); export function registerMailTools( server: McpServer, env: WorkerEnv, - principal: McpPrincipal + principal: McpPrincipal, + schedule: MailEventScheduler ): void { if (principal.scopes.has("mail:read")) registerReadTools(server, env, principal); - if (principal.scopes.has("mail:write")) registerWriteTools(server, env, principal); + if (principal.scopes.has("mail:write")) registerWriteTools(server, env, principal, schedule); } function registerReadTools(server: McpServer, env: WorkerEnv, principal: McpPrincipal): void { @@ -182,7 +184,12 @@ function registerReadTools(server: McpServer, env: WorkerEnv, principal: McpPrin ); } -function registerWriteTools(server: McpServer, env: WorkerEnv, principal: McpPrincipal): void { +function registerWriteTools( + server: McpServer, + env: WorkerEnv, + principal: McpPrincipal, + schedule: MailEventScheduler +): void { server.registerTool( "update_message", { @@ -200,7 +207,7 @@ function registerWriteTools(server: McpServer, env: WorkerEnv, principal: McpPri const message = await updateMessageAction(env.DB, messageId, action); const target = await messageEventTarget(env.DB, message.id); if (target) { - await ignoreMailEventFailure(publishMessageMailEvent(env, [target])); + scheduleMailEvent(schedule, publishMessageMailEvent(env, [target])); } await recordMutation(env, principal, `mcp.message.${action}`, "message", messageId); return message; @@ -235,7 +242,7 @@ function registerWriteTools(server: McpServer, env: WorkerEnv, principal: McpPri scope }); if (eventTargets.length > 0) { - await ignoreMailEventFailure(publishMessageMailEvent(env, eventTargets)); + scheduleMailEvent(schedule, publishMessageMailEvent(env, eventTargets)); } await recordMutation( env, diff --git a/worker/features/mcp/send-tools.ts b/worker/features/mcp/send-tools.ts index f52c1390..fb7fe2c0 100644 --- a/worker/features/mcp/send-tools.ts +++ b/worker/features/mcp/send-tools.ts @@ -8,6 +8,7 @@ import { parseWith } from "../../lib/validation"; import { enforceRateLimit } from "../../security/rate-limit"; import { recordAudit } from "../audit/service"; import { requireDraftAttachmentIdsAccess, requireDraftIdAccess } from "../drafts/access"; +import { type MailEventScheduler, scheduleSentMailEvents } from "../events/service"; import { findMailboxForSending } from "../mailboxes/queries"; import { requireMessageAccess } from "../messages/access"; import { forwardMessage } from "../send/forward"; @@ -23,7 +24,8 @@ const attachmentIds = z.array(z.string().min(1).max(100)).max(20).default([]); export function registerSendTools( server: McpServer, env: WorkerEnv, - principal: McpPrincipal + principal: McpPrincipal, + schedule: MailEventScheduler ): void { if (!principal.scopes.has("mail:send")) return; @@ -53,6 +55,11 @@ export function registerSendTools( await requireDraftIdAccess(env, principal, parsed.draftId); await requireDraftAttachmentIdsAccess(env, principal, parsed.attachmentIds); const message = await sendNewMessage(env, parsed, principal.userId); + scheduleSentMailEvents(env, schedule, { + draftId: parsed.draftId, + mailboxId, + userId: principal.userId + }); await recordSend(env, principal, "mcp.message.send", mailboxId); return message; }) @@ -85,6 +92,11 @@ export function registerSendTools( await requireDraftIdAccess(env, principal, parsed.draftId); await requireDraftAttachmentIdsAccess(env, principal, parsed.attachmentIds); const message = await replyToMessage(env, parsed, principal.userId); + scheduleSentMailEvents(env, schedule, { + draftId: parsed.draftId, + mailboxId, + userId: principal.userId + }); await recordSend(env, principal, "mcp.message.reply", mailboxId); return message; }) @@ -117,6 +129,7 @@ export function registerSendTools( const mailboxId = await requireSendingAccess(env, principal, parsed.from); await requireDraftAttachmentIdsAccess(env, principal, parsed.attachmentIds); const message = await forwardMessage(env, parsed, principal.userId); + scheduleSentMailEvents(env, schedule, { mailboxId, userId: principal.userId }); await recordSend(env, principal, "mcp.message.forward", mailboxId); return message; }) diff --git a/worker/features/mcp/server.ts b/worker/features/mcp/server.ts index 4d487702..b99d7b81 100644 --- a/worker/features/mcp/server.ts +++ b/worker/features/mcp/server.ts @@ -2,6 +2,7 @@ import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; import type { WorkerEnv } from "../../lib/env"; +import type { MailEventScheduler } from "../events/service"; import { registerDraftTools } from "./draft-tools"; import { registerMailTools } from "./mail-tools"; import type { McpPrincipal } from "./route"; @@ -10,11 +11,12 @@ import { registerSendTools } from "./send-tools"; export async function serveMcp( request: Request, env: WorkerEnv, - _ctx: ExecutionContext, + ctx: ExecutionContext, principal: McpPrincipal ): Promise { const server = new McpServer({ name: "HQBase", version: "1.0.0" }); - registerTools(server, env, principal); + const schedule: MailEventScheduler = (promise) => ctx.waitUntil(promise); + registerTools(server, env, principal, schedule); const url = new URL(request.url); const transport = new WebStandardStreamableHTTPServerTransport({ allowedOrigins: [url.origin], @@ -25,8 +27,13 @@ export async function serveMcp( return transport.handleRequest(request); } -function registerTools(server: McpServer, env: WorkerEnv, principal: McpPrincipal): void { - registerMailTools(server, env, principal); - registerDraftTools(server, env, principal); - registerSendTools(server, env, principal); +function registerTools( + server: McpServer, + env: WorkerEnv, + principal: McpPrincipal, + schedule: MailEventScheduler +): void { + registerMailTools(server, env, principal, schedule); + registerDraftTools(server, env, principal, schedule); + registerSendTools(server, env, principal, schedule); } diff --git a/worker/features/send/routes.ts b/worker/features/send/routes.ts index 6446a67b..d250c4d9 100644 --- a/worker/features/send/routes.ts +++ b/worker/features/send/routes.ts @@ -12,6 +12,7 @@ import { requireDraftAttachmentIdsAccess, requireDraftIdAccess } from "../drafts/access"; +import { scheduleSentMailEvents } from "../events/service"; import { findMailboxForSending } from "../mailboxes/queries"; import { requireMessageAccess } from "../messages/access"; @@ -45,6 +46,11 @@ sendRoutes.post("/send", async (c) => { const sent = draft?.forwardOfMessageId ? await sendForwardDraft(c.env, input, draft.id, draft.forwardOfMessageId, authContext.user.id) : await sendNewMessage(c.env, input, authContext.user.id); + scheduleSentMailEvents(c.env, (promise) => c.executionCtx.waitUntil(promise), { + draftId: input.draftId, + mailboxId: mailbox.id, + userId: authContext.user.id + }); await recordAudit(c.env.DB, { correlationId: c.get("correlationId"), actorType: "user", @@ -86,6 +92,11 @@ sendRoutes.post("/reply", async (c) => { await requireDraftIdAccess(c.env, principal, input.draftId); await requireDraftAttachmentIdsAccess(c.env, principal, input.attachmentIds); const sent = await replyToMessage(c.env, input, authContext.user.id); + scheduleSentMailEvents(c.env, (promise) => c.executionCtx.waitUntil(promise), { + draftId: input.draftId, + mailboxId: mailbox.id, + userId: authContext.user.id + }); await recordAudit(c.env.DB, { correlationId: c.get("correlationId"), actorType: "user", @@ -126,6 +137,10 @@ sendRoutes.post("/forward", async (c) => { const principal = { role: authContext.user.role, userId: authContext.user.id }; await requireDraftAttachmentIdsAccess(c.env, principal, input.attachmentIds); const sent = await forwardMessage(c.env, input, authContext.user.id); + scheduleSentMailEvents(c.env, (promise) => c.executionCtx.waitUntil(promise), { + mailboxId: mailbox.id, + userId: authContext.user.id + }); await recordAudit(c.env.DB, { correlationId: c.get("correlationId"), actorType: "user", diff --git a/worker/features/send/service.ts b/worker/features/send/service.ts index 08442d10..b8b605d3 100644 --- a/worker/features/send/service.ts +++ b/worker/features/send/service.ts @@ -6,11 +6,6 @@ import { drafts } from "../../db/schema"; import type { WorkerEnv } from "../../lib/env"; import { AppError } from "../../lib/errors"; import { draftAttachmentObjects } from "../drafts/queries"; -import { - ignoreMailEventFailure, - publishMessageMailEvent, - publishUserMailEvent -} from "../events/service"; import { findAddressIdentity } from "../mailboxes/address-queries"; import { findMailboxForSending } from "../mailboxes/queries"; import { ensureReplySubject } from "../messages/headers"; @@ -216,14 +211,6 @@ async function storeSentMessage( .where(and(eq(drafts.id, input.draftId), eq(drafts.userId, input.userId))) .run(); } - await Promise.all([ - ignoreMailEventFailure( - publishMessageMailEvent(env, [{ isUnassigned: false, mailboxId: mailbox.id }]) - ), - ...(input.draftId && input.userId - ? [ignoreMailEventFailure(publishUserMailEvent(env, input.userId, "drafts"))] - : []) - ]); return message; } From 0a2cbc41f695b9ed7617d5c5e9f3dc5a435ef532 Mon Sep 17 00:00:00 2001 From: olegberman Date: Sat, 22 Aug 2026 15:35:31 -0400 Subject: [PATCH 3/6] feat: show real-time connection status --- app/app.tsx | 14 +- app/components/layout/app-shell.tsx | 6 + app/components/layout/mobile-navigation.tsx | 4 + app/components/layout/sidebar.tsx | 5 + .../sidebar/mail-connection-indicator.tsx | 47 ++++++ app/components/layout/top-bar.tsx | 4 + app/features/drafts/use-drafts.ts | 7 - app/features/events/types.ts | 1 + app/features/events/use-mail-events.ts | 156 ++++++++++++++++-- app/features/messages/use-mail-sync.ts | 11 +- test/integration/worker/mail-api.test.ts | 9 + test/unit/app/events/use-mail-events.test.tsx | 62 +++++++ test/unit/app/layout/mail-shell.test.tsx | 29 ++++ worker/features/events/durable-object.ts | 5 + 14 files changed, 335 insertions(+), 25 deletions(-) create mode 100644 app/components/layout/sidebar/mail-connection-indicator.tsx create mode 100644 app/features/events/types.ts diff --git a/app/app.tsx b/app/app.tsx index 75853336..41344419 100644 --- a/app/app.tsx +++ b/app/app.tsx @@ -122,11 +122,20 @@ export function App(): React.ReactElement { if (!currentUser.passwordSetupRequired) await loadWorkspace(currentUser); }, [currentUserId, loadWorkspace]); const refreshRealtimeState = React.useCallback(async () => { - await Promise.allSettled([refreshWorkspace(), mailSync.refresh(), draftState.refresh()]); + const results = await Promise.allSettled([ + refreshWorkspace(), + mailSync.refresh(), + draftState.refresh() + ]); + if (results.every((result) => result.status === "rejected")) { + const failure = results.find((result) => result.status === "rejected"); + throw failure?.reason; + } }, [draftState.refresh, mailSync.refresh, refreshWorkspace]); - useMailEvents(currentUserId, { + const connectionStatus = useMailEvents(currentUserId, { onDrafts: draftState.refresh, + onFallbackPoll: refreshRealtimeState, onMailboxes: refreshRealtimeState, onMessages: mailSync.refresh, onReconnect: refreshRealtimeState @@ -210,6 +219,7 @@ export function App(): React.ReactElement { activeFolder={activeFolder} activeSettingsTab={settingsTab} canManage={user.role === "owner" || user.role === "admin"} + connectionStatus={connectionStatus} draftCount={draftState.drafts.length} mailboxId={mailboxId} mailboxes={contentMailboxes} diff --git a/app/components/layout/app-shell.tsx b/app/components/layout/app-shell.tsx index c0c1726c..37953e23 100644 --- a/app/components/layout/app-shell.tsx +++ b/app/components/layout/app-shell.tsx @@ -1,5 +1,6 @@ import * as React from "react"; import type { CurrentUser } from "@/features/auth/types"; +import type { MailConnectionStatus } from "@/features/events/types"; import type { Mailbox } from "@/features/mailboxes/types"; import type { UnreadCounts } from "@/features/notifications/types"; import type { UpdateStatus } from "@/features/updates/types"; @@ -15,6 +16,7 @@ type AppShellProps = { activeFolder: FolderId; activeSettingsTab?: import("@/lib/routes").SettingsTabId | undefined; canManage?: boolean | undefined; + connectionStatus: MailConnectionStatus; children: React.ReactNode; mailboxId: string; mailboxes: Mailbox[]; @@ -55,6 +57,7 @@ export function AppShell(props: AppShellProps): React.ReactElement { activeFolder={props.activeFolder} activeSettingsTab={props.activeSettingsTab} canManage={props.canManage} + connectionStatus={props.connectionStatus} draftCount={props.draftCount} mailboxId={props.mailboxId} sidebarCollapsed={sidebarCollapsed} @@ -84,6 +87,7 @@ export function AppShell(props: AppShellProps): React.ReactElement { activeFolder={props.activeFolder} activeSettingsTab={props.activeSettingsTab} canManage={props.canManage} + connectionStatus={props.connectionStatus} draftCount={props.draftCount} mailboxId={props.mailboxId} unread={props.unread} @@ -121,6 +125,7 @@ function ShellContent({ activeSettingsTab, canManage, children, + connectionStatus, draftCount, mailboxId, mailboxes, @@ -150,6 +155,7 @@ function ShellContent({ activeFolder={activeFolder} activeSettingsTab={activeSettingsTab} canManage={canManage} + connectionStatus={connectionStatus} draftCount={draftCount} mailboxId={mailboxId} mailboxes={mailboxes} diff --git a/app/components/layout/mobile-navigation.tsx b/app/components/layout/mobile-navigation.tsx index e2c7aef4..ba4b6488 100644 --- a/app/components/layout/mobile-navigation.tsx +++ b/app/components/layout/mobile-navigation.tsx @@ -4,6 +4,7 @@ import { PiList } from "react-icons/pi"; import { Button } from "@/components/ui/button"; import { Sheet, SheetContent, SheetTitle, SheetTrigger } from "@/components/ui/sheet"; import type { CurrentUser } from "@/features/auth/types"; +import type { MailConnectionStatus } from "@/features/events/types"; import type { Mailbox } from "@/features/mailboxes/types"; import type { UnreadCounts } from "@/features/notifications/types"; import type { FolderId, SettingsTabId } from "@/lib/routes"; @@ -13,6 +14,7 @@ type MobileNavigationProps = { activeFolder: FolderId; activeSettingsTab?: SettingsTabId | undefined; canManage?: boolean | undefined; + connectionStatus?: MailConnectionStatus | undefined; draftCount: number; mailboxId: string; mailboxes: Mailbox[]; @@ -29,6 +31,7 @@ export function MobileNavigation({ activeFolder, activeSettingsTab, canManage, + connectionStatus, draftCount, mailboxId, mailboxes, @@ -93,6 +96,7 @@ export function MobileNavigation({ activeFolder={activeFolder} activeSettingsTab={activeSettingsTab} canManage={canManage} + connectionStatus={connectionStatus} draftCount={draftCount} mailboxId={mailboxId} mailboxFilter={{ diff --git a/app/components/layout/sidebar.tsx b/app/components/layout/sidebar.tsx index a768a20d..14dfa13a 100644 --- a/app/components/layout/sidebar.tsx +++ b/app/components/layout/sidebar.tsx @@ -4,6 +4,7 @@ import { PiSidebar, PiSidebarSimple } from "react-icons/pi"; import { Button } from "@/components/ui/button"; import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from "@/components/ui/tooltip"; import type { CurrentUser } from "@/features/auth/types"; +import type { MailConnectionStatus } from "@/features/events/types"; import type { Mailbox } from "@/features/mailboxes/types"; import type { UnreadCounts } from "@/features/notifications/types"; import { cn } from "@/lib/cn"; @@ -11,6 +12,7 @@ import type { FolderId, SettingsTabId } from "@/lib/routes"; import { appRoutePath } from "@/lib/routes"; import { AccountMenu } from "./account-menu"; import { quickAccess } from "./sidebar/constants"; +import { MailConnectionIndicator } from "./sidebar/mail-connection-indicator"; import { isModifiedNavigation } from "./sidebar/sidebar-helpers"; import { MailNav, SettingsNav } from "./sidebar/sidebar-nav"; @@ -32,6 +34,7 @@ type SidebarProps = { sidebarCollapsed?: boolean; activeSettingsTab?: SettingsTabId | undefined; canManage?: boolean | undefined; + connectionStatus?: MailConnectionStatus | undefined; onSettingsTabChange?: ((tab: SettingsTabId) => void) | undefined; onToggleSidebar?: () => void; }; @@ -50,6 +53,7 @@ export function Sidebar({ sidebarCollapsed = false, activeSettingsTab, canManage = false, + connectionStatus = "connecting", onSettingsTabChange, onToggleSidebar }: SidebarProps): React.ReactElement { @@ -137,6 +141,7 @@ export function Sidebar({ Mail + {onToggleSidebar ? ( diff --git a/app/components/layout/sidebar/mail-connection-indicator.tsx b/app/components/layout/sidebar/mail-connection-indicator.tsx new file mode 100644 index 00000000..40b774aa --- /dev/null +++ b/app/components/layout/sidebar/mail-connection-indicator.tsx @@ -0,0 +1,47 @@ +import type * as React from "react"; + +import type { MailConnectionStatus } from "@/features/events/types"; +import { cn } from "@/lib/cn"; + +const statusPresentation: Record = { + connecting: { + className: "bg-muted-foreground/45 motion-safe:animate-pulse", + label: "Connecting to live updates" + }, + connected: { + className: "bg-emerald-500 ring-2 ring-emerald-500/15", + label: "Live updates connected" + }, + fallback: { + className: "bg-amber-400 ring-2 ring-amber-400/15", + label: "Using fallback sync while live updates reconnect" + }, + unavailable: { + className: "bg-red-500 ring-2 ring-red-500/15", + label: "Cannot connect to HQBase" + } +}; + +export function MailConnectionIndicator({ + status +}: { + status: MailConnectionStatus; +}): React.ReactElement { + const presentation = statusPresentation[status]; + + return ( + + + ); +} diff --git a/app/components/layout/top-bar.tsx b/app/components/layout/top-bar.tsx index 05d08bfa..0d32aa82 100644 --- a/app/components/layout/top-bar.tsx +++ b/app/components/layout/top-bar.tsx @@ -12,6 +12,7 @@ import { SelectValue } from "@/components/ui/select"; import type { CurrentUser } from "@/features/auth/types"; +import type { MailConnectionStatus } from "@/features/events/types"; import type { Mailbox } from "@/features/mailboxes/types"; import type { UnreadCounts } from "@/features/notifications/types"; import { mailboxUnreadLabel } from "@/features/notifications/unread"; @@ -22,6 +23,7 @@ type TopBarProps = { activeFolder: FolderId; activeSettingsTab?: SettingsTabId | undefined; canManage?: boolean | undefined; + connectionStatus?: MailConnectionStatus | undefined; draftCount: number; user: CurrentUser; mailboxes: Mailbox[]; @@ -42,6 +44,7 @@ export function TopBar({ activeFolder, activeSettingsTab, canManage, + connectionStatus, draftCount, user, mailboxes, @@ -76,6 +79,7 @@ export function TopBar({ activeFolder={activeFolder} activeSettingsTab={activeSettingsTab} canManage={canManage} + connectionStatus={connectionStatus} draftCount={draftCount} mailboxId={mailboxId} mailboxes={mailboxes} diff --git a/app/features/drafts/use-drafts.ts b/app/features/drafts/use-drafts.ts index 9680c9ed..857b1846 100644 --- a/app/features/drafts/use-drafts.ts +++ b/app/features/drafts/use-drafts.ts @@ -3,8 +3,6 @@ import * as React from "react"; import { listDrafts } from "./api"; import type { Draft } from "./types"; -const refreshIntervalMs = 60_000; - export function useDrafts(userId: string | null): { drafts: Draft[]; isLoading: boolean; @@ -44,14 +42,9 @@ export function useDrafts(userId: string | null): { const refreshWhenVisible = (): void => { if (document.visibilityState === "visible") void refresh().catch(() => undefined); }; - const interval = window.setInterval( - () => void refresh().catch(() => undefined), - refreshIntervalMs - ); window.addEventListener("focus", refreshWhenVisible); document.addEventListener("visibilitychange", refreshWhenVisible); return () => { - window.clearInterval(interval); window.removeEventListener("focus", refreshWhenVisible); document.removeEventListener("visibilitychange", refreshWhenVisible); }; diff --git a/app/features/events/types.ts b/app/features/events/types.ts new file mode 100644 index 00000000..b7a43476 --- /dev/null +++ b/app/features/events/types.ts @@ -0,0 +1 @@ +export type MailConnectionStatus = "connecting" | "connected" | "fallback" | "unavailable"; diff --git a/app/features/events/use-mail-events.ts b/app/features/events/use-mail-events.ts index 348ed696..5bf9610a 100644 --- a/app/features/events/use-mail-events.ts +++ b/app/features/events/use-mail-events.ts @@ -1,12 +1,20 @@ import * as React from "react"; +import type { MailConnectionStatus } from "./types"; + const reconnectBaseDelayMs = 1_000; const reconnectMaxDelayMs = 30_000; +const connectionTimeoutMs = 10_000; +const fallbackPollBaseDelayMs = 30_000; +const fallbackPollMaxDelayMs = 60_000; +const heartbeatIntervalMs = 30_000; +const heartbeatTimeoutMs = 10_000; type MailEventTopic = "drafts" | "mailboxes" | "messages"; type MailEventHandlers = { onDrafts: () => unknown; + onFallbackPoll: () => unknown; onMailboxes: () => unknown; onMessages: () => unknown; onReconnect: () => unknown; @@ -17,8 +25,12 @@ type MailEvent = { type: "changed"; }; -export function useMailEvents(userId: string | null, handlers: MailEventHandlers): void { +export function useMailEvents( + userId: string | null, + handlers: MailEventHandlers +): MailConnectionStatus { const currentHandlers = React.useRef(handlers); + const [status, setStatus] = React.useState("connecting"); React.useLayoutEffect(() => { currentHandlers.current = handlers; @@ -29,6 +41,12 @@ export function useMailEvents(userId: string | null, handlers: MailEventHandlers let active = true; let attempt = 0; + let connectionTimer: number | null = null; + let fallbackAttempt = 0; + let fallbackInFlight = false; + let fallbackTimer: number | null = null; + let heartbeatTimer: number | null = null; + let heartbeatTimeoutTimer: number | null = null; let reconnectTimer: number | null = null; let socket: WebSocket | null = null; const pendingTopics = new Set(); @@ -61,7 +79,26 @@ export function useMailEvents(userId: string | null, handlers: MailEventHandlers }; const canConnect = (): boolean => active && document.visibilityState === "visible" && navigator.onLine !== false; + const socketIsOpen = (): boolean => socket?.readyState === WebSocket.OPEN; + const clearConnectionTimer = (): void => { + if (connectionTimer === null) return; + window.clearTimeout(connectionTimer); + connectionTimer = null; + }; + const clearFallbackTimer = (): void => { + if (fallbackTimer === null) return; + window.clearTimeout(fallbackTimer); + fallbackTimer = null; + }; + const clearHeartbeatTimers = (): void => { + if (heartbeatTimer !== null) window.clearTimeout(heartbeatTimer); + if (heartbeatTimeoutTimer !== null) window.clearTimeout(heartbeatTimeoutTimer); + heartbeatTimer = null; + heartbeatTimeoutTimer = null; + }; const closeSocket = (): void => { + clearConnectionTimer(); + clearHeartbeatTimers(); const current = socket; socket = null; if (current && current.readyState < WebSocket.CLOSING) { @@ -69,6 +106,72 @@ export function useMailEvents(userId: string | null, handlers: MailEventHandlers } }; + const scheduleHeartbeat = (current: WebSocket): void => { + clearHeartbeatTimers(); + heartbeatTimer = window.setTimeout(() => { + heartbeatTimer = null; + if (!active || socket !== current || current.readyState !== WebSocket.OPEN) return; + try { + current.send("ping"); + } catch { + current.close(); + return; + } + heartbeatTimeoutTimer = window.setTimeout(() => { + heartbeatTimeoutTimer = null; + if (active && socket === current && current.readyState === WebSocket.OPEN) { + current.close(4000, "Heartbeat timed out."); + } + }, heartbeatTimeoutMs); + }, heartbeatIntervalMs); + }; + + const scheduleFallbackPoll = (): void => { + if (!canConnect() || socketIsOpen() || fallbackTimer !== null || fallbackInFlight) return; + const delay = Math.min( + fallbackPollMaxDelayMs, + fallbackPollBaseDelayMs * 2 ** Math.min(fallbackAttempt, 1) + ); + fallbackTimer = window.setTimeout(() => { + fallbackTimer = null; + runFallbackPoll(); + }, delay); + }; + + const runFallbackPoll = (): void => { + if (!canConnect() || socketIsOpen() || fallbackInFlight) return; + fallbackInFlight = true; + setStatus("fallback"); + void Promise.resolve() + .then(currentHandlers.current.onFallbackPoll) + .then( + () => { + if (!active || socketIsOpen()) return; + fallbackAttempt = 0; + setStatus("fallback"); + }, + () => { + if (!active || socketIsOpen()) return; + fallbackAttempt += 1; + setStatus("unavailable"); + } + ) + .finally(() => { + fallbackInFlight = false; + scheduleFallbackPoll(); + }); + }; + + const scheduleReconnect = (): void => { + if (!canConnect() || socket !== null || reconnectTimer !== null) return; + const delay = Math.min(reconnectMaxDelayMs, reconnectBaseDelayMs * 2 ** Math.min(attempt, 5)); + attempt += 1; + reconnectTimer = window.setTimeout(() => { + reconnectTimer = null; + connect(); + }, delay + reconnectJitterMs()); + }; + const connect = (): void => { if (!canConnect() || socket !== null) return; @@ -76,12 +179,34 @@ export function useMailEvents(userId: string | null, handlers: MailEventHandlers url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; const next = new WebSocket(url); socket = next; + connectionTimer = window.setTimeout(() => { + connectionTimer = null; + if (!active || socket !== next || next.readyState !== WebSocket.CONNECTING) return; + socket = null; + next.close(); + runFallbackPoll(); + scheduleReconnect(); + }, connectionTimeoutMs); next.addEventListener("open", () => { if (!active || socket !== next) return; + clearConnectionTimer(); + clearFallbackTimer(); attempt = 0; + fallbackAttempt = 0; + setStatus("connected"); + scheduleHeartbeat(next); invoke(currentHandlers.current.onReconnect); }); next.addEventListener("message", (event) => { + if (event.data === "pong") { + if (heartbeatTimeoutTimer !== null) window.clearTimeout(heartbeatTimeoutTimer); + heartbeatTimeoutTimer = null; + if (socket === next) { + setStatus("connected"); + scheduleHeartbeat(next); + } + return; + } const parsed = parseMailEvent(event.data); if (parsed) queueTopic(parsed.topic); }); @@ -91,16 +216,14 @@ export function useMailEvents(userId: string | null, handlers: MailEventHandlers next.addEventListener("close", () => { if (socket !== next) return; socket = null; - if (!canConnect() || reconnectTimer !== null) return; - const delay = Math.min( - reconnectMaxDelayMs, - reconnectBaseDelayMs * 2 ** Math.min(attempt, 5) - ); - attempt += 1; - reconnectTimer = window.setTimeout(() => { - reconnectTimer = null; - connect(); - }, delay + reconnectJitterMs()); + clearConnectionTimer(); + clearHeartbeatTimers(); + if (!canConnect()) { + if (navigator.onLine === false) setStatus("unavailable"); + return; + } + runFallbackPoll(); + scheduleReconnect(); }); }; const resume = (): void => { @@ -108,6 +231,7 @@ export function useMailEvents(userId: string | null, handlers: MailEventHandlers window.clearTimeout(reconnectTimer); reconnectTimer = null; } + setStatus("connecting"); connect(); }; const handleVisibilityChange = (): void => { @@ -115,14 +239,20 @@ export function useMailEvents(userId: string | null, handlers: MailEventHandlers else closeSocket(); }; const handleOnline = (): void => resume(); - const handleOffline = (): void => closeSocket(); + const handleOffline = (): void => { + clearFallbackTimer(); + closeSocket(); + setStatus("unavailable"); + }; + setStatus("connecting"); document.addEventListener("visibilitychange", handleVisibilityChange); window.addEventListener("online", handleOnline); window.addEventListener("offline", handleOffline); connect(); return () => { active = false; + clearFallbackTimer(); if (reconnectTimer !== null) window.clearTimeout(reconnectTimer); document.removeEventListener("visibilitychange", handleVisibilityChange); window.removeEventListener("online", handleOnline); @@ -130,6 +260,8 @@ export function useMailEvents(userId: string | null, handlers: MailEventHandlers closeSocket(); }; }, [userId]); + + return status; } function parseMailEvent(value: unknown): MailEvent | null { diff --git a/app/features/messages/use-mail-sync.ts b/app/features/messages/use-mail-sync.ts index 1da57150..b07d2ec2 100644 --- a/app/features/messages/use-mail-sync.ts +++ b/app/features/messages/use-mail-sync.ts @@ -8,8 +8,6 @@ import type { FolderId } from "@/lib/routes"; import { listConversations } from "./api"; import type { ConversationAction, ConversationSummary } from "./types"; -const refreshIntervalMs = 60_000; - type MailSyncOptions = { activeFolder: FolderId; mailboxId: string; @@ -100,6 +98,13 @@ export function useMailSync({ activeFolder, mailboxId, search, userId }: MailSyn } if (conversationResult.status === "rejected") throw conversationResult.reason; + if ( + conversationResult.status === "fulfilled" && + conversationResult.value === null && + notificationResult.status === "rejected" + ) { + throw notificationResult.reason; + } })(); inFlight.current = { key: syncKey, promise }; const clearInFlight = (): void => { @@ -150,13 +155,11 @@ export function useMailSync({ activeFolder, mailboxId, search, userId }: MailSyn }; runRefresh(true); - const interval = window.setInterval(runRefresh, refreshIntervalMs); window.addEventListener("focus", refreshWhenVisible); document.addEventListener("visibilitychange", refreshWhenVisible); navigator.serviceWorker?.addEventListener("message", handleServiceWorkerMessage); return () => { active = false; - window.clearInterval(interval); window.removeEventListener("focus", refreshWhenVisible); document.removeEventListener("visibilitychange", refreshWhenVisible); navigator.serviceWorker?.removeEventListener("message", handleServiceWorkerMessage); diff --git a/test/integration/worker/mail-api.test.ts b/test/integration/worker/mail-api.test.ts index 3b4f5b6a..b966cad8 100644 --- a/test/integration/worker/mail-api.test.ts +++ b/test/integration/worker/mail-api.test.ts @@ -349,6 +349,9 @@ describe("HQBase Mail API v1", () => { authorization: `Bearer ${readToken}`, upgrade: "websocket" }); + const pong = nextSocketMessage(messageSocket); + messageSocket.send("ping"); + await expect(pong).resolves.toBe("pong"); const messageFrame = nextSocketFrame(messageSocket); await env.MAIL_EVENTS.getByName("workspace").publish({ topic: "messages", @@ -985,6 +988,12 @@ function nextSocketFrame(socket: WebSocket): Promise { }); } +function nextSocketMessage(socket: WebSocket): Promise { + return new Promise((resolve) => { + socket.addEventListener("message", (event) => resolve(String(event.data)), { once: true }); + }); +} + function nextSocketClose(socket: WebSocket): Promise<{ code: number; reason: string }> { return new Promise((resolve) => { socket.addEventListener( diff --git a/test/unit/app/events/use-mail-events.test.tsx b/test/unit/app/events/use-mail-events.test.tsx index 75cb1b17..96a82d3b 100644 --- a/test/unit/app/events/use-mail-events.test.tsx +++ b/test/unit/app/events/use-mail-events.test.tsx @@ -25,6 +25,8 @@ class FakeWebSocket extends EventTarget { this.dispatchEvent(new CloseEvent("close")); }); + send = vi.fn(); + open(): void { this.readyState = FakeWebSocket.OPEN; this.dispatchEvent(new Event("open")); @@ -38,6 +40,7 @@ class FakeWebSocket extends EventTarget { function handlers() { return { onDrafts: vi.fn(), + onFallbackPoll: vi.fn(), onMailboxes: vi.fn(), onMessages: vi.fn(), onReconnect: vi.fn() @@ -69,7 +72,9 @@ describe("useMailEvents", () => { const socket = FakeWebSocket.instances[0]; expect(socket?.url).toBe("ws://localhost:3000/api/v1/events"); + expect(hook.result).toBe("connecting"); await flushHookEffects(() => socket?.open()); + expect(hook.result).toBe("connected"); expect(callbacks.onReconnect).toHaveBeenCalledOnce(); await flushHookEffects(() => { @@ -142,4 +147,61 @@ describe("useMailEvents", () => { expect(FakeWebSocket.instances).toHaveLength(2); await hook.unmount(); }); + + it("uses successful fallback sync while reconnecting", async () => { + const callbacks = handlers(); + callbacks.onFallbackPoll.mockResolvedValue(undefined); + const hook = await renderHook(() => useMailEvents("user-1", callbacks), undefined); + const socket = FakeWebSocket.instances[0]; + await flushHookEffects(() => socket?.open()); + await flushHookEffects(() => socket?.close()); + + expect(callbacks.onFallbackPoll).toHaveBeenCalledOnce(); + expect(hook.result).toBe("fallback"); + await hook.unmount(); + }); + + it("reports unavailable when both live events and fallback sync fail", async () => { + const callbacks = handlers(); + callbacks.onFallbackPoll.mockRejectedValue(new Error("API unavailable")); + const hook = await renderHook(() => useMailEvents("user-1", callbacks), undefined); + const socket = FakeWebSocket.instances[0]; + await flushHookEffects(() => socket?.open()); + await flushHookEffects(() => socket?.close()); + + expect(hook.result).toBe("unavailable"); + await hook.unmount(); + }); + + it("checks an open socket with an application heartbeat", async () => { + vi.useFakeTimers(); + const hook = await renderHook(() => useMailEvents("user-1", handlers()), undefined); + const socket = FakeWebSocket.instances[0]; + await flushHookEffects(() => socket?.open()); + + await flushHookEffects(() => vi.advanceTimersByTime(30_000)); + expect(socket?.send).toHaveBeenCalledWith("ping"); + await flushHookEffects(() => socket?.message("pong")); + await flushHookEffects(() => vi.advanceTimersByTime(10_000)); + + expect(socket?.close).not.toHaveBeenCalled(); + expect(hook.result).toBe("connected"); + await hook.unmount(); + }); + + it("falls back and reconnects when the heartbeat expires", async () => { + vi.useFakeTimers(); + const callbacks = handlers(); + callbacks.onFallbackPoll.mockResolvedValue(undefined); + const hook = await renderHook(() => useMailEvents("user-1", callbacks), undefined); + const socket = FakeWebSocket.instances[0]; + await flushHookEffects(() => socket?.open()); + + await flushHookEffects(() => vi.advanceTimersByTime(40_000)); + + expect(socket?.close).toHaveBeenCalledWith(4000, "Heartbeat timed out."); + expect(callbacks.onFallbackPoll).toHaveBeenCalledOnce(); + expect(hook.result).toBe("fallback"); + await hook.unmount(); + }); }); diff --git a/test/unit/app/layout/mail-shell.test.tsx b/test/unit/app/layout/mail-shell.test.tsx index a44cd76e..bf1b4b39 100644 --- a/test/unit/app/layout/mail-shell.test.tsx +++ b/test/unit/app/layout/mail-shell.test.tsx @@ -112,6 +112,35 @@ describe("mail shell", () => { expect(topBarHtml).not.toContain('aria-label="Hide sidebar"'); }); + it("shows an accessible mail connection status beside the sidebar title", () => { + const labels = { + connecting: "Connecting to live updates", + connected: "Live updates connected", + fallback: "Using fallback sync while live updates reconnect", + unavailable: "Cannot connect to HQBase" + } as const; + + for (const [connectionStatus, label] of Object.entries(labels)) { + const html = renderToStaticMarkup( + undefined} + onSignedOut={() => undefined} + /> + ); + + expect(html).toContain(`data-connection-status="${connectionStatus}"`); + expect(html).toContain(`aria-label="${label}"`); + expect(html.indexOf(">Mail")).toBeLessThan( + html.indexOf(`data-connection-status="${connectionStatus}"`) + ); + } + }); + it("renders the canonical logo instead of the HQ placeholder", () => { const html = renderToStaticMarkup( { + constructor(ctx: DurableObjectState, env: WorkerEnv) { + super(ctx, env); + ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong")); + } + override async fetch(request: Request): Promise { if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") { return new Response("WebSocket upgrade required.", { status: 426 }); From 7bc196448fe596cdcf37ccb27af24338cf929960 Mon Sep 17 00:00:00 2001 From: olegberman Date: Sat, 22 Aug 2026 15:51:03 -0400 Subject: [PATCH 4/6] fix: harden real-time event connections --- test/integration/worker/mail-api.test.ts | 45 ++++++++++++++++++++++++ worker/features/events/durable-object.ts | 1 + worker/features/events/route.ts | 2 +- 3 files changed, 47 insertions(+), 1 deletion(-) diff --git a/test/integration/worker/mail-api.test.ts b/test/integration/worker/mail-api.test.ts index b966cad8..e34f29eb 100644 --- a/test/integration/worker/mail-api.test.ts +++ b/test/integration/worker/mail-api.test.ts @@ -344,6 +344,24 @@ describe("HQBase Mail API v1", () => { headers: { cookie, origin: "https://other.example", upgrade: "websocket" } }); expect(invalidOrigin.status).toBe(403); + await expect(invalidOrigin.json()).resolves.toMatchObject({ + error: { code: "ORIGIN_FORBIDDEN" } + }); + + const missingOrigin = await SELF.fetch(`${origin}/api/v1/events`, { + headers: { cookie, upgrade: "websocket" } + }); + expect(missingOrigin.status).toBe(403); + await expect(missingOrigin.json()).resolves.toMatchObject({ + error: { code: "ORIGIN_FORBIDDEN" } + }); + + const sessionSocket = await openEventSocket({ + cookie, + origin, + upgrade: "websocket" + }); + sessionSocket.close(1000, "Test complete."); const messageSocket = await openEventSocket({ authorization: `Bearer ${readToken}`, @@ -373,6 +391,33 @@ describe("HQBase Mail API v1", () => { draftSocket.close(1000, "Test complete."); }); + it("does not count a closing event socket toward the per-user limit", async () => { + const headers = { + authorization: `Bearer ${readToken}`, + upgrade: "websocket" + }; + const firstSocket = await openEventSocket(headers); + const secondSocket = await openEventSocket(headers); + const closingSocket = await openEventSocket(headers); + closingSocket.close(1000, "Reconnect test."); + + const replacementSocket = await openEventSocket(headers); + const firstFrame = nextSocketFrame(firstSocket); + const secondFrame = nextSocketFrame(secondSocket); + const replacementFrame = nextSocketFrame(replacementSocket); + await env.MAIL_EVENTS.getByName("workspace").publish({ + topic: "messages", + userIds: [userId] + }); + + await expect(firstFrame).resolves.toEqual({ type: "changed", topic: "messages" }); + await expect(secondFrame).resolves.toEqual({ type: "changed", topic: "messages" }); + await expect(replacementFrame).resolves.toEqual({ type: "changed", topic: "messages" }); + firstSocket.close(1000, "Test complete."); + secondSocket.close(1000, "Test complete."); + replacementSocket.close(1000, "Test complete."); + }); + it("closes an event socket when its authorization lease expires", async () => { const socket = await openEventSocket({ authorization: `Bearer ${readToken}`, diff --git a/worker/features/events/durable-object.ts b/worker/features/events/durable-object.ts index adbe4f44..802353e7 100644 --- a/worker/features/events/durable-object.ts +++ b/worker/features/events/durable-object.ts @@ -110,6 +110,7 @@ export class MailEvents extends DurableObject { private liveConnections(now: number): WebSocket[] { const live: WebSocket[] = []; for (const socket of this.ctx.getWebSockets()) { + if (socket.readyState !== WebSocket.OPEN) continue; const connection = readConnection(socket); if (!connection || connection.expiresAt <= now) { socket.close(1008, "Reconnect to renew authentication."); diff --git a/worker/features/events/route.ts b/worker/features/events/route.ts index c73ac44e..2c1b41d7 100644 --- a/worker/features/events/route.ts +++ b/worker/features/events/route.ts @@ -61,7 +61,7 @@ export async function handleMailEventRoute( function validateSessionOrigin(request: Request, authentication: "bearer" | "session"): void { if (authentication !== "session") return; const origin = request.headers.get("origin"); - if (origin !== null && origin !== new URL(request.url).origin) { + if (origin !== new URL(request.url).origin) { throw new AppError("ORIGIN_FORBIDDEN", "WebSocket origin is not allowed.", 403); } } From 2adce20a36650019ea4e542169514890b6397b8f Mon Sep 17 00:00:00 2001 From: olegberman Date: Sat, 22 Aug 2026 16:14:16 -0400 Subject: [PATCH 5/6] fix: retry real-time event publication --- test/integration/worker/mail-api.test.ts | 89 +++++++++++++++---- .../worker/features/events/service.test.ts | 39 ++++++++ worker/features/events/service.ts | 24 ++++- 3 files changed, 132 insertions(+), 20 deletions(-) create mode 100644 test/unit/worker/features/events/service.test.ts diff --git a/test/integration/worker/mail-api.test.ts b/test/integration/worker/mail-api.test.ts index e34f29eb..db4f2aaa 100644 --- a/test/integration/worker/mail-api.test.ts +++ b/test/integration/worker/mail-api.test.ts @@ -2,6 +2,7 @@ import { env, runDurableObjectAlarm, runInDurableObject, SELF } from "cloudflare import { beforeAll, describe, expect, it, vi } from "vitest"; import mailApiOpenApi from "../../../api/hqbase-mail-api-v1.openapi.json"; import { createAuth } from "../../../worker/auth/auth"; +import { mailEventInternalHeaders } from "../../../worker/features/events/durable-object"; import { applyCurrentMigrations } from "./current-migrations"; import { tokenRow } from "./mail-api-token-fixture"; @@ -361,7 +362,7 @@ describe("HQBase Mail API v1", () => { origin, upgrade: "websocket" }); - sessionSocket.close(1000, "Test complete."); + await closeEventSocket(sessionSocket); const messageSocket = await openEventSocket({ authorization: `Bearer ${readToken}`, @@ -376,7 +377,7 @@ describe("HQBase Mail API v1", () => { userIds: [userId] }); await expect(messageFrame).resolves.toEqual({ type: "changed", topic: "messages" }); - messageSocket.close(1000, "Test complete."); + await closeEventSocket(messageSocket); const draftSocket = await openEventSocket({ authorization: `Bearer ${fullToken}`, @@ -388,7 +389,7 @@ describe("HQBase Mail API v1", () => { userIds: [userId] }); await expect(draftFrame).resolves.toEqual({ type: "changed", topic: "drafts" }); - draftSocket.close(1000, "Test complete."); + await closeEventSocket(draftSocket); }); it("does not count a closing event socket toward the per-user limit", async () => { @@ -398,24 +399,49 @@ describe("HQBase Mail API v1", () => { }; const firstSocket = await openEventSocket(headers); const secondSocket = await openEventSocket(headers); - const closingSocket = await openEventSocket(headers); - closingSocket.close(1000, "Reconnect test."); + const thirdSocket = await openEventSocket(headers); + const clientCloses = [firstSocket, secondSocket, thirdSocket].map(nextSocketClose); + const stub = env.MAIL_EVENTS.getByName("workspace"); - const replacementSocket = await openEventSocket(headers); - const firstFrame = nextSocketFrame(firstSocket); - const secondFrame = nextSocketFrame(secondSocket); - const replacementFrame = nextSocketFrame(replacementSocket); - await env.MAIL_EVENTS.getByName("workspace").publish({ - topic: "messages", - userIds: [userId] + const state = await runInDurableObject(stub, async (instance, durableState) => { + const initialSockets = durableState + .getWebSockets() + .filter((socket) => socket.readyState === WebSocket.OPEN); + const closingSocket = initialSockets.at(-1); + if (!closingSocket) throw new Error("Expected an open event socket."); + closingSocket.close(1000, "Reconnect test."); + const closingReadyState = closingSocket.readyState; + + const response = await instance.fetch( + new Request(`${origin}/api/v1/events`, { + headers: { + [mailEventInternalHeaders.requestId]: "request_reconnect_test", + [mailEventInternalHeaders.topics]: "messages,mailboxes", + [mailEventInternalHeaders.user]: userId, + upgrade: "websocket" + } + }) + ); + const replacementSocket = response.webSocket; + if (!replacementSocket) throw new Error("Expected a replacement event socket."); + replacementSocket.accept(); + const openSocketCount = durableState + .getWebSockets() + .filter((socket) => socket.readyState === WebSocket.OPEN).length; + + for (const socket of durableState.getWebSockets()) { + if (socket.readyState === WebSocket.OPEN) socket.close(1000, "Test complete."); + } + return { closingReadyState, openSocketCount, status: response.status }; }); - await expect(firstFrame).resolves.toEqual({ type: "changed", topic: "messages" }); - await expect(secondFrame).resolves.toEqual({ type: "changed", topic: "messages" }); - await expect(replacementFrame).resolves.toEqual({ type: "changed", topic: "messages" }); - firstSocket.close(1000, "Test complete."); - secondSocket.close(1000, "Test complete."); - replacementSocket.close(1000, "Test complete."); + expect(state).toEqual({ + closingReadyState: WebSocket.CLOSING, + openSocketCount: 3, + status: 101 + }); + await Promise.all(clientCloses); + await expectOpenEventSocketCount(0); }); it("closes an event socket when its authorization lease expires", async () => { @@ -491,7 +517,7 @@ describe("HQBase Mail API v1", () => { expect(archived.status, await archived.clone().text()).toBe(200); await expect(archived.json()).resolves.toMatchObject({ folder: "archived" }); await expect(changed).resolves.toEqual({ type: "changed", topic: "messages" }); - socket.close(1000, "Test complete."); + await closeEventSocket(socket); const unarchived = await apiFetch("/api/v1/messages/msg_api/unarchive", writeToken, { method: "POST" @@ -1048,3 +1074,28 @@ function nextSocketClose(socket: WebSocket): Promise<{ code: number; reason: str ); }); } + +async function closeEventSocket(socket: WebSocket): Promise { + const closed = nextSocketClose(socket); + const stub = env.MAIL_EVENTS.getByName("workspace"); + await runInDurableObject(stub, (_instance, state) => { + const openSockets = state + .getWebSockets() + .filter((serverSocket) => serverSocket.readyState === WebSocket.OPEN); + if (openSockets.length !== 1) { + throw new Error(`Expected one open event socket, found ${openSockets.length}.`); + } + openSockets[0]?.close(1000, "Test complete."); + }); + await closed; +} + +async function expectOpenEventSocketCount(expected: number): Promise { + const stub = env.MAIL_EVENTS.getByName("workspace"); + const count = await runInDurableObject( + stub, + (_instance, state) => + state.getWebSockets().filter((socket) => socket.readyState === WebSocket.OPEN).length + ); + expect(count).toBe(expected); +} diff --git a/test/unit/worker/features/events/service.test.ts b/test/unit/worker/features/events/service.test.ts new file mode 100644 index 00000000..5a6166db --- /dev/null +++ b/test/unit/worker/features/events/service.test.ts @@ -0,0 +1,39 @@ +import { retryMailEventPublish } from "@worker/features/events/service"; +import { describe, expect, it, vi } from "vitest"; + +describe("mail event publication", () => { + it("retries a failed wake publication with bounded delays", async () => { + const failure = new Error("event hub unavailable"); + const publish = vi + .fn() + .mockRejectedValueOnce(failure) + .mockRejectedValueOnce(failure) + .mockResolvedValue(undefined); + const wait = vi.fn().mockResolvedValue(undefined); + + await retryMailEventPublish(publish, wait); + + expect(publish).toHaveBeenCalledTimes(3); + expect(wait.mock.calls).toEqual([[100], [200]]); + }); + + it("returns after the first successful wake publication", async () => { + const publish = vi.fn().mockResolvedValue(undefined); + const wait = vi.fn().mockResolvedValue(undefined); + + await retryMailEventPublish(publish, wait); + + expect(publish).toHaveBeenCalledOnce(); + expect(wait).not.toHaveBeenCalled(); + }); + + it("reports failure after all three wake attempts", async () => { + const failure = new Error("event hub unavailable"); + const publish = vi.fn().mockRejectedValue(failure); + const wait = vi.fn().mockResolvedValue(undefined); + + await expect(retryMailEventPublish(publish, wait)).rejects.toBe(failure); + expect(publish).toHaveBeenCalledTimes(3); + expect(wait.mock.calls).toEqual([[100], [200]]); + }); +}); diff --git a/worker/features/events/service.ts b/worker/features/events/service.ts index c721afb0..4e45480c 100644 --- a/worker/features/events/service.ts +++ b/worker/features/events/service.ts @@ -6,6 +6,8 @@ import type { WorkerEnv } from "../../lib/env"; import type { MailEventTopic } from "./types"; const workspaceHubName = "workspace"; +const publishAttempts = 3; +const publishRetryBaseDelayMs = 100; export type MailEventScheduler = (promise: Promise) => void; @@ -83,6 +85,21 @@ export function scheduleSentMailEvents( } } +export async function retryMailEventPublish( + publish: () => Promise, + wait: (delayMs: number) => Promise = (delayMs) => scheduler.wait(delayMs) +): Promise { + for (let attempt = 0; attempt < publishAttempts; attempt += 1) { + try { + await publish(); + return; + } catch (error) { + if (attempt === publishAttempts - 1) throw error; + await wait(publishRetryBaseDelayMs * 2 ** attempt); + } + } +} + async function publishMailEvent( env: WorkerEnv, userIds: readonly string[], @@ -90,7 +107,12 @@ async function publishMailEvent( ): Promise { const recipients = [...new Set(userIds)]; if (recipients.length === 0) return; - await env.MAIL_EVENTS.getByName(workspaceHubName).publish({ topic, userIds: recipients }); + await retryMailEventPublish(() => + env.MAIL_EVENTS.getByName(workspaceHubName).publish({ + topic, + userIds: recipients + }) + ); } async function messageEventUserIds( From 8b605d9bc1255a8a3dafa2d7cb2ddd5cf3553a4a Mon Sep 17 00:00:00 2001 From: olegberman Date: Sat, 22 Aug 2026 16:47:34 -0400 Subject: [PATCH 6/6] fix: bound fallback sync cadence --- app/features/events/use-mail-events.ts | 2 +- test/unit/app/events/use-mail-events.test.tsx | 23 +++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/app/features/events/use-mail-events.ts b/app/features/events/use-mail-events.ts index 5bf9610a..d1c2cf84 100644 --- a/app/features/events/use-mail-events.ts +++ b/app/features/events/use-mail-events.ts @@ -139,7 +139,7 @@ export function useMailEvents( }; const runFallbackPoll = (): void => { - if (!canConnect() || socketIsOpen() || fallbackInFlight) return; + if (!canConnect() || socketIsOpen() || fallbackTimer !== null || fallbackInFlight) return; fallbackInFlight = true; setStatus("fallback"); void Promise.resolve() diff --git a/test/unit/app/events/use-mail-events.test.tsx b/test/unit/app/events/use-mail-events.test.tsx index 96a82d3b..69262792 100644 --- a/test/unit/app/events/use-mail-events.test.tsx +++ b/test/unit/app/events/use-mail-events.test.tsx @@ -161,6 +161,29 @@ describe("useMailEvents", () => { await hook.unmount(); }); + it("keeps failed reconnects within the fallback polling cadence", async () => { + vi.useFakeTimers(); + vi.spyOn(crypto, "getRandomValues").mockImplementation((values) => { + if (values instanceof Uint32Array) values[0] = 0; + return values; + }); + const callbacks = handlers(); + callbacks.onFallbackPoll.mockResolvedValue(undefined); + const hook = await renderHook(() => useMailEvents("user-1", callbacks), undefined); + const first = FakeWebSocket.instances[0]; + await flushHookEffects(() => first?.open()); + await flushHookEffects(() => first?.close()); + + await flushHookEffects(() => vi.advanceTimersByTime(1_000)); + await flushHookEffects(() => FakeWebSocket.instances[1]?.close()); + await flushHookEffects(() => vi.advanceTimersByTime(28_999)); + expect(callbacks.onFallbackPoll).toHaveBeenCalledOnce(); + + await flushHookEffects(() => vi.advanceTimersByTime(1)); + expect(callbacks.onFallbackPoll).toHaveBeenCalledTimes(2); + await hook.unmount(); + }); + it("reports unavailable when both live events and fallback sync fail", async () => { const callbacks = handlers(); callbacks.onFallbackPoll.mockRejectedValue(new Error("API unavailable"));