diff --git a/api/hqbase-mail-api-v1.openapi.json b/api/hqbase-mail-api-v1.openapi.json index 09bc1625..45b66067 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`. 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"] + }, + { + "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..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", 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/app.tsx b/app/app.tsx index 7979b658..41344419 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,32 @@ 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 () => { + 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]); + + const connectionStatus = useMailEvents(currentUserId, { + onDrafts: draftState.refresh, + onFallbackPoll: refreshRealtimeState, + onMailboxes: refreshRealtimeState, + onMessages: mailSync.refresh, + onReconnect: refreshRealtimeState + }); + React.useEffect(() => { if (publicAuthenticationPath) return; void reload(); @@ -192,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 8f8003fe..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 = 10_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 new file mode 100644 index 00000000..d1c2cf84 --- /dev/null +++ b/app/features/events/use-mail-events.ts @@ -0,0 +1,289 @@ +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; +}; + +type MailEvent = { + topic: MailEventTopic; + type: "changed"; +}; + +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; + }, [handlers]); + + React.useEffect(() => { + if (!userId) return; + + 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(); + 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 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) { + current.close(1000, "Connection paused."); + } + }; + + 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() || fallbackTimer !== null || 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; + + const url = new URL("/api/v1/events", window.location.href); + 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); + }); + next.addEventListener("error", () => { + if (socket === next && next.readyState < WebSocket.CLOSING) next.close(); + }); + next.addEventListener("close", () => { + if (socket !== next) return; + socket = null; + clearConnectionTimer(); + clearHeartbeatTimers(); + if (!canConnect()) { + if (navigator.onLine === false) setStatus("unavailable"); + return; + } + runFallbackPoll(); + scheduleReconnect(); + }); + }; + const resume = (): void => { + if (reconnectTimer !== null) { + window.clearTimeout(reconnectTimer); + reconnectTimer = null; + } + setStatus("connecting"); + connect(); + }; + const handleVisibilityChange = (): void => { + if (document.visibilityState === "visible") resume(); + else closeSocket(); + }; + const handleOnline = (): void => resume(); + 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); + window.removeEventListener("offline", handleOffline); + closeSocket(); + }; + }, [userId]); + + return status; +} + +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..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 = 10_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/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..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" }, @@ -83,6 +87,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 +181,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}}}` })), @@ -209,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/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..db4f2aaa 100644 --- a/test/integration/worker/mail-api.test.ts +++ b/test/integration/worker/mail-api.test.ts @@ -1,7 +1,8 @@ -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 { mailEventInternalHeaders } from "../../../worker/features/events/durable-object"; import { applyCurrentMigrations } from "./current-migrations"; import { tokenRow } from "./mail-api-token-fixture"; @@ -329,6 +330,148 @@ 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); + 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" + }); + await closeEventSocket(sessionSocket); + + const messageSocket = await openEventSocket({ + 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", + userIds: [userId] + }); + await expect(messageFrame).resolves.toEqual({ type: "changed", topic: "messages" }); + await closeEventSocket(messageSocket); + + 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" }); + await closeEventSocket(draftSocket); + }); + + 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 thirdSocket = await openEventSocket(headers); + const clientCloses = [firstSocket, secondSocket, thirdSocket].map(nextSocketClose); + const stub = env.MAIL_EVENTS.getByName("workspace"); + + 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 }; + }); + + 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 () => { + 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); @@ -363,11 +506,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" }); + await closeEventSocket(socket); const unarchived = await apiFetch("/api/v1/messages/msg_api/unarchive", writeToken, { method: "POST" @@ -882,3 +1032,70 @@ 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 } + ); + }); +} + +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( + "close", + (event) => resolve({ code: event.code, reason: event.reason }), + { once: true } + ); + }); +} + +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/app/events/use-mail-events.test.tsx b/test/unit/app/events/use-mail-events.test.tsx new file mode 100644 index 00000000..69262792 --- /dev/null +++ b/test/unit/app/events/use-mail-events.test.tsx @@ -0,0 +1,230 @@ +// @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")); + }); + + send = vi.fn(); + + 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(), + onFallbackPoll: 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"); + expect(hook.result).toBe("connecting"); + await flushHookEffects(() => socket?.open()); + expect(hook.result).toBe("connected"); + 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(); + }); + + 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("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")); + 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( { + 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..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,6 +95,17 @@ describe("HQBase installation resources", () => { } ] }); + 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 129ed087..168036bc 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,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("{{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" @@ -62,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/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/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 4da7d814..84b67a7a 100644 --- a/test/unit/worker/features/send/send-service.test.ts +++ b/test/unit/worker/features/send/send-service.test.ts @@ -11,7 +11,6 @@ vi.mock("@worker/features/mailboxes/queries", () => ({ vi.mock("@worker/features/mailboxes/address-queries", () => ({ findAddressIdentity: vi.fn().mockResolvedValue(null) })); - vi.mock("@worker/features/messages/queries", () => ({ getMessageDetail: vi.fn(), getMessageHtmlKey: vi.fn(), @@ -76,6 +75,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..802353e7 --- /dev/null +++ b/worker/features/events/durable-object.ts @@ -0,0 +1,166 @@ +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 { + 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 }); + } + + 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 }); + } + + let connections = this.liveConnections(Date.now()); + const userTag = `user:${userId}`; + const existing = connections.filter((socket) => readConnection(socket)?.userId === userId); + if (existing.length >= maxConnectionsPerUser) { + 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(); + 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); + await this.scheduleExpiryAlarm(connection.expiresAt); + + 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."); + } + + 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()) { + if (socket.readyState !== WebSocket.OPEN) continue; + 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 = { + 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..2c1b41d7 --- /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 !== 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..4e45480c --- /dev/null +++ b/worker/features/events/service.ts @@ -0,0 +1,154 @@ +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"; +const publishAttempts = 3; +const publishRetryBaseDelayMs = 100; + +export type MailEventScheduler = (promise: Promise) => void; + +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); +} + +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")); + } +} + +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[], + topic: MailEventTopic +): Promise { + const recipients = [...new Set(userIds)]; + if (recipients.length === 0) return; + await retryMailEventPublish(() => + 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) { + // 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 + 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..b1f27cff 100644 --- a/worker/features/mcp/draft-tools.ts +++ b/worker/features/mcp/draft-tools.ts @@ -13,6 +13,11 @@ import { saveDraft } from "../drafts/queries"; import { draftSchema } from "../drafts/validation"; +import { + type MailEventScheduler, + publishUserMailEvent, + scheduleMailEvent +} from "../events/service"; import type { McpPrincipal } from "./route"; import { base64File, maxMcpAttachmentBase64Length, toolResult } from "./tool-result"; @@ -34,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; @@ -73,6 +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); + notifyDraftChange(env, principal.userId, schedule); return draft; }) ); @@ -110,6 +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); + notifyDraftChange(env, principal.userId, schedule); return draft; }) ); @@ -126,17 +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); + 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", @@ -173,6 +183,7 @@ function registerDraftAttachmentTools( httpMetadata: { contentType: added.attachment.contentType } }); await recordDraftMutation(env, principal, "mcp.draft.attachment.add", added.attachment.id); + notifyDraftChange(env, principal.userId, schedule); return added.attachment; }) ); @@ -202,6 +213,7 @@ function registerDraftAttachmentTools( throw new AppError("ATTACHMENT_NOT_FOUND", "Attachment not found.", 404); } await recordDraftMutation(env, principal, "mcp.draft.attachment.remove", attachmentId); + notifyDraftChange(env, principal.userId, schedule); return { deleted: true, attachmentId, draftId }; }) ); @@ -223,3 +235,7 @@ function recordDraftMutation( outcome: "success" }); } + +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 f93e7acf..fbce732f 100644 --- a/worker/features/mcp/mail-tools.ts +++ b/worker/features/mcp/mail-tools.ts @@ -5,6 +5,12 @@ import { accessibleMessageScope } from "../../auth/mailbox-access"; import type { WorkerEnv } from "../../lib/env"; import { AppError } from "../../lib/errors"; import { recordAudit } from "../audit/service"; +import { + type MailEventScheduler, + messageEventTarget, + publishMessageMailEvent, + scheduleMailEvent +} from "../events/service"; import { listMailboxesForUser } from "../mailboxes/queries"; import { requireAttachmentAccess, requireMessageAccess } from "../messages/access"; import { listConversations, updateConversationAction } from "../messages/conversation-queries"; @@ -36,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 { @@ -177,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", { @@ -193,6 +205,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) { + scheduleMailEvent(schedule, publishMessageMailEvent(env, [target])); + } await recordMutation(env, principal, `mcp.message.${action}`, "message", messageId); return message; }) @@ -219,12 +235,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) { + scheduleMailEvent(schedule, publishMessageMailEvent(env, eventTargets)); + } await recordMutation( env, principal, 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/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/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/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": [