From cee9f95d6d5e56dc451803381305af5414c04a33 Mon Sep 17 00:00:00 2001 From: Raj D <25481060+radroid@users.noreply.github.com> Date: Mon, 27 Jul 2026 00:30:26 -0400 Subject: [PATCH 1/2] feat(web): closed-tab Web Push notifications (SW + PWA + server VAPID push) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the accepted v1 gap from #14: notifications previously only fired while a T3 tab was open. This delivers a push when the tab is fully closed. Client (apps/web): - PWA manifest + icons + installability (required for iOS Safari Web Push). - Push-only service worker (no fetch handler, so no offline/stale-asset risk): shows a notification on push, focuses/opens the waiting thread on click, and suppresses when any T3 tab is open so the in-page NotificationCoordinator stays the single source while a tab is alive. - PushSubscriptionManager keeps the subscription in sync with the existing notifyOnNeedsInput setting + permission (reacts to a post-mount grant via the Permissions API). Pure, tested VAPID/subscription helpers. Server (apps/server/src/t3x/webPush — fork-seam clean, no upstream edits): - Reactor taps OrchestrationEngine.streamDomainEvents, recomputes the awareness phase with the shared projectThreadAwareness, edge-detects the same transitions the web client does (waiting_for_input/approval; running-> completed), and sends Web Push to registered subscriptions, pruning expired (404/410) ones. - JSON subscription store in the state dir; VAPID keypair via ServerSecretStore (T3X_VAPID_* env override supported); raw /api/t3x/push/{subscribe,unsubscribe,vapid-public-key} routes mounted through the existing T3xRoutesLive seam. Verified: apps/web + apps/server typecheck clean (0 errors); web + server unit tests pass; web-push loads at runtime; server+bin construction tests (277) pass. Browser end-to-end (grant permission, close tab, receive push) is manual QA. Closes #19 Co-Authored-By: Claude Opus 4.8 --- apps/server/package.json | 2 + apps/server/src/t3x/index.ts | 38 ++++- apps/server/src/t3x/webPush/Reactor.ts | 127 ++++++++++++++ apps/server/src/t3x/webPush/attention.test.ts | 113 +++++++++++++ apps/server/src/t3x/webPush/attention.ts | 99 +++++++++++ apps/server/src/t3x/webPush/config.ts | 44 +++++ apps/server/src/t3x/webPush/http.ts | 157 ++++++++++++++++++ apps/server/src/t3x/webPush/send.ts | 65 ++++++++ apps/server/src/t3x/webPush/state.ts | 127 ++++++++++++++ apps/server/src/t3x/webPush/vapid.ts | 91 ++++++++++ apps/web/index.html | 5 + apps/web/public/icon-192.png | Bin 0 -> 18303 bytes apps/web/public/icon-512.png | Bin 0 -> 138991 bytes apps/web/public/manifest.webmanifest | 17 ++ apps/web/public/sw.js | 90 ++++++++++ .../components/PushSubscriptionManager.tsx | 64 +++++++ apps/web/src/notifications/push.logic.test.ts | 62 +++++++ apps/web/src/notifications/push.logic.ts | 45 +++++ apps/web/src/notifications/push.ts | 143 ++++++++++++++++ apps/web/src/notifications/serviceWorker.ts | 31 ++++ apps/web/src/routes/__root.tsx | 2 + pnpm-lock.yaml | 83 +++++++++ 22 files changed, 1403 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/t3x/webPush/Reactor.ts create mode 100644 apps/server/src/t3x/webPush/attention.test.ts create mode 100644 apps/server/src/t3x/webPush/attention.ts create mode 100644 apps/server/src/t3x/webPush/config.ts create mode 100644 apps/server/src/t3x/webPush/http.ts create mode 100644 apps/server/src/t3x/webPush/send.ts create mode 100644 apps/server/src/t3x/webPush/state.ts create mode 100644 apps/server/src/t3x/webPush/vapid.ts create mode 100644 apps/web/public/icon-192.png create mode 100644 apps/web/public/icon-512.png create mode 100644 apps/web/public/manifest.webmanifest create mode 100644 apps/web/public/sw.js create mode 100644 apps/web/src/components/PushSubscriptionManager.tsx create mode 100644 apps/web/src/notifications/push.logic.test.ts create mode 100644 apps/web/src/notifications/push.logic.ts create mode 100644 apps/web/src/notifications/push.ts create mode 100644 apps/web/src/notifications/serviceWorker.ts diff --git a/apps/server/package.json b/apps/server/package.json index 6be498e803d..9aa8314d1af 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -32,6 +32,7 @@ "@pierre/diffs": "catalog:", "effect": "catalog:", "node-pty": "^1.1.0", + "web-push": "^3.6.7", "yaml": "catalog:" }, "devDependencies": { @@ -42,6 +43,7 @@ "@t3tools/web": "workspace:*", "@types/bun": "1.3.14", "@types/node": "catalog:", + "@types/web-push": "^3.6.4", "effect-acp": "workspace:*", "effect-codex-app-server": "workspace:*", "vite-plus": "catalog:" diff --git a/apps/server/src/t3x/index.ts b/apps/server/src/t3x/index.ts index f98bc956f4d..9eda72e3cd9 100644 --- a/apps/server/src/t3x/index.ts +++ b/apps/server/src/t3x/index.ts @@ -22,8 +22,14 @@ import { ServerConfig } from "../config.ts"; import { autoResumeRouteLayer } from "./autoResume/http.ts"; import { AutoResumeReactorLive } from "./autoResume/Reactor.ts"; import { AutoResumeStore, makeAutoResumeStore } from "./autoResume/state.ts"; +import { resolveConfig as resolveWebPushConfig } from "./webPush/config.ts"; +import { webPushRouteLayer } from "./webPush/http.ts"; +import { WebPushReactorLive } from "./webPush/Reactor.ts"; +import { makePushSubscriptionStore, PushSubscriptionStore } from "./webPush/state.ts"; +import { makeWebPushVapid, WebPushVapid } from "./webPush/vapid.ts"; const AUTO_RESUME_STATE_FILENAME = "t3x-auto-resume.json"; +const WEB_PUSH_STATE_FILENAME = "t3x-web-push-subscriptions.json"; /** Wires the durable store to the server state directory. */ const AutoResumeStoreLive = Layer.effect( @@ -35,12 +41,37 @@ const AutoResumeStoreLive = Layer.effect( }), ); +/** Web Push subscription store, wired to the server state directory. */ +const PushSubscriptionStoreLive = Layer.effect( + PushSubscriptionStore, + Effect.gen(function* () { + const config = yield* ServerConfig; + const path = yield* Path.Path; + return yield* makePushSubscriptionStore(path.join(config.stateDir, WEB_PUSH_STATE_FILENAME)); + }), +); + +/** VAPID keypair (env override, else generated + persisted in the secret store). */ +const WebPushVapidLive = Layer.effect(WebPushVapid, makeWebPushVapid(resolveWebPushConfig())); + +/** + * Shared fork-local deps for the Web Push reactor AND routes. Defined once at module scope so + * both `T3xLayerLive` and `T3xRoutesLive` reference the SAME layer value — Effect memoises + * construction by layer identity across the shared MemoMap, so the reactor and the subscribe + * route mutate one store/keypair rather than racing two copies over a single file (the same + * property `AutoResumeStoreLive` relies on above). + */ +const WebPushDepsLive = Layer.mergeAll(PushSubscriptionStoreLive, WebPushVapidLive); + /** * The single fork-local layer merged into the server. The auto-resume supervisor * self-starts on construction; its store is provided here so `server.ts` merges only * this one layer. */ -export const T3xLayerLive = AutoResumeReactorLive.pipe(Layer.provide(AutoResumeStoreLive)); +export const T3xLayerLive = Layer.mergeAll( + AutoResumeReactorLive.pipe(Layer.provide(AutoResumeStoreLive)), + WebPushReactorLive.pipe(Layer.provide(WebPushDepsLive)), +); /** * All fork-local HTTP routes, fanned in here for the same reason as `T3xLayerLive`: @@ -66,4 +97,7 @@ export const T3xLayerLive = AutoResumeReactorLive.pipe(Layer.provide(AutoResumeS * probes rather than importing these two layers, so treat it as a guard on the * assumption, not proof of this file's graph. */ -export const T3xRoutesLive = autoResumeRouteLayer.pipe(Layer.provide(AutoResumeStoreLive)); +export const T3xRoutesLive = Layer.mergeAll( + autoResumeRouteLayer.pipe(Layer.provide(AutoResumeStoreLive)), + webPushRouteLayer.pipe(Layer.provide(WebPushDepsLive)), +); diff --git a/apps/server/src/t3x/webPush/Reactor.ts b/apps/server/src/t3x/webPush/Reactor.ts new file mode 100644 index 00000000000..6e9c28645e3 --- /dev/null +++ b/apps/server/src/t3x/webPush/Reactor.ts @@ -0,0 +1,127 @@ +/** + * WebPushReactor — closed-tab notification supervisor. + * + * Self-starts one scoped fiber at layer construction (no external `.start()`, so the only + * upstream seam stays the lines already in server.ts). It taps the hot orchestration + * domain-event stream, and for each thread event recomputes the awareness phase from the + * projected shell (the SAME `projectThreadAwareness` the web client uses), edge-detects, and + * on a firing transition sends a Web Push to every registered subscription. Dead + * subscriptions (404/410) are pruned as they surface. + * + * De-dup with the in-page coordinator is the service worker's job: it suppresses when a tab + * is open, so this reactor always sends and the worker only shows when no tab exists. + * + * @module t3x/webPush/Reactor + */ + +import type { OrchestrationEvent, OrchestrationProjectShell, ThreadId } from "@t3tools/contracts"; +import { projectThreadAwareness } from "@t3tools/shared/agentAwareness"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; + +import { ServerEnvironment } from "../../environment/ServerEnvironment.ts"; +import { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { buildAttentionPayload, createAttentionEdgeTracker } from "./attention.ts"; +import { resolveConfig } from "./config.ts"; +import { sendWebPush } from "./send.ts"; +import { PushSubscriptionStore } from "./state.ts"; +import { WebPushVapid } from "./vapid.ts"; + +const SEND_CONCURRENCY = 4; + +/** Local mirror of the relay's `eventThreadId`, kept in-seam to avoid coupling to upstream. */ +function eventThreadId(event: OrchestrationEvent): ThreadId | null { + const payload = event.payload as { readonly threadId?: unknown }; + if (typeof payload.threadId === "string") { + return payload.threadId as ThreadId; + } + if (event.aggregateKind === "thread" && typeof event.aggregateId === "string") { + return event.aggregateId as ThreadId; + } + return null; +} + +const makeSupervisor = Effect.gen(function* () { + const config = resolveConfig(); + if (!config.enabled) { + yield* Effect.logInfo("t3x web-push: disabled via T3X_WEB_PUSH_ENABLED"); + return; + } + + const engine = yield* OrchestrationEngineService; + const snapshotQuery = yield* ProjectionSnapshotQuery; + const serverEnvironment = yield* ServerEnvironment; + const store = yield* PushSubscriptionStore; + const vapid = yield* WebPushVapid; + const tracker = createAttentionEdgeTracker(); + + const handleThread = (threadId: ThreadId) => + Effect.gen(function* () { + // No subscribers → no work. The tracker only builds history while subscribed, which is + // fine: a subscription added mid-session is first-seen and so never replays a stale edge. + const subscriptions = yield* store.list; + if (subscriptions.length === 0) { + return; + } + + const threadOpt = yield* snapshotQuery.getThreadShellById(threadId); + if (Option.isNone(threadOpt)) { + tracker.forget(threadId); + return; + } + const thread = threadOpt.value; + const projectOpt = yield* snapshotQuery.getProjectShellById(thread.projectId); + const environmentId = yield* serverEnvironment.getEnvironmentId; + + const projectTitle = ( + Option.isSome(projectOpt) ? projectOpt.value.title : "T3 Code" + ) as OrchestrationProjectShell["title"]; + + const state = projectThreadAwareness({ + environmentId, + project: { title: projectTitle }, + thread, + }); + const kind = tracker.observe(threadId, state?.phase ?? null); + if (kind === null || state === null) { + return; + } + + const payload = buildAttentionPayload(state, kind); + yield* Effect.forEach( + subscriptions, + (subscription) => + sendWebPush(vapid, subscription, payload).pipe( + Effect.flatMap((result) => + result.expired ? store.removeByEndpoint(subscription.endpoint) : Effect.void, + ), + ), + { concurrency: SEND_CONCURRENCY, discard: true }, + ); + }); + + yield* Effect.forkScoped( + Stream.runForEach(engine.streamDomainEvents, (event) => { + const threadId = eventThreadId(event); + if (threadId === null) { + return Effect.void; + } + return handleThread(threadId).pipe( + Effect.catchCause((cause) => + Effect.logWarning("t3x web-push: notification handler failed", { + eventType: event.type, + cause: Cause.pretty(cause), + }), + ), + ); + }), + ); + + yield* Effect.logInfo("t3x web-push: reactor started"); +}); + +export const WebPushReactorLive = Layer.effectDiscard(makeSupervisor); diff --git a/apps/server/src/t3x/webPush/attention.test.ts b/apps/server/src/t3x/webPush/attention.test.ts new file mode 100644 index 00000000000..a7086ee4e9e --- /dev/null +++ b/apps/server/src/t3x/webPush/attention.test.ts @@ -0,0 +1,113 @@ +import type { AgentAwarenessState } from "@t3tools/shared/agentAwareness"; +import { describe, expect, it } from "vite-plus/test"; + +import { + attentionKey, + attentionKindForEdge, + buildAttentionPayload, + createAttentionEdgeTracker, +} from "./attention.ts"; + +describe("attentionKindForEdge", () => { + it("announces waiting phases on entry regardless of the previous phase", () => { + expect(attentionKindForEdge("running", "waiting_for_approval")).toBe("waiting_for_approval"); + expect(attentionKindForEdge("completed", "waiting_for_input")).toBe("waiting_for_input"); + }); + + it("announces completed only out of running", () => { + expect(attentionKindForEdge("running", "completed")).toBe("completed"); + expect(attentionKindForEdge("starting", "completed")).toBeNull(); + expect(attentionKindForEdge(null, "completed")).toBeNull(); + }); + + it("stays silent for non-attention phases", () => { + expect(attentionKindForEdge("waiting_for_input", "running")).toBeNull(); + expect(attentionKindForEdge("running", "starting")).toBeNull(); + expect(attentionKindForEdge("running", "failed")).toBeNull(); + expect(attentionKindForEdge("running", "stale")).toBeNull(); + }); +}); + +describe("createAttentionEdgeTracker", () => { + it("never fires on a thread's first observation", () => { + const tracker = createAttentionEdgeTracker(); + expect(tracker.observe("t1", "waiting_for_input")).toBeNull(); + }); + + it("fires completed only on running -> completed", () => { + const tracker = createAttentionEdgeTracker(); + expect(tracker.observe("t1", "running")).toBeNull(); // first-seen + expect(tracker.observe("t1", "completed")).toBe("completed"); + }); + + it("does not fire completed when the thread was never seen running", () => { + const tracker = createAttentionEdgeTracker(); + expect(tracker.observe("t1", "starting")).toBeNull(); // first-seen + expect(tracker.observe("t1", "completed")).toBeNull(); // starting -> completed + }); + + it("fires waiting_for_input on entry after a prior observation", () => { + const tracker = createAttentionEdgeTracker(); + expect(tracker.observe("t1", "running")).toBeNull(); + expect(tracker.observe("t1", "waiting_for_input")).toBe("waiting_for_input"); + }); + + it("does not re-fire while the phase is unchanged", () => { + const tracker = createAttentionEdgeTracker(); + tracker.observe("t1", "running"); + expect(tracker.observe("t1", "waiting_for_approval")).toBe("waiting_for_approval"); + expect(tracker.observe("t1", "waiting_for_approval")).toBeNull(); + }); + + it("treats a null phase as an observation that never fires", () => { + const tracker = createAttentionEdgeTracker(); + tracker.observe("t1", "running"); + expect(tracker.observe("t1", null)).toBeNull(); + // Coming back to completed from a null observation is not a running->completed edge. + expect(tracker.observe("t1", "completed")).toBeNull(); + }); + + it("tracks threads independently", () => { + const tracker = createAttentionEdgeTracker(); + tracker.observe("t1", "running"); + expect(tracker.observe("t2", "completed")).toBeNull(); // t2 first-seen + expect(tracker.observe("t1", "completed")).toBe("completed"); + }); + + it("re-arms first-seen after forget", () => { + const tracker = createAttentionEdgeTracker(); + tracker.observe("t1", "running"); + tracker.forget("t1"); + // First observation again -> no fire even though it's a completed value. + expect(tracker.observe("t1", "completed")).toBeNull(); + }); +}); + +describe("buildAttentionPayload", () => { + it("maps an awareness state + edge into the service-worker payload", () => { + const state = { + environmentId: "env-1", + threadId: "thread-1", + projectTitle: "My Project", + threadTitle: "Fix the bug", + phase: "waiting_for_input", + headline: "Waiting for input", + modelTitle: "claude", + updatedAt: "2026-07-27T00:00:00.000Z", + deepLink: "/threads/env-1/thread-1", + } as unknown as AgentAwarenessState; + + expect(buildAttentionPayload(state, "waiting_for_input")).toEqual({ + title: "Waiting for input", + body: "My Project · Fix the bug", + key: "env-1::thread-1", + environmentId: "env-1", + threadId: "thread-1", + kind: "waiting_for_input", + }); + }); + + it("keys by environment + thread", () => { + expect(attentionKey("env-1", "thread-1")).toBe("env-1::thread-1"); + }); +}); diff --git a/apps/server/src/t3x/webPush/attention.ts b/apps/server/src/t3x/webPush/attention.ts new file mode 100644 index 00000000000..3ae6b0fdc2e --- /dev/null +++ b/apps/server/src/t3x/webPush/attention.ts @@ -0,0 +1,99 @@ +/** + * Server-side attention edge detection for Web Push. + * + * A local port of the web client's `apps/web/src/notifications/needsAttention.logic.ts`, so + * closed-tab push announces the exact same transitions the in-page coordinator does. It + * edge-detects on the COMPUTED phase (via `projectThreadAwareness`) rather than raw event + * types, which is inherently robust against the premature-"Done" problem: a stale + * `completed` observed while a new turn is being requested equals the previous `completed` + * and so never fires. Deliberately duplicated across the fork seam (a ~20-line pure function) + * rather than importing web-app internals into the server. + * + * @module t3x/webPush/attention + */ + +import type { AgentAwarenessPhase, AgentAwarenessState } from "@t3tools/shared/agentAwareness"; + +export type AttentionKind = "waiting_for_approval" | "waiting_for_input" | "completed"; + +/** The push payload the service worker renders; mirrors the web `AttentionEvent`. */ +export interface AttentionPushPayload { + readonly title: string; + readonly body: string; + readonly key: string; + readonly environmentId: string; + readonly threadId: string; + readonly kind: AttentionKind; +} + +export function attentionKey(environmentId: string, threadId: string): string { + return `${environmentId}::${threadId}`; +} + +/** + * Which transitions announce. Same two rules as the web tracker: `waiting_*` announce on + * entry; `completed` announces only out of `running` (threads are born reading as + * `completed`, so any other rule announces a "Done" for work that never ran). + */ +export function attentionKindForEdge( + previousPhase: AgentAwarenessPhase | null, + phase: AgentAwarenessPhase, +): AttentionKind | null { + switch (phase) { + case "waiting_for_approval": + return "waiting_for_approval"; + case "waiting_for_input": + return "waiting_for_input"; + case "completed": + return previousPhase === "running" ? "completed" : null; + default: + return null; + } +} + +export interface AttentionEdgeTracker { + /** Feed a thread's latest phase (null when unresolved); returns a kind on a firing edge. */ + readonly observe: (threadId: string, phase: AgentAwarenessPhase | null) => AttentionKind | null; + /** Drop a thread's tracked phase so a re-created thread is first-seen again. */ + readonly forget: (threadId: string) => void; +} + +/** + * Edge-detects phases per thread. A thread's first observation never fires (it replays state + * that was already true), matching the web tracker so a server restart does not re-announce + * every thread that happens to be waiting. + */ +export function createAttentionEdgeTracker(): AttentionEdgeTracker { + const phases = new Map(); + return { + observe(threadId, phase) { + const wasTracked = phases.has(threadId); + const previous = phases.get(threadId) ?? null; + phases.set(threadId, phase); + if (!wasTracked || phase === previous || phase === null) { + return null; + } + return attentionKindForEdge(previous, phase); + }, + forget(threadId) { + phases.delete(threadId); + }, + }; +} + +/** Builds the push payload from an awareness state + the firing edge. */ +export function buildAttentionPayload( + state: AgentAwarenessState, + kind: AttentionKind, +): AttentionPushPayload { + const environmentId = String(state.environmentId); + const threadId = String(state.threadId); + return { + title: state.headline, + body: `${state.projectTitle} · ${state.threadTitle}`, + key: attentionKey(environmentId, threadId), + environmentId, + threadId, + kind, + }; +} diff --git a/apps/server/src/t3x/webPush/config.ts b/apps/server/src/t3x/webPush/config.ts new file mode 100644 index 00000000000..d01f3b381fa --- /dev/null +++ b/apps/server/src/t3x/webPush/config.ts @@ -0,0 +1,44 @@ +/** + * Web Push configuration. + * + * Read from env with safe defaults so the feature works with zero setup (the VAPID keypair + * is generated + persisted on first boot; see vapid.ts). `resolveConfig` is pure (env passed + * in) for testability, mirroring t3x/autoResume/config.ts. + * + * @module t3x/webPush/config + */ + +export interface WebPushConfig { + readonly enabled: boolean; + /** VAPID `sub` claim — a mailto: or https: contact URL for the push service. */ + readonly subject: string; + /** Explicit VAPID keys via env; when both are set they win over the secret store. */ + readonly publicKeyOverride: string | null; + readonly privateKeyOverride: string | null; +} + +const DEFAULT_SUBJECT = "mailto:web-push@t3code.local"; + +function parseBool(value: string | undefined, fallback: boolean): boolean { + if (value === undefined) return fallback; + const v = value.trim().toLowerCase(); + if (v === "false" || v === "0" || v === "no" || v === "off") return false; + if (v === "true" || v === "1" || v === "yes" || v === "on") return true; + return fallback; +} + +function nonEmpty(value: string | undefined): string | null { + const trimmed = value?.trim(); + return trimmed ? trimmed : null; +} + +export function resolveConfig( + env: Record = process.env, +): WebPushConfig { + return { + enabled: parseBool(env.T3X_WEB_PUSH_ENABLED, true), + subject: nonEmpty(env.T3X_VAPID_SUBJECT) ?? DEFAULT_SUBJECT, + publicKeyOverride: nonEmpty(env.T3X_VAPID_PUBLIC_KEY), + privateKeyOverride: nonEmpty(env.T3X_VAPID_PRIVATE_KEY), + }; +} diff --git a/apps/server/src/t3x/webPush/http.ts b/apps/server/src/t3x/webPush/http.ts new file mode 100644 index 00000000000..7439739628d --- /dev/null +++ b/apps/server/src/t3x/webPush/http.ts @@ -0,0 +1,157 @@ +/** + * Fork-owned raw HTTP routes backing Web Push subscription management. + * + * GET /api/t3x/push/vapid-public-key -> { publicKey } (read scope) + * POST /api/t3x/push/subscribe -> { ok: true } (operate scope) + * POST /api/t3x/push/unsubscribe -> { ok: true } (operate scope) + * + * Raw routes, not WS-RPC: an RPC would force edits to `@t3tools/contracts` + `ws.ts` + its + * scope map. These mount via `T3xRoutesLive` in t3x/index.ts, so server.ts is untouched. + * See docs/t3x/SEAMS.md and t3x/autoResume/http.ts (the mirrored template). + * + * @module t3x/webPush/http + */ + +import { + type AuthEnvironmentScope, + AuthOrchestrationOperateScope, + AuthOrchestrationReadScope, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import { + HttpRouter, + HttpServerRequest, + HttpServerRespondable, + HttpServerResponse, +} from "effect/unstable/http"; + +import * as EnvironmentAuth from "../../auth/EnvironmentAuth.ts"; +import { + failEnvironmentAuthInvalid, + failEnvironmentInternal, + failEnvironmentScopeRequired, +} from "../../auth/http.ts"; +import { PushSubscriptionStore, type PushSubscriptionStoreShape } from "./state.ts"; +import { WebPushVapid, type WebPushVapidKeys } from "./vapid.ts"; + +export const VAPID_KEY_ROUTE_PATH = "/api/t3x/push/vapid-public-key"; +export const SUBSCRIBE_ROUTE_PATH = "/api/t3x/push/subscribe"; +export const UNSUBSCRIBE_ROUTE_PATH = "/api/t3x/push/unsubscribe"; + +/** + * MIRROR of the module-private raw-route auth in apps/server/src/http.ts (also mirrored by + * t3x/autoResume/http.ts). Returns the session so the subscribe route can record which device + * registered. Registered as a logic mirror in docs/t3x/SEAMS.md. + */ +const authenticateWithScope = (scope: AuthEnvironmentScope) => + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; + const session = yield* serverAuth.authenticateHttpRequest(request).pipe( + Effect.catchIf(EnvironmentAuth.isServerAuthCredentialError, (error) => + failEnvironmentAuthInvalid(EnvironmentAuth.serverAuthCredentialReason(error)), + ), + Effect.catchIf(EnvironmentAuth.isServerAuthInternalError, (error) => + failEnvironmentInternal("internal_error", error), + ), + ); + if (!session.scopes.includes(scope)) { + return yield* failEnvironmentScopeRequired(scope); + } + return session; + }); + +const respondableTags = { + EnvironmentAuthInvalidError: HttpServerRespondable.toResponse, + EnvironmentInternalError: HttpServerRespondable.toResponse, + EnvironmentScopeRequiredError: HttpServerRespondable.toResponse, +} as const; + +const SubscribeBody = Schema.Struct({ + endpoint: Schema.String, + keys: Schema.Struct({ + p256dh: Schema.String, + auth: Schema.String, + }), +}); +const decodeSubscribeBody = Schema.decodeUnknownEffect(SubscribeBody); + +const UnsubscribeBody = Schema.Struct({ + endpoint: Schema.String, +}); +const decodeUnsubscribeBody = Schema.decodeUnknownEffect(UnsubscribeBody); + +const makeVapidKeyRoute = (vapid: WebPushVapidKeys) => + HttpRouter.add( + "GET", + VAPID_KEY_ROUTE_PATH, + Effect.gen(function* () { + yield* authenticateWithScope(AuthOrchestrationReadScope); + return HttpServerResponse.jsonUnsafe({ publicKey: vapid.publicKey }); + }).pipe(Effect.catchTags(respondableTags)), + ); + +const makeSubscribeRoute = (store: PushSubscriptionStoreShape) => + HttpRouter.add( + "POST", + SUBSCRIBE_ROUTE_PATH, + Effect.gen(function* () { + const session = yield* authenticateWithScope(AuthOrchestrationOperateScope); + const request = yield* HttpServerRequest.HttpServerRequest; + const body = yield* decodeSubscribeBody(yield* request.json).pipe( + Effect.map((decoded): typeof SubscribeBody.Type | null => decoded), + Effect.orElseSucceed(() => null), + ); + if (body === null || body.endpoint === "") { + return HttpServerResponse.text("Invalid body", { status: 400 }); + } + const createdAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); + yield* store.upsert({ + endpoint: body.endpoint, + p256dh: body.keys.p256dh, + auth: body.keys.auth, + sessionId: String(session.sessionId), + createdAt, + }); + return HttpServerResponse.jsonUnsafe({ ok: true }); + }).pipe(Effect.catchTags(respondableTags)), + ); + +const makeUnsubscribeRoute = (store: PushSubscriptionStoreShape) => + HttpRouter.add( + "POST", + UNSUBSCRIBE_ROUTE_PATH, + Effect.gen(function* () { + yield* authenticateWithScope(AuthOrchestrationOperateScope); + const request = yield* HttpServerRequest.HttpServerRequest; + const body = yield* decodeUnsubscribeBody(yield* request.json).pipe( + Effect.map((decoded): typeof UnsubscribeBody.Type | null => decoded), + Effect.orElseSucceed(() => null), + ); + if (body === null || body.endpoint === "") { + return HttpServerResponse.text("Invalid body", { status: 400 }); + } + yield* store.removeByEndpoint(body.endpoint); + return HttpServerResponse.jsonUnsafe({ ok: true }); + }).pipe(Effect.catchTags(respondableTags)), + ); + +/** + * Mounted from `T3xRoutesLive`. `Layer.unwrap` resolves the store + VAPID once at layer + * construction and the handlers close over the values, so neither requirement propagates into + * upstream's `makeRoutesLayer` signature (see t3x/autoResume/http.ts for the full rationale). + */ +export const webPushRouteLayer = Layer.unwrap( + Effect.gen(function* () { + const store = yield* PushSubscriptionStore; + const vapid = yield* WebPushVapid; + return Layer.mergeAll( + makeVapidKeyRoute(vapid), + makeSubscribeRoute(store), + makeUnsubscribeRoute(store), + ); + }), +); diff --git a/apps/server/src/t3x/webPush/send.ts b/apps/server/src/t3x/webPush/send.ts new file mode 100644 index 00000000000..9586c278453 --- /dev/null +++ b/apps/server/src/t3x/webPush/send.ts @@ -0,0 +1,65 @@ +/** + * Web Push delivery — a thin Effect wrapper over the `web-push` library. + * + * Isolated in one module so the CommonJS dependency and its VAPID/AES128GCM encryption are + * easy to swap or mock. Never fails its Effect: a rejected send resolves to a result flagging + * whether the subscription is gone (404/410) so the caller can prune it. + * + * @module t3x/webPush/send + */ + +import * as Effect from "effect/Effect"; +import webpush from "web-push"; + +import type { AttentionPushPayload } from "./attention.ts"; +import type { PushSubscriptionRecord } from "./state.ts"; +import type { WebPushVapidKeys } from "./vapid.ts"; + +/** How long the push service holds an undelivered message. A needs-you alert is stale after a few hours. */ +const PUSH_TTL_SECONDS = 4 * 60 * 60; + +export interface WebPushSendResult { + readonly ok: boolean; + /** True when the push service reports the subscription is gone (404/410) — prune it. */ + readonly expired: boolean; +} + +function extractStatusCode(error: unknown): number | undefined { + const outer = error as { statusCode?: unknown; error?: unknown; cause?: unknown } | null; + if (typeof outer?.statusCode === "number") return outer.statusCode; + const inner = (outer?.error ?? outer?.cause) as { statusCode?: unknown } | undefined; + return typeof inner?.statusCode === "number" ? inner.statusCode : undefined; +} + +export const sendWebPush = ( + vapid: WebPushVapidKeys, + subscription: PushSubscriptionRecord, + payload: AttentionPushPayload, +): Effect.Effect => + Effect.tryPromise(() => + webpush.sendNotification( + { + endpoint: subscription.endpoint, + keys: { p256dh: subscription.p256dh, auth: subscription.auth }, + }, + // @effect-diagnostics-next-line preferSchemaOverJson:off - web-push requires a JSON string payload for the wire. + JSON.stringify(payload), + { + TTL: PUSH_TTL_SECONDS, + vapidDetails: { + subject: vapid.subject, + publicKey: vapid.publicKey, + privateKey: vapid.privateKey, + }, + }, + ), + ).pipe( + Effect.as({ ok: true, expired: false } satisfies WebPushSendResult), + Effect.catch((error: unknown) => { + const statusCode = extractStatusCode(error); + return Effect.succeed({ + ok: false, + expired: statusCode === 404 || statusCode === 410, + } satisfies WebPushSendResult); + }), + ); diff --git a/apps/server/src/t3x/webPush/state.ts b/apps/server/src/t3x/webPush/state.ts new file mode 100644 index 00000000000..fa076f8ea1e --- /dev/null +++ b/apps/server/src/t3x/webPush/state.ts @@ -0,0 +1,127 @@ +/** + * Durable Web Push subscription store. + * + * A single JSON file (in the server state dir) holding the browser push subscriptions + * registered against this environment, keyed by endpoint. Rehydrated on boot so + * subscriptions survive a restart. Mutations are serialized through a `SynchronizedRef` and + * persisted atomically inside the critical section. + * + * Deliberately NOT a DB migration: the migration registry is upstream-owned, and adding to it + * would buy permanent conflict surface for what one JSON file does fine. Mirrors + * t3x/autoResume/state.ts. + * + * @module t3x/webPush/state + */ + +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as SynchronizedRef from "effect/SynchronizedRef"; + +import { writeFileStringAtomically } from "../../atomicWrite.ts"; + +export const PushSubscriptionRecord = Schema.Struct({ + endpoint: Schema.String, + p256dh: Schema.String, + auth: Schema.String, + /** The auth session (device) that registered this subscription; for future cleanup. */ + sessionId: Schema.String, + createdAt: Schema.String, +}); +export type PushSubscriptionRecord = typeof PushSubscriptionRecord.Type; + +export const WebPushState = Schema.Struct({ + version: Schema.Literal(1), + subscriptions: Schema.Record(Schema.String, PushSubscriptionRecord), +}); +export type WebPushState = typeof WebPushState.Type; + +const EMPTY_STATE: WebPushState = { version: 1, subscriptions: {} }; + +const decodeState = Schema.decodeUnknownEffect(Schema.fromJsonString(WebPushState)); + +export interface PushSubscriptionStoreShape { + readonly list: Effect.Effect>; + readonly upsert: (record: PushSubscriptionRecord) => Effect.Effect; + readonly removeByEndpoint: (endpoint: string) => Effect.Effect; +} + +export class PushSubscriptionStore extends Context.Service< + PushSubscriptionStore, + PushSubscriptionStoreShape +>()("t3/t3x/webPush/state/PushSubscriptionStore") {} + +export const makePushSubscriptionStore = ( + stateFilePath: string, +): Effect.Effect => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const pathSvc = yield* Path.Path; + + // Rehydrate. Distinguish an unreadable file (I/O error) from a fresh/corrupt one so a + // transient read error does not overwrite still-valid subscriptions with an empty file. + const fileExists = yield* fs.exists(stateFilePath).pipe(Effect.orElseSucceed(() => false)); + let initial: WebPushState = EMPTY_STATE; + let persistEnabled = true; + if (fileExists) { + const contents = yield* fs.readFileString(stateFilePath).pipe( + Effect.map((c): string | null => c), + Effect.orElseSucceed(() => null), + ); + if (contents === null) { + persistEnabled = false; + yield* Effect.logWarning( + "t3x web-push: state file present but unreadable; running in-memory only this session so the existing file is not overwritten", + { stateFilePath }, + ); + } else { + initial = yield* decodeState(contents).pipe(Effect.orElseSucceed(() => EMPTY_STATE)); + } + } + + const ref = yield* SynchronizedRef.make(initial); + + const persist = (state: WebPushState): Effect.Effect => { + if (!persistEnabled) return Effect.void; + return writeFileStringAtomically({ + filePath: stateFilePath, + contents: `${JSON.stringify(state)}\n`, + }).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, pathSvc), + Effect.catch((cause) => + Effect.logWarning("t3x web-push: failed to persist state", { stateFilePath, cause }), + ), + ); + }; + + const mutate = (f: (state: WebPushState) => WebPushState) => + SynchronizedRef.updateEffect(ref, (state) => { + const next = f(state); + return persist(next).pipe(Effect.as(next)); + }); + + return { + list: SynchronizedRef.get(ref).pipe( + Effect.map((state) => Object.values(state.subscriptions)), + ), + + upsert: (record) => + mutate((state) => ({ + ...state, + subscriptions: { ...state.subscriptions, [record.endpoint]: record }, + })), + + removeByEndpoint: (endpoint) => + mutate((state) => { + // `Object.hasOwn`, not truthiness: `endpoint` is caller-controlled via the HTTP + // route, so a value like `__proto__` must not resolve on Object.prototype. + if (!Object.hasOwn(state.subscriptions, endpoint)) return state; + const next = { ...state.subscriptions }; + delete next[endpoint]; + return { ...state, subscriptions: next }; + }), + } satisfies PushSubscriptionStoreShape; + }); diff --git a/apps/server/src/t3x/webPush/vapid.ts b/apps/server/src/t3x/webPush/vapid.ts new file mode 100644 index 00000000000..179caa14f65 --- /dev/null +++ b/apps/server/src/t3x/webPush/vapid.ts @@ -0,0 +1,91 @@ +/** + * VAPID keypair provisioning for Web Push. + * + * Env-provided keys (`T3X_VAPID_PUBLIC_KEY` + `T3X_VAPID_PRIVATE_KEY`) win; otherwise a + * keypair is generated once and persisted in the ServerSecretStore (same place link/relay + * secrets live), then reused. Never fails: a persistence error falls back to an in-memory + * keypair for the session (logged) so push still works until the next restart. + * + * @module t3x/webPush/vapid + */ + +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import webpush from "web-push"; + +import { ServerSecretStore } from "../../auth/ServerSecretStore.ts"; +import type { WebPushConfig } from "./config.ts"; + +export interface WebPushVapidKeys { + readonly publicKey: string; + readonly privateKey: string; + readonly subject: string; +} + +export class WebPushVapid extends Context.Service()( + "t3/t3x/webPush/vapid/WebPushVapid", +) {} + +const SECRET_NAME = "t3x-web-push-vapid"; + +const VapidKeypair = Schema.Struct({ + publicKey: Schema.String, + privateKey: Schema.String, +}); +const decodeKeypair = Schema.decodeUnknownEffect(Schema.fromJsonString(VapidKeypair)); + +export const makeWebPushVapid = ( + config: WebPushConfig, +): Effect.Effect => + Effect.gen(function* () { + if (config.publicKeyOverride && config.privateKeyOverride) { + return { + publicKey: config.publicKeyOverride, + privateKey: config.privateKeyOverride, + subject: config.subject, + }; + } + + const secrets = yield* ServerSecretStore; + const existing = yield* secrets + .get(SECRET_NAME) + .pipe(Effect.orElseSucceed(() => Option.none())); + if (Option.isSome(existing)) { + const decoded = yield* decodeKeypair(new TextDecoder().decode(existing.value)).pipe( + Effect.map((k): { publicKey: string; privateKey: string } | null => k), + Effect.orElseSucceed(() => null), + ); + if (decoded) { + return { + publicKey: decoded.publicKey, + privateKey: decoded.privateKey, + subject: config.subject, + }; + } + } + + const generated = webpush.generateVAPIDKeys(); + yield* secrets + .set( + SECRET_NAME, + new TextEncoder().encode( + // @effect-diagnostics-next-line preferSchemaOverJson:off - persisting a small known keypair as a secret blob. + JSON.stringify({ publicKey: generated.publicKey, privateKey: generated.privateKey }), + ), + ) + .pipe( + Effect.catch((cause) => + Effect.logWarning( + "t3x web-push: failed to persist VAPID keys; using an in-memory keypair for this session (existing browser subscriptions will need to re-register after restart)", + { cause }, + ), + ), + ); + return { + publicKey: generated.publicKey, + privateKey: generated.privateKey, + subject: config.subject, + }; + }); diff --git a/apps/web/index.html b/apps/web/index.html index dadef17d3bc..2f7fe0be0b2 100644 --- a/apps/web/index.html +++ b/apps/web/index.html @@ -11,6 +11,11 @@ + + + + +