diff --git a/.env.example b/.env.example index 8ebb743..edc0245 100644 --- a/.env.example +++ b/.env.example @@ -166,8 +166,12 @@ SMTP_URL= EMAIL_FROM= # Local development only: capture password-reset email and print its link to the API console. EMAIL_EMULATOR= +# Optional web push keys (set both). Unset: generated once and stored encrypted in the database. +# Browsers only allow push on https (or localhost). VAPID_PUBLIC_KEY= VAPID_PRIVATE_KEY= +# Contact for push services: mailto: or https: URL. Defaults to WEB_ORIGIN when it is https. +VAPID_SUBJECT= # Structured logs to stdout (json in production, pretty in development unless overridden). LOG_LEVEL=info LOG_FORMAT= diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index c8b5057..b2b0faf 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -1,5 +1,4 @@ import { randomUUID } from "node:crypto"; -import { rm } from "node:fs/promises"; import { ORPCError, onError } from "@orpc/server"; import { RPCHandler } from "@orpc/server/fetch"; import type { @@ -35,13 +34,14 @@ import { destroyBot, EmailEmulator, EncryptedSecretStore, - ExpoPushProvider, + ExpoPushSender, GraphileJobPublisher, GraphileJobWorkerHost, InMemoryJobQueue, InMemoryRealtimeFanout, InstalledConnectorProvider, IntegrationProviderSettings, + importLegacyPushTokens, isComposioEnabled, isMessagingSurfaceEnabled, isPipedreamEnabled, @@ -54,9 +54,9 @@ import { PiOAuthLogins, PipedreamConnector, PostgresRealtimeFanout, + PushNotificationProvider, pipedreamConfigFromEnv, piSessionsRoot, - pushTokenPath, reconcileCloudAgents, reconcileComputerUpdates, reconcileDelegatedTasks, @@ -67,6 +67,7 @@ import { toTeamChatInbound, VapidKeys, vapidEnvFromProcess, + WebPushSender, } from "@rakazo/adapters"; import { blockedAuthPaths, createAuth } from "@rakazo/auth"; import { signupPolicyFromEnv } from "@rakazo/core"; @@ -311,8 +312,14 @@ export async function createApp( : new PiAgentRuntime({ sessionRoot: env.piSessionRecording ? piSessionsRoot(env.dataDir) : undefined, }); - const notifications = new ExpoPushProvider(env.dataDir); const vapid = new VapidKeys(prisma, secrets, vapidEnvFromProcess(process.env)); + const notifications = new PushNotificationProvider(prisma, { + webpush: new WebPushSender(vapid), + expo: new ExpoPushSender(), + }); + void importLegacyPushTokens(prisma, env.dataDir).catch((error) => + getLogger().error("legacy push token import failed", error), + ); const auth = createAuth(prisma, { secret: env.authSecret, baseURL: env.authUrl, @@ -368,7 +375,6 @@ export async function createApp( ), ); await removePiUserSessions(env.dataDir, userId); - await rm(pushTokenPath(env.dataDir, userId), { force: true }).catch(() => undefined); }, }); // One provider instance so emulator launches and polls share the same Map. diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index d95e1fc..b5606a4 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -45,7 +45,6 @@ import { computerSupportsUpdate, computerUpdateView, createVoiceProvider, - deletePushToken, deploymentAutoReviewDefault, destroyBot, displayBotWorkspacePath, @@ -77,7 +76,6 @@ import { resolveAutoReviewChecker, resolveBotWorkspacePath, sanitizeComposioError, - savePushToken, scheduleComputerControlExpiry, scheduleComputerSleep, screenLeaseIdForRun, @@ -140,6 +138,7 @@ import { projectWorkspaceDto, releaseSpaceDeletionClaim, removePushSubscription, + removePushSubscriptionsOfKind, renewSpaceDeletionClaim, requireMembership, SPACE_DELETION_CLAIM_TIMEOUT_MS, @@ -4798,11 +4797,16 @@ export function createRouter(deps: RouterDeps) { }, notifications: { registerPush: authed.notifications.registerPush.handler(async ({ context, input }) => { - await savePushToken(deps.dataDir, context.actor.userId, input.token); + await savePushSubscription(deps.prisma, context.actor.userId, { + kind: "expo", + endpoint: input.token, + userAgent: context.userAgent, + }); return { ok: true as const }; }), + // The mobile app signs out without naming its token, so all of its tokens go. unregisterPush: authed.notifications.unregisterPush.handler(async ({ context }) => { - await deletePushToken(deps.dataDir, context.actor.userId); + await removePushSubscriptionsOfKind(deps.prisma, context.actor.userId, "expo"); return { ok: true as const }; }), settings: { diff --git a/apps/web/e2e/helpers.ts b/apps/web/e2e/helpers.ts index 84d7794..32b77a3 100644 --- a/apps/web/e2e/helpers.ts +++ b/apps/web/e2e/helpers.ts @@ -242,6 +242,7 @@ export async function openUserSettings( | "memory" | "profile" | "voice" + | "notifications" | "usage" | "computer" | "updates", diff --git a/apps/web/e2e/notification-settings.spec.ts b/apps/web/e2e/notification-settings.spec.ts new file mode 100644 index 0000000..27e1571 --- /dev/null +++ b/apps/web/e2e/notification-settings.spec.ts @@ -0,0 +1,129 @@ +/** + * E2E: Settings → Notifications turns push on for this browser (the Push API is stubbed: headless + * Chromium has no push service), sets quiet hours and switches nudges off; all of it survives a + * reload. + */ +import { expect, type Page, test } from "@playwright/test"; +import { captureScreenshot, completeOnboarding, openUserSettings, rpc, signup } from "./helpers"; + +const ENDPOINT = "https://push.example.test/e2e-device"; + +/** + * A fake Notification permission and service worker registration, kept in sessionStorage so they + * survive a reload. Headless Chromium reports notifications as denied even when granted. + */ +async function stubPushApi(page: Page, endpoint: string) { + await page.addInitScript((pushEndpoint) => { + const permissionKey = "e2e:notification-permission"; + const notification = Object.defineProperties(function Notification() {}, { + permission: { get: () => sessionStorage.getItem(permissionKey) ?? "default" }, + requestPermission: { + value: async () => { + sessionStorage.setItem(permissionKey, "granted"); + return "granted"; + }, + }, + }); + Object.defineProperty(window, "Notification", { + configurable: true, + writable: true, + value: notification, + }); + const storageKey = "e2e:push-subscription"; + const current = () => { + const stored = sessionStorage.getItem(storageKey); + if (!stored) return null; + const saved = JSON.parse(stored) as { serverKey: number[] }; + return { + endpoint: pushEndpoint, + options: { applicationServerKey: new Uint8Array(saved.serverKey).buffer }, + toJSON: () => ({ + endpoint: pushEndpoint, + keys: { p256dh: `B${"A".repeat(86)}`, auth: "A".repeat(22) }, + }), + unsubscribe: async () => { + sessionStorage.removeItem(storageKey); + return true; + }, + }; + }; + const registration = { + scope: `${location.origin}/`, + pushManager: { + getSubscription: async () => current(), + subscribe: async (options: { applicationServerKey: Uint8Array }) => { + sessionStorage.setItem( + storageKey, + JSON.stringify({ serverKey: [...new Uint8Array(options.applicationServerKey)] }), + ); + return current(); + }, + }, + }; + Object.defineProperty(navigator, "serviceWorker", { + configurable: true, + value: { + register: async () => registration, + getRegistration: async () => registration, + ready: Promise.resolve(registration), + }, + }); + if (!("PushManager" in window)) { + Object.defineProperty(window, "PushManager", { configurable: true, value: class {} }); + } + }, endpoint); +} + +test("notification settings: push on this device, quiet hours and nudges", async ({ + page, +}, testInfo) => { + await stubPushApi(page, ENDPOINT); + await signup(page, `notifications-${Date.now()}@rakazo.test`, "password12", "Push Tester"); + await completeOnboarding(page); + + let settings = await openUserSettings(page, "notifications"); + const panel = settings.getByTestId("notification-settings"); + const push = panel.getByTestId("notification-push-switch"); + await expect(push).not.toBeChecked(); + + const subscribed = page.waitForRequest("**/rpc/notifications/webPush/subscribe"); + await push.click(); + expect((await subscribed).postDataJSON()).toMatchObject({ json: { endpoint: ENDPOINT } }); + await expect(push).toBeChecked(); + + const quiet = panel.getByTestId("notification-quiet-hours-switch"); + await expect(quiet).not.toBeChecked(); + await quiet.click(); + await expect(quiet).toBeChecked(); + const start = panel.getByTestId("notification-quiet-hours-start"); + await expect(start).toHaveValue("22:00"); + await expect(panel.getByTestId("notification-quiet-hours-end")).toHaveValue("07:00"); + const saved = page.waitForResponse( + (response) => + response.url().includes("/rpc/notifications/settings/update") && + response.request().postData()?.includes("23:30") === true, + ); + await start.fill("23:30"); + await saved; + + const nudges = panel.getByTestId("notification-nudges-switch"); + await expect(nudges).toBeChecked(); + await nudges.click(); + await expect(nudges).not.toBeChecked(); + await expect + .poll(() => rpc<{ nudges: boolean }>(page, "notifications/settings/get", null)) + .toMatchObject({ nudges: false, quietHours: { start: "23:30", end: "07:00" } }); + + await captureScreenshot(page, testInfo, "notification-settings"); + + await page.reload(); + settings = await openUserSettings(page, "notifications"); + await expect(settings.getByTestId("notification-push-switch")).toBeChecked(); + await expect(settings.getByTestId("notification-quiet-hours-start")).toHaveValue("23:30"); + await expect(settings.getByTestId("notification-nudges-switch")).not.toBeChecked(); + + const unsubscribed = page.waitForRequest("**/rpc/notifications/webPush/unsubscribe"); + await settings.getByTestId("notification-push-switch").click(); + expect((await unsubscribed).postDataJSON()).toMatchObject({ json: { endpoint: ENDPOINT } }); + await expect(settings.getByTestId("notification-push-switch")).not.toBeChecked(); +}); diff --git a/apps/web/src/components/NotificationSettingsPanel.tsx b/apps/web/src/components/NotificationSettingsPanel.tsx new file mode 100644 index 0000000..de71979 --- /dev/null +++ b/apps/web/src/components/NotificationSettingsPanel.tsx @@ -0,0 +1,197 @@ +import { Trans, useLingui } from "@lingui/react/macro"; +import type { NotificationSettings, NotificationSettingsPatch } from "@rakazo/contracts"; +import { Input, Switch } from "@rakazo/ui-web"; +import type { ReactNode } from "react"; +import { useEffect, useId, useRef, useState } from "react"; +import { rpc } from "../lib/rpc"; +import { disableWebPush, enableWebPush, webPushEnabled, webPushSupported } from "../lib/web-push"; + +const DEFAULT_QUIET_HOURS = { start: "22:00", end: "07:00" }; +const CLOCK_TIME = /^([01]\d|2[0-3]):[0-5]\d$/; + +type PushState = "loading" | "on" | "off" | "blocked"; + +function SettingRow({ + label, + testId, + checked, + disabled, + onCheckedChange, + children, +}: { + label: ReactNode; + testId: string; + checked: boolean; + disabled?: boolean; + onCheckedChange: (checked: boolean) => void; + children?: ReactNode; +}) { + const labelId = useId(); + return ( +
  • +
    + + {label} + + +
    + {children} +
    • + ); +} + +/** Push on this browser, quiet hours and nudges. Quiet hours and nudges apply to every device. */ +export function NotificationSettingsPanel() { + const { t } = useLingui(); + const pushSupported = webPushSupported(); + const [settings, setSettings] = useState(null); + const [push, setPush] = useState("loading"); + const [pushBusy, setPushBusy] = useState(false); + const [error, setError] = useState(null); + const [quietDraft, setQuietDraft] = useState(DEFAULT_QUIET_HOURS); + const saves = useRef(Promise.resolve()); + + useEffect(() => { + let cancelled = false; + rpc.notifications.settings + .get() + .then((next) => { + if (cancelled) return; + setSettings(next); + if (next.quietHours) setQuietDraft(next.quietHours); + }) + .catch(() => { + if (!cancelled) setError(t`Couldn't load`); + }); + return () => { + cancelled = true; + }; + }, [t]); + + useEffect(() => { + if (!pushSupported) return; + let cancelled = false; + void webPushEnabled().then((enabled) => { + if (cancelled) return; + setPush(enabled ? "on" : Notification.permission === "denied" ? "blocked" : "off"); + }); + return () => { + cancelled = true; + }; + }, [pushSupported]); + + /** Saves in order, so the last change the user made is the one that sticks. */ + function save(patch: NotificationSettingsPatch) { + setError(null); + setSettings((current) => (current ? { ...current, ...patch } : current)); + saves.current = saves.current.then(async () => { + try { + setSettings(await rpc.notifications.settings.update(patch)); + } catch { + setError(t`Couldn't save`); + setSettings(await rpc.notifications.settings.get().catch(() => null)); + } + }); + } + + async function changePush(next: boolean) { + setPushBusy(true); + setError(null); + try { + if (!next) { + await disableWebPush(); + setPush("off"); + return; + } + const result = await enableWebPush(); + if (result === "enabled") setPush("on"); + else if (result === "denied") setPush("blocked"); + else setError(t`Couldn't save`); + } catch { + setError(t`Couldn't save`); + } finally { + setPushBusy(false); + } + } + + function changeQuietTime(field: "start" | "end", value: string) { + const next = { ...quietDraft, [field]: value }; + setQuietDraft(next); + if (CLOCK_TIME.test(next.start) && CLOCK_TIME.test(next.end) && next.start !== next.end) { + save({ quietHours: next }); + } + } + + return ( +
    + {error ? ( +

    + {error} +

    + ) : null} + {settings ? ( +
    +
      + {pushSupported ? ( + Push on this device} + testId="notification-push-switch" + checked={push === "on"} + disabled={push === "loading" || push === "blocked" || pushBusy} + onCheckedChange={(next) => void changePush(next)} + > + {push === "blocked" ? ( +

      + Blocked in your browser settings +

      + ) : null} +       + ) : null} + Quiet hours} + testId="notification-quiet-hours-switch" + checked={settings.quietHours !== null} + onCheckedChange={(next) => save({ quietHours: next ? quietDraft : null })} + > + {settings.quietHours ? ( +
      + changeQuietTime("start", event.target.value)} + className="w-36" + /> + + changeQuietTime("end", event.target.value)} + className="w-36" + /> +
      + ) : null} +       + Nudges from your Oo} + testId="notification-nudges-switch" + checked={settings.nudges} + onCheckedChange={(next) => save({ nudges: next })} + /> +
    +
    + ) : null} +
    + ); +} diff --git a/apps/web/src/pages/SettingsOverlay.tsx b/apps/web/src/pages/SettingsOverlay.tsx index d71d40a..9f1b8f8 100644 --- a/apps/web/src/pages/SettingsOverlay.tsx +++ b/apps/web/src/pages/SettingsOverlay.tsx @@ -2,6 +2,7 @@ import { useLingui } from "@lingui/react/macro"; import type { AvatarStyle, SpaceMemoryConfig } from "@rakazo/contracts"; import { Button, Dialog, DialogClose, DialogContent, DialogTitle } from "@rakazo/ui-web"; import { + Bell, Brain, CloudDownload, Cpu, @@ -16,6 +17,7 @@ import { } from "lucide-react"; import { type ComponentType, useEffect, useRef, useState } from "react"; import { computersAreUnavailable } from "../components/ComputersUnavailableHint"; +import { NotificationSettingsPanel } from "../components/NotificationSettingsPanel"; import { UserProfileSettings } from "../components/UserProfileSettings"; import type { TradingBotUiState } from "../lib/trading-bot-status"; import { @@ -37,6 +39,7 @@ export type SettingsSection = | "memory" | "profile" | "voice" + | "notifications" | "usage" | "computer" | "updates" @@ -111,6 +114,7 @@ export function SettingsOverlay({ { id: "memory", label: t`Memory`, icon: Brain }, { id: "profile", label: t`About you`, icon: UserRound }, { id: "voice", label: t`Voice`, icon: Volume2 }, + { id: "notifications", label: t`Notifications`, icon: Bell }, { id: "usage", label: t`Usage`, icon: Gauge }, ...(showComputer ? [{ id: "computer" as const, label: t`Computer`, icon: Monitor }] : []), { id: "updates", label: t`Updates`, icon: CloudDownload }, @@ -250,6 +254,7 @@ export function SettingsOverlay({ /> ) : null} {section === "profile" ? : null} + {section === "notifications" ? : null} {section === "usage" ? ( ) : null} diff --git a/apps/web/src/pages/Shell.tsx b/apps/web/src/pages/Shell.tsx index a3b1e3a..4656244 100644 --- a/apps/web/src/pages/Shell.tsx +++ b/apps/web/src/pages/Shell.tsx @@ -247,6 +247,7 @@ import { transcriptMovedDown, } from "../lib/transcript-scroll"; import { speaker } from "../lib/tts"; +import { disableWebPush } from "../lib/web-push"; import { ActivityList } from "./ActivityList"; import type { ContextMenuPosition } from "./BotContextMenu"; import { CreateGroupForm, GroupSettings, memberName } from "./GroupPanel"; @@ -4296,10 +4297,14 @@ export function ShellPage() { variant="ghost" className="w-full justify-start font-normal" onClick={() => - void authClient.signOut().then(() => { - clearSpaceSelection(); - navigate("/"); - }) + // This browser stops receiving the account's pushes once it signs out. + void disableWebPush() + .catch(() => undefined) + .then(() => authClient.signOut()) + .then(() => { + clearSpaceSelection(); + navigate("/"); + }) } > diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 804281c..39fba39 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -19,7 +19,7 @@ import { createWebProvider, databaseCapacityBackoffMs, EncryptedSecretStore, - ExpoPushProvider, + ExpoPushSender, GraphileJobPublisher, GraphileJobWorkerHost, InMemoryJobQueue, @@ -36,6 +36,7 @@ import { PiAgentRuntime, PipedreamConnector, PostgresRealtimeFanout, + PushNotificationProvider, pipedreamConfigFromEnv, reconcileCloudAgents, reconcileComputerUpdates, @@ -45,6 +46,9 @@ import { resolveSandboxProvider, ScriptedAgentRuntime, SpaceMemoryProviderResolver, + VapidKeys, + vapidEnvFromProcess, + WebPushSender, } from "@rakazo/adapters"; import { resolveEncryptionKey, resolveSupervisorToken } from "@rakazo/core"; import { @@ -193,7 +197,10 @@ async function main() { secretStore: secrets, deploymentModelKey, dataDir, - notifications: new ExpoPushProvider(dataDir), + notifications: new PushNotificationProvider(prisma, { + webpush: new WebPushSender(new VapidKeys(prisma, secrets, vapidEnvFromProcess(process.env))), + expo: new ExpoPushSender(), + }), jobs, events, messaging: messaging ? createMessagingContextLoader(prisma) : undefined, diff --git a/packages/adapter-kit/src/types.ts b/packages/adapter-kit/src/types.ts index 8ca259f..f47f460 100644 --- a/packages/adapter-kit/src/types.ts +++ b/packages/adapter-kit/src/types.ts @@ -555,7 +555,7 @@ export interface ArtifactPut { } export interface NotificationMessage { - kind: "completion" | "failure" | "help" | "takeover"; + kind: "completion" | "failure" | "help" | "takeover" | "nudge"; title: string; body: string; botId: string; diff --git a/packages/adapters/src/builtin-tools.ts b/packages/adapters/src/builtin-tools.ts index 8502640..560893f 100644 --- a/packages/adapters/src/builtin-tools.ts +++ b/packages/adapters/src/builtin-tools.ts @@ -8,6 +8,7 @@ import { USER_FACT_CATEGORIES, } from "@rakazo/contracts"; import { z } from "zod"; +import { NOTIFY_USER_BODY_MAX_LENGTH, NOTIFY_USER_TITLE_MAX_LENGTH } from "./notify-user-tool.js"; export const SUBAGENT_MODEL_TIER_DESCRIPTION = "Optional model tier: fast for quick simple work, code for programming, standard otherwise. Uses the model this space maps to the tier, or your own model when it maps none. Ignored when model_provider and model_id are set."; @@ -428,6 +429,27 @@ export const builtinAgentTools: ConnectorTool[] = [ }, }, }, + { + name: "notify_user", + description: + "Send the user a short push notification that opens the main conversation. Use it sparingly, for a check-in or reminder that matters now, and say the same in the conversation. The user's quiet hours, their nudge setting and a daily limit apply; the result says whether it was sent.", + inputSchema: { + type: "object", + properties: { + body: { + type: "string", + maxLength: NOTIFY_USER_BODY_MAX_LENGTH, + description: "One or two short sentences.", + }, + title: { + type: "string", + maxLength: NOTIFY_USER_TITLE_MAX_LENGTH, + description: "Optional; defaults to your name.", + }, + }, + required: ["body"], + }, + }, { name: "web_search", description: diff --git a/packages/adapters/src/executor-tools.test.ts b/packages/adapters/src/executor-tools.test.ts index d987728..bc2d541 100644 --- a/packages/adapters/src/executor-tools.test.ts +++ b/packages/adapters/src/executor-tools.test.ts @@ -64,7 +64,11 @@ describe("builtin tools per bot", () => { }); it("lists main-only tools for the main bot alone", () => { - expect([...MAIN_ONLY_BUILTIN_TOOLS].sort()).toEqual(["delegate_task", "update_user_profile"]); + expect([...MAIN_ONLY_BUILTIN_TOOLS].sort()).toEqual([ + "delegate_task", + "notify_user", + "update_user_profile", + ]); const specialist = builtinToolNamesForBot({ kind: "specialist", chatLinked: false }); const main = builtinToolNamesForBot({ kind: "main", chatLinked: false }); for (const name of MAIN_ONLY_BUILTIN_TOOLS) { diff --git a/packages/adapters/src/executor-tools.ts b/packages/adapters/src/executor-tools.ts index 2e819b8..671ad46 100644 --- a/packages/adapters/src/executor-tools.ts +++ b/packages/adapters/src/executor-tools.ts @@ -19,6 +19,7 @@ export const PAGE_BROWSER_TOOL_NAMES = new Set([ /** Builtin tools only the user's main bot may use; every other bot's runs never see them. */ export const MAIN_ONLY_BUILTIN_TOOLS: ReadonlySet = new Set([ "delegate_task", + "notify_user", "update_user_profile", ]); @@ -105,7 +106,10 @@ export function selectBuiltinToolsForRun(options: { Boolean(options.cloudAgentEnabled), ).filter( (tool) => - !(options.foreignAgentRun && tool.name === "update_user_profile") && + !( + options.foreignAgentRun && + (tool.name === "update_user_profile" || tool.name === "notify_user") + ) && (!options.messagingChannelRun || (![ "remember", @@ -113,6 +117,7 @@ export function selectBuiltinToolsForRun(options: { "recall_memory", "forget_memory", "update_user_profile", + "notify_user", ].includes(tool.name) && !tool.name.startsWith("scratchpad_"))), ); diff --git a/packages/adapters/src/executor.ts b/packages/adapters/src/executor.ts index 7181880..994507a 100644 --- a/packages/adapters/src/executor.ts +++ b/packages/adapters/src/executor.ts @@ -298,6 +298,7 @@ import { modelAcceptsImageInput, modelIdSupportsImages, } from "./model-vision.js"; +import { notifyUserFromTool } from "./notify-user-tool.js"; import { toOAuthCredential } from "./pi-credentials.js"; import { parseModelSecret, @@ -2571,6 +2572,15 @@ export function createRunExecutor(deps: ExecutorDeps) { ), ); } + if (name === "notify_user") { + return finish( + await notifyUserFromTool( + deps, + { userId: run.userId, spaceId: run.spaceId, runId, bot, signal: context.signal }, + args, + ), + ); + } if (name === "web_search") { return finish(await webSearchFromTool(web, context, args)); } diff --git a/packages/adapters/src/expo-push.test.ts b/packages/adapters/src/expo-push.test.ts index c8a7f4a..9052444 100644 --- a/packages/adapters/src/expo-push.test.ts +++ b/packages/adapters/src/expo-push.test.ts @@ -1,15 +1,16 @@ -import { chmod, mkdtemp, readFile, rm, stat, symlink, writeFile } from "node:fs/promises"; +import { mkdir, mkdtemp, readdir, rm, symlink, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; +import type { PrismaClient } from "@rakazo/db"; import { afterEach, describe, expect, it, vi } from "vitest"; import { - deletePushToken, - ExpoPushProvider, + ExpoPushSender, expoPushErrorMessage, - loadPushToken, + importLegacyPushTokens, + legacyPushTokenDir, MAX_EXPO_PUSH_RESPONSE_BYTES, - savePushToken, } from "./expo-push.js"; +import type { PushPayload } from "./push-notifications.js"; const dirs: string[] = []; @@ -18,13 +19,19 @@ afterEach(async () => { await Promise.all(dirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); }); -const notifyContext = { - operationId: "n", - traceId: "n", - spaceId: "w", - userId: "user-1", - signal: new AbortController().signal, -}; +const device = { endpoint: "ExponentPushToken[test]", p256dh: null, auth: null }; + +function payload(overrides: Partial = {}): PushPayload { + return { + kind: "completion", + title: "done", + body: "ok", + botId: "b", + threadId: "t", + url: "/app/b", + ...overrides, + }; +} function jsonResponse(body: unknown, status = 200) { return Response.json(body, { status }); @@ -49,75 +56,24 @@ describe("expo push tickets", () => { }); }); -describe("expo push", () => { - it("keeps refreshed push tokens owner-only", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - const tokenFile = path.join(dataDir, "push-tokens", "user-1.txt"); - await savePushToken(dataDir, "user-1", "ExponentPushToken[old]"); - await chmod(tokenFile, 0o644); - - await savePushToken(dataDir, "user-1", "ExponentPushToken[new]"); - - expect((await stat(tokenFile)).mode & 0o777).toBe(0o600); - }); - - it("does not follow a token-file symlink for reads or writes", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - const tokenDir = path.join(dataDir, "push-tokens"); - const tokenFile = path.join(tokenDir, "user-1.txt"); - const target = path.join(dataDir, "outside.txt"); - await savePushToken(dataDir, "user-1", "ExponentPushToken[old]"); - await writeFile(target, "not-a-push-token"); - await rm(tokenFile); - await symlink(target, tokenFile); - - await expect(loadPushToken(dataDir, "user-1")).resolves.toBeUndefined(); - await expect(savePushToken(dataDir, "user-1", "ExponentPushToken[new]")).rejects.toThrow(); - await expect(readFile(target, "utf8")).resolves.toBe("not-a-push-token"); - }); - - it("removes a registered token", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); - await deletePushToken(dataDir, "user-1"); - await expect(loadPushToken(dataDir, "user-1")).resolves.toBeUndefined(); - }); - - it("does not call Expo when the user has no token", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - const fetchMock = vi.fn(); - vi.stubGlobal("fetch", fetchMock); - const push = new ExpoPushProvider(dataDir); - await push.send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - { - operationId: "n", - traceId: "n", - spaceId: "w", - userId: "missing", - signal: new AbortController().signal, - }, - ); - expect(fetchMock).not.toHaveBeenCalled(); - }); - - it("posts to Expo when a token is registered", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); +describe("expo push sender", () => { + it("posts the notification to Expo for the device token", async () => { const fetchMock = vi .fn() .mockResolvedValue(jsonResponse({ data: { status: "ok", id: "ticket" } })); vi.stubGlobal("fetch", fetchMock); - const push = new ExpoPushProvider(dataDir); - await push.send( - { kind: "takeover", title: "Need you", body: "on screen", botId: "bot-1", threadId: "th-1" }, - notifyContext, - ); + await expect( + new ExpoPushSender().send( + device, + payload({ + kind: "takeover", + title: "Need you", + body: "on screen", + botId: "bot-1", + threadId: "th-1", + }), + ), + ).resolves.toBe("sent"); expect(fetchMock).toHaveBeenCalledOnce(); const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit]; expect(url).toBe("https://exp.host/--/api/v2/push/send"); @@ -126,55 +82,51 @@ describe("expo push", () => { title: string; collapseId: string; tag: string; - data: { kind: string }; + data: { kind: string; botId: string; threadId: string }; }; - expect(body.to).toBe("ExponentPushToken[test]"); - expect(body.title).toBe("Need you"); - expect(body.collapseId).toBe("th-1"); - expect(body.tag).toBe("th-1"); - expect(body.data.kind).toBe("takeover"); + expect(body).toMatchObject({ + to: "ExponentPushToken[test]", + title: "Need you", + collapseId: "th-1", + tag: "th-1", + data: { kind: "takeover", botId: "bot-1", threadId: "th-1" }, + }); + }); + + it("reports a device Expo no longer knows as gone", async () => { + vi.stubGlobal( + "fetch", + vi + .fn() + .mockResolvedValue( + jsonResponse({ data: { status: "error", details: { error: "DeviceNotRegistered" } } }), + ), + ); + await expect(new ExpoPushSender().send(device, payload())).resolves.toBe("gone"); }); it("throws when Expo rejects the request", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); vi.stubGlobal( "fetch", vi.fn().mockResolvedValue(jsonResponse({ errors: [{ message: "boom" }] }, 500)), ); - const push = new ExpoPushProvider(dataDir); - await expect( - push.send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - notifyContext, - ), - ).rejects.toThrow("boom"); + await expect(new ExpoPushSender().send(device, payload())).rejects.toThrow("boom"); }); it("rejects and cancels a declared oversized response", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); const response = new Response("oversized", { headers: { "content-length": String(MAX_EXPO_PUSH_RESPONSE_BYTES + 1) }, }); const cancel = vi.spyOn(response.body!, "cancel"); vi.stubGlobal("fetch", vi.fn().mockResolvedValue(response)); - await expect( - new ExpoPushProvider(dataDir).send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - notifyContext, - ), - ).rejects.toThrow("Expo push response is too large."); + await expect(new ExpoPushSender().send(device, payload())).rejects.toThrow( + "Expo push response is too large.", + ); expect(cancel).toHaveBeenCalledOnce(); }); it("does not wait past cancellation when an oversized body cancel hangs", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); let cancelStarted = false; const hangingBody = new ReadableStream({ cancel() { @@ -194,51 +146,31 @@ describe("expo push", () => { ); const started = Date.now(); - await expect( - new ExpoPushProvider(dataDir).send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - { ...notifyContext, signal: abort.signal }, - ), - ).rejects.toThrow("Expo push response is too large."); + await expect(new ExpoPushSender().send(device, payload(), abort.signal)).rejects.toThrow( + "Expo push response is too large.", + ); expect(cancelStarted).toBe(true); expect(Date.now() - started).toBeLessThan(500); }); it("caps a streamed response without a content length", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); vi.stubGlobal( "fetch", vi.fn().mockResolvedValue(new Response(new Uint8Array(MAX_EXPO_PUSH_RESPONSE_BYTES + 1))), ); - - await expect( - new ExpoPushProvider(dataDir).send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - notifyContext, - ), - ).rejects.toThrow("Expo push response is too large."); + await expect(new ExpoPushSender().send(device, payload())).rejects.toThrow( + "Expo push response is too large.", + ); }); it("rejects malformed successful responses", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); vi.stubGlobal("fetch", vi.fn().mockResolvedValue(new Response("not json"))); - - await expect( - new ExpoPushProvider(dataDir).send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - notifyContext, - ), - ).rejects.toThrow("Expo push returned an invalid response."); + await expect(new ExpoPushSender().send(device, payload())).rejects.toThrow( + "Expo push returned an invalid response.", + ); }); it("passes caller cancellation to the Expo request", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); const controller = new AbortController(); controller.abort(new Error("notification cancelled")); const fetchMock = vi.fn(async (_url: string, init?: RequestInit) => { @@ -247,49 +179,69 @@ describe("expo push", () => { }); vi.stubGlobal("fetch", fetchMock); - await expect( - new ExpoPushProvider(dataDir).send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - { ...notifyContext, signal: controller.signal }, - ), - ).rejects.toThrow("notification cancelled"); + await expect(new ExpoPushSender().send(device, payload(), controller.signal)).rejects.toThrow( + "notification cancelled", + ); expect(fetchMock).toHaveBeenCalledOnce(); }); it("throws when the Expo request never reaches the network", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[test]"); vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("offline"))); - const push = new ExpoPushProvider(dataDir); - await expect( - push.send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - notifyContext, - ), - ).rejects.toThrow("offline"); - }); - - it("reports DeviceNotRegistered without deleting a stored or replacement token", async () => { - const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); - dirs.push(dataDir); - await savePushToken(dataDir, "user-1", "ExponentPushToken[old]"); - vi.stubGlobal( - "fetch", - vi.fn(async () => { - await savePushToken(dataDir, "user-1", "ExponentPushToken[new]"); - return jsonResponse({ - data: { status: "error", details: { error: "DeviceNotRegistered" } }, - }); - }), - ); - const push = new ExpoPushProvider(dataDir); - await expect( - push.send( - { kind: "completion", title: "done", body: "ok", botId: "b", threadId: "t" }, - notifyContext, - ), - ).rejects.toThrow("DeviceNotRegistered"); - await expect(loadPushToken(dataDir, "user-1")).resolves.toBe("ExponentPushToken[new]"); + await expect(new ExpoPushSender().send(device, payload())).rejects.toThrow("offline"); + }); +}); + +describe("legacy push token import", () => { + function fakePrisma(existingUsers: string[]) { + const upsert = vi.fn().mockResolvedValue({}); + const prisma = { + user: { + findUnique: vi.fn(async ({ where }: { where: { id: string } }) => + existingUsers.includes(where.id) ? { id: where.id } : null, + ), + }, + pushSubscription: { + upsert, + findMany: vi.fn().mockResolvedValue([]), + deleteMany: vi.fn().mockResolvedValue({ count: 0 }), + }, + } as unknown as PrismaClient; + return { prisma, upsert }; + } + + it("imports each token once, skips unknown users and removes the files", async () => { + const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); + dirs.push(dataDir); + const dir = legacyPushTokenDir(dataDir); + await mkdir(dir, { recursive: true }); + await writeFile(path.join(dir, "user-1.txt"), "ExponentPushToken[one]\n"); + await writeFile(path.join(dir, "gone-user.txt"), "ExponentPushToken[two]"); + await writeFile(path.join(dir, "user-2.txt"), "not a token"); + const { prisma, upsert } = fakePrisma(["user-1", "user-2"]); + + await expect(importLegacyPushTokens(prisma, dataDir)).resolves.toBe(1); + + expect(upsert).toHaveBeenCalledOnce(); + expect(upsert.mock.calls[0]![0]).toMatchObject({ + where: { endpoint: "ExponentPushToken[one]" }, + create: { endpoint: "ExponentPushToken[one]", userId: "user-1", kind: "expo" }, + }); + await expect(readdir(dataDir)).resolves.toEqual([]); + await expect(importLegacyPushTokens(prisma, dataDir)).resolves.toBe(0); + }); + + it("does not follow a token-file symlink", async () => { + const dataDir = await mkdtemp(path.join(tmpdir(), "rakazo-push-")); + dirs.push(dataDir); + const dir = legacyPushTokenDir(dataDir); + await mkdir(dir, { recursive: true }); + const outside = path.join(dataDir, "outside.txt"); + await writeFile(outside, "ExponentPushToken[outside]"); + await symlink(outside, path.join(dir, "user-1.txt")); + const { prisma, upsert } = fakePrisma(["user-1"]); + + await expect(importLegacyPushTokens(prisma, dataDir)).resolves.toBe(0); + expect(upsert).not.toHaveBeenCalled(); + await expect(readdir(dataDir)).resolves.toEqual(["outside.txt"]); }); }); diff --git a/packages/adapters/src/expo-push.ts b/packages/adapters/src/expo-push.ts index 3ff5dd9..20192f4 100644 --- a/packages/adapters/src/expo-push.ts +++ b/packages/adapters/src/expo-push.ts @@ -1,29 +1,29 @@ import { constants } from "node:fs"; -import { mkdir, open, unlink } from "node:fs/promises"; +import { open, readdir, rmdir, unlink } from "node:fs/promises"; import path from "node:path"; -import type { - AdapterContext, - NotificationMessage, - NotificationProvider, -} from "@rakazo/adapter-kit"; -import { getLogger } from "@rakazo/logging"; +import { ExpoPushTokenSchema } from "@rakazo/contracts"; +import type { PrismaClient } from "@rakazo/db"; +import { savePushSubscription } from "@rakazo/db"; import { combineSignals } from "./connector-safety.js"; +import type { PushOutcome, PushPayload, PushTarget } from "./push-notifications.js"; import { readBodyCapped, withAbort } from "./web-ssrf.js"; const O_NOFOLLOW = constants.O_NOFOLLOW ?? 0; const EXPO_PUSH_TIMEOUT_MS = 15_000; export const MAX_EXPO_PUSH_RESPONSE_BYTES = 64 * 1024; +const LEGACY_TOKEN_FILE = /^([A-Za-z0-9_-]{1,128})\.txt$/; -export function pushTokenPath(dataDir: string, userId: string) { - return path.join(dataDir, "push-tokens", `${userId}.txt`); +/** Where the API kept the single Expo token per user before push_subscriptions existed. */ +export function legacyPushTokenDir(dataDir: string) { + return path.join(dataDir, "push-tokens"); } -export async function loadPushToken(dataDir: string, userId: string): Promise { +async function readLegacyToken(file: string): Promise { try { - const handle = await open(pushTokenPath(dataDir, userId), constants.O_RDONLY | O_NOFOLLOW); + const handle = await open(file, constants.O_RDONLY | O_NOFOLLOW); try { - const token = (await handle.readFile("utf8")).trim(); - return token || undefined; + const token = ExpoPushTokenSchema.safeParse(await handle.readFile("utf8")); + return token.success ? token.data : undefined; } finally { await handle.close(); } @@ -32,26 +32,39 @@ export async function loadPushToken(dataDir: string, userId: string): Promise { - const file = pushTokenPath(dataDir, userId); - await mkdir(path.dirname(file), { recursive: true }); - const handle = await open( - file, - constants.O_WRONLY | constants.O_CREAT | constants.O_TRUNC | O_NOFOLLOW, - 0o600, - ); +/** + * Moves tokens from the legacy files into push_subscriptions once, then deletes each file. A file + * whose user no longer exists, or that holds no usable token, is deleted without import; a + * database failure keeps the file for the next start. + */ +export async function importLegacyPushTokens( + prisma: PrismaClient, + dataDir: string, +): Promise { + const dir = legacyPushTokenDir(dataDir); + let names: string[]; try { - await handle.chmod(0o600); - await handle.writeFile(token.trim(), "utf8"); - } finally { - await handle.close(); + names = await readdir(dir); + } catch { + return 0; } -} - -export async function deletePushToken(dataDir: string, userId: string): Promise { - await unlink(pushTokenPath(dataDir, userId)).catch((error: NodeJS.ErrnoException) => { - if (error.code !== "ENOENT") throw error; - }); + let imported = 0; + for (const name of names) { + const userId = LEGACY_TOKEN_FILE.exec(name)?.[1]; + if (!userId) continue; + const file = path.join(dir, name); + const token = await readLegacyToken(file); + if (token) { + const user = await prisma.user.findUnique({ where: { id: userId }, select: { id: true } }); + if (user) { + await savePushSubscription(prisma, userId, { kind: "expo", endpoint: token }); + imported += 1; + } + } + await unlink(file).catch(() => undefined); + } + await rmdir(dir).catch(() => undefined); + return imported; } export type ExpoPushTicket = { @@ -87,49 +100,33 @@ export function expoPushErrorMessage(body: unknown, status: number): string | un return undefined; } -export class ExpoPushProvider implements NotificationProvider { - constructor(private readonly dataDir: string) {} - - describe() { - return { - id: "expo-push", - contractVersion: "1", - adapterVersion: "0.1.0", - capabilities: { push: true, email: false }, - }; - } - - async send(message: NotificationMessage, context: AdapterContext): Promise { - const token = await loadPushToken(this.dataDir, context.userId); - if (!token) return; - const signal = combineSignals(context.signal, AbortSignal.timeout(EXPO_PUSH_TIMEOUT_MS)); - let response: Response; - try { - response = await fetch("https://exp.host/--/api/v2/push/send", { - method: "POST", - headers: { "content-type": "application/json", accept: "application/json" }, - body: JSON.stringify({ - to: token, - title: message.title, - body: message.body, - collapseId: message.threadId, - tag: message.threadId, - data: { kind: message.kind, botId: message.botId, threadId: message.threadId }, - }), - signal, - }); - } catch (error) { - getLogger().error("expo push request failed", error); - throw error; - } - const body = await readExpoPushBody(response, signal); +/** Sends to one Expo push token; "gone" when Expo reports the device is no longer registered. */ +export class ExpoPushSender { + async send(target: PushTarget, payload: PushPayload, signal?: AbortSignal): Promise { + const deadline = combineSignals(signal, AbortSignal.timeout(EXPO_PUSH_TIMEOUT_MS)); + const response = await fetch("https://exp.host/--/api/v2/push/send", { + method: "POST", + headers: { "content-type": "application/json", accept: "application/json" }, + body: JSON.stringify({ + to: target.endpoint, + title: payload.title, + body: payload.body, + collapseId: payload.threadId, + tag: payload.threadId, + data: { kind: payload.kind, botId: payload.botId, threadId: payload.threadId }, + }), + signal: deadline, + }); + const body = await readExpoPushBody(response, deadline); if (response.ok && body === undefined) { throw new Error("Expo push returned an invalid response."); } + if (expoPushTickets(body).some((ticket) => ticket.details?.error === "DeviceNotRegistered")) { + return "gone"; + } const failure = expoPushErrorMessage(body, response.status); - if (!failure) return; - getLogger().error(failure); - throw new Error(failure); + if (failure) throw new Error(failure); + return "sent"; } } diff --git a/packages/adapters/src/index.ts b/packages/adapters/src/index.ts index 2a841cc..7312219 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -101,6 +101,7 @@ export * from "./pipedream-connector.js"; export * from "./profile-consolidation.js"; export * from "./project-workspace-files.js"; export * from "./project-workspace-path.js"; +export * from "./push-notifications.js"; export * from "./realtime.js"; export * from "./release-watch.js"; export * from "./remote-mcp.js"; diff --git a/packages/adapters/src/notify-user-tool.ts b/packages/adapters/src/notify-user-tool.ts new file mode 100644 index 0000000..6b10256 --- /dev/null +++ b/packages/adapters/src/notify-user-tool.ts @@ -0,0 +1,85 @@ +import type { NotificationProvider } from "@rakazo/adapter-kit"; +import { NUDGES_PER_DAY } from "@rakazo/contracts"; +import type { PrismaClient } from "@rakazo/db"; +import { + claimDailyNudge, + countPushSubscriptions, + ensureMainConversation, + loadNotificationPolicy, +} from "@rakazo/db"; +import { z } from "zod"; + +export const NOTIFY_USER_BODY_MAX_LENGTH = 200; +export const NOTIFY_USER_TITLE_MAX_LENGTH = 60; + +const NotifyUserArgs = z.object({ + body: z.string().trim().min(1).max(NOTIFY_USER_BODY_MAX_LENGTH), + title: z.string().trim().max(NOTIFY_USER_TITLE_MAX_LENGTH).optional(), +}); + +export interface NotifyUserToolContext { + userId: string; + spaceId: string; + runId: string; + bot: { id: string; name: string; kind: string }; + signal: AbortSignal; +} + +export type NotifyUserResult = { error: string } | { sent: true } | { sent: false; reason: string }; + +/** + * A nudge from the main bot: a push that opens the main conversation. The user's nudge switch, + * quiet hours and the daily cap all apply; a held-back nudge is reported, not an error. + */ +export async function notifyUserFromTool( + deps: { prisma: PrismaClient; notifications?: NotificationProvider }, + context: NotifyUserToolContext, + rawArgs: Record, +): Promise { + if (context.bot.kind !== "main") { + return { error: "Only the user's main bot can notify the user." }; + } + const parsed = NotifyUserArgs.safeParse(rawArgs); + if (!parsed.success) { + return { + error: `Invalid notification: body is required, up to ${NOTIFY_USER_BODY_MAX_LENGTH} characters; title up to ${NOTIFY_USER_TITLE_MAX_LENGTH}.`, + }; + } + const { prisma, notifications } = deps; + if (!notifications) { + return { sent: false, reason: "Push notifications are not available on this server." }; + } + const policy = await loadNotificationPolicy(prisma, context.userId); + if (!policy.nudges) return { sent: false, reason: "The user turned nudges off." }; + if (policy.quiet) return { sent: false, reason: "It is the user's quiet hours." }; + if ((await countPushSubscriptions(prisma, context.userId)) === 0) { + return { sent: false, reason: "The user has no device with notifications turned on." }; + } + if (!(await claimDailyNudge(prisma, context.userId, policy.localDate))) { + return { sent: false, reason: `The daily limit of ${NUDGES_PER_DAY} nudges is reached.` }; + } + const conversation = await ensureMainConversation(prisma, context.userId); + try { + await notifications.send( + { + kind: "nudge", + title: parsed.data.title || context.bot.name, + body: parsed.data.body, + botId: context.bot.id, + threadId: conversation.threadId, + }, + { + operationId: `notify_user:${context.runId}`, + traceId: context.runId, + spaceId: context.spaceId, + userId: context.userId, + botId: context.bot.id, + runId: context.runId, + signal: context.signal, + }, + ); + } catch { + return { sent: false, reason: "No device accepted the notification." }; + } + return { sent: true }; +} diff --git a/packages/adapters/src/notify-user.postgres.test.ts b/packages/adapters/src/notify-user.postgres.test.ts new file mode 100644 index 0000000..e5e63d2 --- /dev/null +++ b/packages/adapters/src/notify-user.postgres.test.ts @@ -0,0 +1,219 @@ +import { randomUUID } from "node:crypto"; +import type { PrismaClient } from "@rakazo/db"; +import { + bootstrapUserSpace, + createDb, + ensureMainConversation, + savePushSubscription, + updateNotificationSettings, +} from "@rakazo/db"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { notifyUserFromTool } from "./notify-user-tool.js"; +import { + PushNotificationProvider, + type PushPayload, + type PushSender, +} from "./push-notifications.js"; + +const databaseUrl = process.env.DATABASE_URL; +const describePostgres = + process.env.VERIFY_DATABASE && databaseUrl ? describe.sequential : describe.skip; + +const signupEnv = { signupsEnabled: "true", signupAllowlist: "" }; +const keys = { p256dh: `B${"A".repeat(86)}`, auth: "A".repeat(22) }; + +describePostgres("push delivery and notify_user (PostgreSQL)", () => { + let prisma: PrismaClient; + let close: () => Promise; + const userIds: string[] = []; + + async function newUser() { + const id = `nu${randomUUID().replaceAll("-", "")}`; + userIds.push(id); + await prisma.user.create({ data: { id, name: "Ada", email: `${id}@rakazo.test` } }); + await bootstrapUserSpace(prisma, { id }, signupEnv, { claimDeploymentOwner: false }); + const conversation = await ensureMainConversation(prisma, id); + const main = await prisma.bot.findUniqueOrThrow({ + where: { id: conversation.leadBotId }, + select: { id: true, name: true, kind: true }, + }); + return { userId: id, conversation, main }; + } + + function recordingSender(outcome: "sent" | "gone" = "sent") { + const sent: PushPayload[] = []; + const sender: PushSender = { + send: vi.fn(async (_target, payload: PushPayload) => { + sent.push(payload); + return outcome; + }), + }; + return { sender, sent }; + } + + function toolContext(user: Awaited>, bot = user.main) { + return { + userId: user.userId, + spaceId: user.conversation.spaceId, + runId: `run-${randomUUID()}`, + bot, + signal: new AbortController().signal, + }; + } + + beforeAll(() => { + const db = createDb(databaseUrl!); + prisma = db.prisma; + close = async () => { + await db.prisma.$disconnect(); + await db.pool.end(); + }; + }); + + afterAll(async () => { + if (!prisma) return; + const memberships = await prisma.member.findMany({ + where: { userId: { in: userIds } }, + select: { organizationId: true }, + }); + await prisma.organization.deleteMany({ + where: { id: { in: memberships.map((member) => member.organizationId) } }, + }); + await prisma.user.deleteMany({ where: { id: { in: userIds } } }); + await close(); + }); + + it("delivers to each device, links the conversation and drops gone devices", async () => { + const user = await newUser(); + const browser = `https://push.example.test/${randomUUID()}`; + await savePushSubscription(prisma, user.userId, { + kind: "webpush", + endpoint: browser, + ...keys, + }); + await savePushSubscription(prisma, user.userId, { + kind: "expo", + endpoint: `ExponentPushToken[${user.userId}]`, + }); + const web = recordingSender("sent"); + const expo = recordingSender("gone"); + await new PushNotificationProvider(prisma, { webpush: web.sender, expo: expo.sender }).send( + { + kind: "completion", + title: "Chief finished", + body: "Done", + botId: user.main.id, + threadId: user.conversation.threadId, + }, + { + operationId: "notify", + traceId: "notify", + spaceId: user.conversation.spaceId, + userId: user.userId, + signal: new AbortController().signal, + }, + ); + expect(web.sent[0]?.url).toBe(`/app/g/${user.conversation.groupId}`); + const devices = await prisma.pushSubscription.findMany({ where: { userId: user.userId } }); + expect(devices).toHaveLength(1); + expect(devices[0]).toMatchObject({ endpoint: browser, lastUsedAt: expect.any(Date) }); + }); + + it("lets only the main bot nudge, within the user's settings and the daily cap", async () => { + const user = await newUser(); + const web = recordingSender(); + const deps = { + prisma, + notifications: new PushNotificationProvider(prisma, { webpush: web.sender }), + }; + + await expect( + notifyUserFromTool(deps, toolContext(user, { ...user.main, kind: "specialist" }), { + body: "Hi", + }), + ).resolves.toEqual({ error: "Only the user's main bot can notify the user." }); + await expect(notifyUserFromTool(deps, toolContext(user), {})).resolves.toMatchObject({ + error: expect.stringContaining("body is required"), + }); + await expect( + notifyUserFromTool({ prisma }, toolContext(user), { body: "Hi" }), + ).resolves.toMatchObject({ sent: false }); + await expect(notifyUserFromTool(deps, toolContext(user), { body: "Hi" })).resolves.toEqual({ + sent: false, + reason: "The user has no device with notifications turned on.", + }); + + await savePushSubscription(prisma, user.userId, { + kind: "webpush", + endpoint: `https://push.example.test/${randomUUID()}`, + ...keys, + }); + await updateNotificationSettings(prisma, user.userId, { nudges: false }); + await expect(notifyUserFromTool(deps, toolContext(user), { body: "Hi" })).resolves.toEqual({ + sent: false, + reason: "The user turned nudges off.", + }); + + // Quiet hours around the current time. + await updateNotificationSettings(prisma, user.userId, { + nudges: true, + quietHours: quietAroundNow(), + }); + await expect(notifyUserFromTool(deps, toolContext(user), { body: "Hi" })).resolves.toEqual({ + sent: false, + reason: "It is the user's quiet hours.", + }); + expect(web.sent).toHaveLength(0); + + await updateNotificationSettings(prisma, user.userId, { quietHours: null }); + for (let index = 0; index < 3; index += 1) { + await expect( + notifyUserFromTool(deps, toolContext(user), { body: `Check-in ${index}` }), + ).resolves.toEqual({ sent: true }); + } + await expect( + notifyUserFromTool(deps, toolContext(user), { body: "One too many" }), + ).resolves.toEqual({ sent: false, reason: "The daily limit of 3 nudges is reached." }); + + expect(web.sent).toHaveLength(3); + expect(web.sent[0]).toMatchObject({ + kind: "nudge", + title: user.main.name, + body: "Check-in 0", + threadId: user.conversation.threadId, + url: `/app/g/${user.conversation.groupId}`, + }); + }); + + it("reports a nudge no device accepted", async () => { + const user = await newUser(); + await savePushSubscription(prisma, user.userId, { + kind: "webpush", + endpoint: `https://push.example.test/${randomUUID()}`, + ...keys, + }); + const failing: PushSender = { + send: async () => { + throw new Error("Web push failed (500)"); + }, + }; + await expect( + notifyUserFromTool( + { prisma, notifications: new PushNotificationProvider(prisma, { webpush: failing }) }, + toolContext(user), + { body: "Hi", title: "Evening check-in" }, + ), + ).resolves.toEqual({ sent: false, reason: "No device accepted the notification." }); + }); +}); + +/** A UTC range that covers the current minute (users without a time zone use UTC). */ +function quietAroundNow() { + const now = new Date(); + const minute = now.getUTCHours() * 60 + now.getUTCMinutes(); + const clock = (value: number) => { + const wrapped = (value + 1440) % 1440; + return `${String(Math.floor(wrapped / 60)).padStart(2, "0")}:${String(wrapped % 60).padStart(2, "0")}`; + }; + return { start: clock(minute - 60), end: clock(minute + 60) }; +} diff --git a/packages/adapters/src/push-notifications.test.ts b/packages/adapters/src/push-notifications.test.ts new file mode 100644 index 0000000..e457023 --- /dev/null +++ b/packages/adapters/src/push-notifications.test.ts @@ -0,0 +1,183 @@ +import type { NotificationMessage } from "@rakazo/adapter-kit"; +import type { PrismaClient } from "@rakazo/db"; +import { describe, expect, it, vi } from "vitest"; +import { + PushNotificationProvider, + type PushOutcome, + type PushPayload, + type PushSender, +} from "./push-notifications.js"; + +const context = { + operationId: "notify", + traceId: "t", + spaceId: "space-1", + userId: "user-1", + signal: new AbortController().signal, +}; +const message: NotificationMessage = { + kind: "completion", + title: "Chief finished", + body: "Booked the table for Friday.", + botId: "bot-1", + threadId: "thread-1", +}; + +function fakeDb(options: { + devices: Array<{ id: string; kind: "webpush" | "expo" }>; + quietHours?: { start: number; end: number }; + timeZone?: string; + groupId?: string | null; +}) { + const forgotten: string[] = []; + const used: string[] = []; + const prisma = { + userNotificationSettings: { + findUnique: async () => + options.quietHours + ? { + quietHoursStart: options.quietHours.start, + quietHoursEnd: options.quietHours.end, + nudges: true, + } + : null, + }, + userProfile: { findUnique: async () => ({ timeZone: options.timeZone ?? null }) }, + pushSubscription: { + findMany: async () => + options.devices.map((device) => ({ + ...device, + endpoint: `endpoint-${device.id}`, + p256dh: null, + auth: null, + })), + deleteMany: async ({ where }: { where: { id: string } }) => { + forgotten.push(where.id); + return { count: 1 }; + }, + updateMany: async ({ where }: { where: { id: string } }) => { + used.push(where.id); + return { count: 1 }; + }, + }, + thread: { findUnique: async () => ({ groupId: options.groupId ?? null }) }, + } as unknown as PrismaClient; + return { prisma, forgotten, used }; +} + +function fakeSender(outcome: PushOutcome | Error) { + const sent: PushPayload[] = []; + const sender: PushSender = { + send: vi.fn(async (_target, payload) => { + sent.push(payload); + if (outcome instanceof Error) throw outcome; + return outcome; + }), + }; + return { sender, sent }; +} + +// 23:30 in Vienna (UTC+2 in summer). +const lateEvening = () => new Date("2026-07-01T21:30:00Z"); + +describe("push notification fan-out", () => { + it("sends to every device with a link to the conversation", async () => { + const { prisma, used } = fakeDb({ + devices: [ + { id: "browser", kind: "webpush" }, + { id: "phone", kind: "expo" }, + ], + groupId: "group-1", + }); + const web = fakeSender("sent"); + const expo = fakeSender("sent"); + await new PushNotificationProvider(prisma, { webpush: web.sender, expo: expo.sender }).send( + message, + context, + ); + expect(web.sent).toEqual([{ ...message, url: "/app/g/group-1" }]); + expect(expo.sent).toHaveLength(1); + expect(used.sort()).toEqual(["browser", "phone"]); + }); + + it("links a direct thread to its bot and shortens long bodies", async () => { + const { prisma } = fakeDb({ devices: [{ id: "browser", kind: "webpush" }] }); + const web = fakeSender("sent"); + await new PushNotificationProvider(prisma, { webpush: web.sender }).send( + { ...message, body: "x".repeat(1_000) }, + context, + ); + expect(web.sent[0]?.url).toBe("/app/bot-1"); + expect(web.sent[0]?.body.length).toBeLessThanOrEqual(240); + }); + + it("holds every push back during quiet hours in the user's time zone", async () => { + const quiet = fakeDb({ + devices: [{ id: "browser", kind: "webpush" }], + quietHours: { start: 22 * 60, end: 7 * 60 }, + timeZone: "Europe/Vienna", + }); + const web = fakeSender("sent"); + await new PushNotificationProvider(quiet.prisma, { webpush: web.sender }, lateEvening).send( + message, + context, + ); + expect(web.sent).toHaveLength(0); + + // The same moment is mid-afternoon in Los Angeles. + const awake = fakeDb({ + devices: [{ id: "browser", kind: "webpush" }], + quietHours: { start: 22 * 60, end: 7 * 60 }, + timeZone: "America/Los_Angeles", + }); + await new PushNotificationProvider(awake.prisma, { webpush: web.sender }, lateEvening).send( + message, + context, + ); + expect(web.sent).toHaveLength(1); + }); + + it("drops devices the push service no longer knows", async () => { + const { prisma, forgotten, used } = fakeDb({ + devices: [ + { id: "old-browser", kind: "webpush" }, + { id: "phone", kind: "expo" }, + ], + }); + await new PushNotificationProvider(prisma, { + webpush: fakeSender("gone").sender, + expo: fakeSender("sent").sender, + }).send(message, context); + expect(forgotten).toEqual(["old-browser"]); + expect(used).toEqual(["phone"]); + }); + + it("succeeds when one device works and fails only when all of them fail", async () => { + const mixed = fakeDb({ + devices: [ + { id: "browser", kind: "webpush" }, + { id: "phone", kind: "expo" }, + ], + }); + await expect( + new PushNotificationProvider(mixed.prisma, { + webpush: fakeSender(new Error("Web push failed (500)")).sender, + expo: fakeSender("sent").sender, + }).send(message, context), + ).resolves.toBeUndefined(); + + const broken = fakeDb({ devices: [{ id: "browser", kind: "webpush" }] }); + await expect( + new PushNotificationProvider(broken.prisma, { + webpush: fakeSender(new Error("Web push failed (500)")).sender, + }).send(message, context), + ).rejects.toThrow("Push delivery failed"); + }); + + it("does nothing without devices or senders", async () => { + const { prisma } = fakeDb({ devices: [{ id: "phone", kind: "expo" }] }); + const web = fakeSender("sent"); + await new PushNotificationProvider(prisma, { webpush: web.sender }).send(message, context); + expect(web.sent).toHaveLength(0); + }); +}); diff --git a/packages/adapters/src/push-notifications.ts b/packages/adapters/src/push-notifications.ts new file mode 100644 index 0000000..820a132 --- /dev/null +++ b/packages/adapters/src/push-notifications.ts @@ -0,0 +1,102 @@ +import type { + AdapterContext, + NotificationMessage, + NotificationProvider, +} from "@rakazo/adapter-kit"; +import type { PrismaClient, PushSubscriptionKind } from "@rakazo/db"; +import { + forgetPushSubscription, + listPushSubscriptions, + loadNotificationPolicy, + markPushSubscriptionUsed, +} from "@rakazo/db"; +import { getLogger } from "@rakazo/logging"; + +/** Lock screens show a line or two; the rest lives in the conversation. */ +const PUSH_BODY_MAX_LENGTH = 240; + +/** What a device shows: the notification plus where a tap leads (a path in the web app). */ +export interface PushPayload extends NotificationMessage { + url: string; +} + +/** "gone": the push service no longer knows this device, so its registration can be dropped. */ +export type PushOutcome = "sent" | "gone"; + +export interface PushTarget { + endpoint: string; + p256dh: string | null; + auth: string | null; +} + +export interface PushSender { + send(target: PushTarget, payload: PushPayload, signal?: AbortSignal): Promise; +} + +function clip(text: string): string { + const trimmed = text.trim(); + return trimmed.length > PUSH_BODY_MAX_LENGTH + ? `${trimmed.slice(0, PUSH_BODY_MAX_LENGTH - 1).trimEnd()}…` + : trimmed; +} + +/** + * Delivers one notification to every device the user registered (browsers and the mobile app), + * except during their quiet hours. Devices the push service reports gone are dropped. Throws only + * when every device failed. + */ +export class PushNotificationProvider implements NotificationProvider { + constructor( + private readonly prisma: PrismaClient, + private readonly senders: Partial>, + private readonly now: () => Date = () => new Date(), + ) {} + + describe() { + return { + id: "push", + contractVersion: "1", + adapterVersion: "0.1.0", + capabilities: { push: true, email: false }, + }; + } + + async send(message: NotificationMessage, context: AdapterContext): Promise { + const policy = await loadNotificationPolicy(this.prisma, context.userId, this.now()); + if (policy.quiet) return; + const devices = (await listPushSubscriptions(this.prisma, context.userId)).filter( + (device) => this.senders[device.kind], + ); + if (devices.length === 0) return; + const payload: PushPayload = { + ...message, + title: message.title.trim().slice(0, 120), + body: clip(message.body), + url: await this.conversationPath(message), + }; + const results = await Promise.allSettled( + devices.map(async (device) => { + const outcome = await this.senders[device.kind]!.send(device, payload, context.signal); + if (outcome === "gone") await forgetPushSubscription(this.prisma, device.id); + else await markPushSubscriptionUsed(this.prisma, device.id, this.now()); + }), + ); + const failures = results.filter((result) => result.status === "rejected"); + for (const failure of failures) { + // Errors carry a status or a reason, never the endpoint or token. + getLogger().error("push delivery failed", failure.reason); + } + if (failures.length === devices.length) throw new Error("Push delivery failed"); + } + + /** The web app path of the conversation the notification is about. */ + private async conversationPath(message: NotificationMessage): Promise { + const thread = await this.prisma.thread.findUnique({ + where: { id: message.threadId }, + select: { groupId: true }, + }); + return thread?.groupId + ? `/app/g/${encodeURIComponent(thread.groupId)}` + : `/app/${encodeURIComponent(message.botId)}`; + } +} diff --git a/packages/adapters/src/web-push.test.ts b/packages/adapters/src/web-push.test.ts new file mode 100644 index 0000000..df381de --- /dev/null +++ b/packages/adapters/src/web-push.test.ts @@ -0,0 +1,156 @@ +import { createECDH, randomBytes } from "node:crypto"; +import type { PrismaClient } from "@rakazo/db"; +import { describe, expect, it, vi } from "vitest"; +import webPush from "web-push"; +import type { PushPayload } from "./push-notifications.js"; +import { EncryptedSecretStore } from "./secrets.js"; +import { VapidKeys, vapidEnvFromProcess, WebPushSender } from "./web-push.js"; + +const browserKey = createECDH("prime256v1"); +browserKey.generateKeys(); +const device = { + endpoint: "https://push.example.test/device-1", + p256dh: browserKey.getPublicKey().toString("base64url"), + auth: randomBytes(16).toString("base64url"), +}; +const publicHost = async () => [{ address: "93.184.216.34", family: 4 }]; +const payload: PushPayload = { + kind: "completion", + title: "Chief finished", + body: "Done", + botId: "bot-1", + threadId: "thread-1", + url: "/app/g/group-1", +}; + +function settingsTable() { + const rows = new Map(); + const prisma = { + integrationProviderConfig: { + findUnique: vi.fn(async ({ where }: { where: { id: string } }) => + rows.has(where.id) ? { ciphertext: rows.get(where.id)! } : null, + ), + createMany: vi.fn(async ({ data }: { data: Array<{ id: string; ciphertext: string }> }) => { + for (const row of data) if (!rows.has(row.id)) rows.set(row.id, row.ciphertext); + return { count: data.length }; + }), + }, + } as unknown as PrismaClient; + return { prisma, rows }; +} + +const fixedKeys = { subject: "mailto:ops@example.test", ...webPush.generateVAPIDKeys() }; + +describe("VAPID keys", () => { + it("uses the configured key pair", async () => { + const { prisma, rows } = settingsTable(); + const keys = new VapidKeys(prisma, new EncryptedSecretStore("test-key"), fixedKeys); + await expect(keys.details()).resolves.toEqual(fixedKeys); + expect(rows.size).toBe(0); + }); + + it("generates one key pair, stores it encrypted and shares it across processes", async () => { + const { prisma, rows } = settingsTable(); + const secrets = new EncryptedSecretStore("test-key"); + const api = new VapidKeys(prisma, secrets, { publicKey: fixedKeys.publicKey }); + const worker = new VapidKeys(prisma, secrets); + const [fromApi, fromWorker] = await Promise.all([api.details(), worker.details()]); + expect(fromWorker).toEqual(fromApi); + expect(fromApi.publicKey).not.toBe(fixedKeys.publicKey); + expect(fromApi.subject).toBe("mailto:admin@localhost"); + expect(rows.size).toBe(1); + const [ciphertext] = [...rows.values()]; + expect(ciphertext).not.toContain(fromApi.privateKey); + await expect(new VapidKeys(prisma, secrets).publicKey()).resolves.toBe(fromApi.publicKey); + }); + + it("reports no public key when the configured keys are invalid", async () => { + const { prisma } = settingsTable(); + const keys = new VapidKeys(prisma, new EncryptedSecretStore("test-key"), { + ...fixedKeys, + privateKey: "not-a-key", + }); + await expect(keys.publicKey()).resolves.toBeNull(); + }); + + it("reads the environment and uses an https web origin as the contact", () => { + expect( + vapidEnvFromProcess({ + VAPID_PUBLIC_KEY: " pub ", + VAPID_PRIVATE_KEY: "priv", + WEB_ORIGIN: "https://app.example.test", + }), + ).toEqual({ publicKey: "pub", privateKey: "priv", subject: "https://app.example.test" }); + expect(vapidEnvFromProcess({ WEB_ORIGIN: "http://127.0.0.1:5173" }).subject).toBeUndefined(); + expect( + vapidEnvFromProcess({ VAPID_SUBJECT: "mailto:a@example.test", WEB_ORIGIN: "https://x.test" }) + .subject, + ).toBe("mailto:a@example.test"); + }); +}); + +describe("web push sender", () => { + const vapid = { details: async () => fixedKeys }; + + it("posts an encrypted, VAPID-signed message to the endpoint", async () => { + const fetchMock = vi.fn(async () => new Response(null, { status: 201 })); + const sender = new WebPushSender(vapid, { fetch: fetchMock, resolveHostname: publicHost }); + await expect(sender.send(device, payload)).resolves.toBe("sent"); + expect(fetchMock).toHaveBeenCalledOnce(); + const [url, init] = fetchMock.mock.calls[0] as unknown as [string, RequestInit]; + expect(url).toBe(device.endpoint); + expect(init.method).toBe("POST"); + expect(init.redirect).toBe("manual"); + const headers = new Headers(init.headers); + expect(headers.get("content-encoding")).toBe("aes128gcm"); + expect(headers.get("ttl")).toBe("86400"); + expect(headers.get("authorization")).toMatch(/^vapid t=.+, k=/); + expect(headers.has("content-length")).toBe(false); + // The body is ciphertext: the notification text never travels in the clear. + expect(Buffer.from(init.body as Uint8Array).toString("utf8")).not.toContain("Chief finished"); + }); + + it("reports an expired subscription as gone", async () => { + for (const status of [404, 410]) { + const sender = new WebPushSender(vapid, { + fetch: async () => new Response(null, { status }), + resolveHostname: publicHost, + }); + await expect(sender.send(device, payload)).resolves.toBe("gone"); + } + }); + + it("throws on other failures without naming the endpoint", async () => { + const sender = new WebPushSender(vapid, { + fetch: async () => new Response("quota exceeded", { status: 429 }), + resolveHostname: publicHost, + }); + const error = await sender.send(device, payload).catch((caught: Error) => caught); + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toBe("Web push failed (429)"); + expect((error as Error).message).not.toContain("push.example.test"); + }); + + it("never connects to a private or non-https endpoint", async () => { + const fetchMock = vi.fn(); + const privateHost = new WebPushSender(vapid, { + fetch: fetchMock, + resolveHostname: async () => [{ address: "10.0.0.5", family: 4 }], + }); + await expect(privateHost.send(device, payload)).rejects.toThrow("private address"); + const plainHttp = new WebPushSender(vapid, { fetch: fetchMock, resolveHostname: publicHost }); + await expect( + plainHttp.send({ ...device, endpoint: "http://push.example.test/device-1" }, payload), + ).rejects.toThrow("https"); + const loopback = new WebPushSender(vapid, { fetch: fetchMock, resolveHostname: publicHost }); + await expect( + loopback.send({ ...device, endpoint: "https://127.0.0.1/device-1" }, payload), + ).rejects.toThrow("private or internal host"); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it("treats a subscription without keys as gone", async () => { + const sender = new WebPushSender(vapid, { fetch: vi.fn(), resolveHostname: publicHost }); + await expect(sender.send({ ...device, p256dh: null }, payload)).resolves.toBe("gone"); + }); +}); diff --git a/packages/adapters/src/web-push.ts b/packages/adapters/src/web-push.ts index 3abb071..e851d9e 100644 --- a/packages/adapters/src/web-push.ts +++ b/packages/adapters/src/web-push.ts @@ -1,11 +1,11 @@ import { lookup } from "node:dns/promises"; -import type { NotificationMessage } from "@rakazo/adapter-kit"; import type { PrismaClient } from "@rakazo/db"; import { getLogger } from "@rakazo/logging"; import { Agent } from "undici"; import webPush from "web-push"; import { combineSignals } from "./connector-safety.js"; import { createAddressCheckedLookup, type ResolveHostname } from "./network-address.js"; +import type { PushOutcome, PushPayload, PushTarget } from "./push-notifications.js"; import type { EncryptedSecretStore } from "./secrets.js"; import { dispatcherFetch } from "./undici-fetch.js"; import { assertPublicAddresses, assertSafeWebUrl } from "./web-ssrf.js"; @@ -137,20 +137,6 @@ export class VapidKeys { } } -/** What a device shows: the notification plus where a tap leads (a path in the web app). */ -export interface PushPayload extends NotificationMessage { - url: string; -} - -/** "gone": the push service no longer knows this device, so its registration can be dropped. */ -export type PushOutcome = "sent" | "gone"; - -export interface PushTarget { - endpoint: string; - p256dh: string | null; - auth: string | null; -} - export interface WebPushSenderOptions { fetch?: typeof globalThis.fetch; resolveHostname?: ResolveHostname; diff --git a/packages/contracts/src/notifications.ts b/packages/contracts/src/notifications.ts index 5b09a13..bcb7cc3 100644 --- a/packages/contracts/src/notifications.ts +++ b/packages/contracts/src/notifications.ts @@ -41,6 +41,13 @@ export const WebPushEndpointSchema = z } }, "Endpoint must be an https URL"); +/** The mobile app's Expo push token, e.g. "ExponentPushToken[xxxxxxxx]". */ +export const ExpoPushTokenSchema = z + .string() + .trim() + .max(512) + .regex(/^Expo(nent)?PushToken\[[^\]\s]+\]$/, "Invalid Expo push token"); + /** The JSON form of a browser PushSubscription (RFC 8291 keys, base64url). */ export const WebPushSubscriptionInput = z.object({ endpoint: WebPushEndpointSchema, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 96a8a3c..249f518 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -84,6 +84,7 @@ import { } from "./integration-settings.js"; import { ModelTierMappingSchema, SetModelTierInput } from "./model-tiers.js"; import { + ExpoPushTokenSchema, NotificationSettingsSchema, UpdateNotificationSettingsInput, WebPushConfigSchema, @@ -811,7 +812,7 @@ export const appContract = { }, notifications: { registerPush: oc - .input(z.object({ token: z.string().min(8).max(512) })) + .input(z.object({ token: ExpoPushTokenSchema })) .output(z.object({ ok: z.literal(true) })), unregisterPush: oc.output(z.object({ ok: z.literal(true) })), settings: { diff --git a/packages/core/src/quiet-hours.test.ts b/packages/core/src/quiet-hours.test.ts new file mode 100644 index 0000000..183ee65 --- /dev/null +++ b/packages/core/src/quiet-hours.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, it } from "vitest"; +import { isQuietMinute, isQuietTime, localDateAndMinute } from "./quiet-hours.js"; + +const minute = (clock: string) => { + const [hour, min] = clock.split(":").map(Number); + return hour! * 60 + min!; +}; + +describe("quiet hours", () => { + it("covers a same-day range with an inclusive start and an exclusive end", () => { + const lunch = { start: minute("12:00"), end: minute("13:30") }; + expect(isQuietMinute(lunch, minute("11:59"))).toBe(false); + expect(isQuietMinute(lunch, minute("12:00"))).toBe(true); + expect(isQuietMinute(lunch, minute("13:29"))).toBe(true); + expect(isQuietMinute(lunch, minute("13:30"))).toBe(false); + }); + + it("covers a range that crosses midnight", () => { + const night = { start: minute("22:00"), end: minute("07:00") }; + expect(isQuietMinute(night, minute("21:59"))).toBe(false); + expect(isQuietMinute(night, minute("22:00"))).toBe(true); + expect(isQuietMinute(night, minute("23:59"))).toBe(true); + expect(isQuietMinute(night, minute("00:00"))).toBe(true); + expect(isQuietMinute(night, minute("06:59"))).toBe(true); + expect(isQuietMinute(night, minute("07:00"))).toBe(false); + expect(isQuietMinute(night, minute("12:00"))).toBe(false); + }); + + it("is never quiet without a range or with an empty one", () => { + expect(isQuietMinute(null, minute("03:00"))).toBe(false); + expect(isQuietMinute({ start: 300, end: 300 }, 300)).toBe(false); + }); + + it("reads the wall clock in the user's time zone", () => { + // 21:30 UTC is 23:30 in Vienna (CEST, UTC+2) and 14:30 in Los Angeles (PDT, UTC-7). + const now = new Date("2026-07-01T21:30:00Z"); + const night = { start: minute("22:00"), end: minute("07:00") }; + expect(isQuietTime(night, "Europe/Vienna", now)).toBe(true); + expect(isQuietTime(night, "America/Los_Angeles", now)).toBe(false); + expect(isQuietTime(night, null, now)).toBe(false); + }); + + it("falls back to UTC for an unknown zone and reports the local date", () => { + const now = new Date("2026-01-31T23:15:00Z"); + expect(localDateAndMinute(now, "Not/AZone")).toEqual({ + date: "2026-01-31", + minute: minute("23:15"), + }); + // Already the next day in Tokyo (UTC+9). + expect(localDateAndMinute(now, "Asia/Tokyo")).toEqual({ + date: "2026-02-01", + minute: minute("08:15"), + }); + }); + + it("follows daylight saving changes", () => { + const night = { start: minute("22:00"), end: minute("07:00") }; + // 05:30 UTC is 06:30 in Vienna in winter (CET) and 07:30 in summer (CEST). + expect(isQuietTime(night, "Europe/Vienna", new Date("2026-01-15T05:30:00Z"))).toBe(true); + expect(isQuietTime(night, "Europe/Vienna", new Date("2026-07-15T05:30:00Z"))).toBe(false); + }); +}); diff --git a/packages/db/prisma/migrations/20261003123000_push_subscriptions_and_notification_settings/migration.sql b/packages/db/prisma/migrations/20261003123000_push_subscriptions_and_notification_settings/migration.sql index c684498..cd83c49 100644 --- a/packages/db/prisma/migrations/20261003123000_push_subscriptions_and_notification_settings/migration.sql +++ b/packages/db/prisma/migrations/20261003123000_push_subscriptions_and_notification_settings/migration.sql @@ -44,7 +44,9 @@ ALTER TABLE "push_subscriptions" ADD CONSTRAINT "push_subscriptions_webpush_keys ALTER TABLE "user_notification_settings" ADD CONSTRAINT "user_notification_settings_quiet_hours_chk" CHECK ( ("quietHoursStart" IS NULL AND "quietHoursEnd" IS NULL) OR ( - "quietHoursStart" BETWEEN 0 AND 1439 + "quietHoursStart" IS NOT NULL + AND "quietHoursEnd" IS NOT NULL + AND "quietHoursStart" BETWEEN 0 AND 1439 AND "quietHoursEnd" BETWEEN 0 AND 1439 AND "quietHoursStart" <> "quietHoursEnd" ) diff --git a/packages/db/src/notification-settings.postgres.test.ts b/packages/db/src/notification-settings.postgres.test.ts new file mode 100644 index 0000000..624f9dc --- /dev/null +++ b/packages/db/src/notification-settings.postgres.test.ts @@ -0,0 +1,184 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import type { PrismaClient } from "./client.js"; +import { createDb } from "./client.js"; +import { + claimDailyNudge, + getNotificationSettings, + loadNotificationPolicy, + updateNotificationSettings, +} from "./notification-settings.js"; +import { + listPushSubscriptions, + PUSH_SUBSCRIPTION_LIMIT, + removePushSubscription, + removePushSubscriptionsOfKind, + savePushSubscription, +} from "./push-subscriptions.js"; + +const databaseUrl = process.env.DATABASE_URL; +const describePostgres = + process.env.VERIFY_DATABASE && databaseUrl ? describe.sequential : describe.skip; + +const keys = { p256dh: `B${"A".repeat(86)}`, auth: "A".repeat(22) }; + +describePostgres("notification settings and push subscriptions (PostgreSQL)", () => { + let prisma: PrismaClient; + let close: () => Promise; + const userIds: string[] = []; + + async function newUser() { + const id = `ns${randomUUID().replaceAll("-", "")}`; + userIds.push(id); + await prisma.user.create({ data: { id, name: "Ada", email: `${id}@rakazo.test` } }); + return id; + } + + beforeAll(() => { + const db = createDb(databaseUrl!); + prisma = db.prisma; + close = async () => { + await db.prisma.$disconnect(); + await db.pool.end(); + }; + }); + + afterAll(async () => { + if (!prisma) return; + await prisma.user.deleteMany({ where: { id: { in: userIds } } }); + await close(); + }); + + it("starts with no quiet hours and nudges on, and applies partial updates", async () => { + const userId = await newUser(); + await expect(getNotificationSettings(prisma, userId)).resolves.toEqual({ + quietHours: null, + nudges: true, + }); + + await expect( + updateNotificationSettings(prisma, userId, { quietHours: { start: "22:30", end: "07:00" } }), + ).resolves.toEqual({ quietHours: { start: "22:30", end: "07:00" }, nudges: true }); + const row = await prisma.userNotificationSettings.findUniqueOrThrow({ where: { userId } }); + expect([row.quietHoursStart, row.quietHoursEnd]).toEqual([22 * 60 + 30, 7 * 60]); + + await expect(updateNotificationSettings(prisma, userId, { nudges: false })).resolves.toEqual({ + quietHours: { start: "22:30", end: "07:00" }, + nudges: false, + }); + await expect(updateNotificationSettings(prisma, userId, { quietHours: null })).resolves.toEqual( + { quietHours: null, nudges: false }, + ); + }); + + it("rejects an empty quiet-hours range in the repository and in the database", async () => { + const userId = await newUser(); + await expect( + updateNotificationSettings(prisma, userId, { quietHours: { start: "08:00", end: "08:00" } }), + ).rejects.toThrow(); + await expect( + prisma.userNotificationSettings.create({ + data: { userId, quietHoursStart: 60, quietHoursEnd: null }, + }), + ).rejects.toThrow(); + await expect( + prisma.userNotificationSettings.create({ + data: { userId, quietHoursStart: 1440, quietHoursEnd: 60 }, + }), + ).rejects.toThrow(); + }); + + it("judges quiet hours in the profile time zone", async () => { + const userId = await newUser(); + await updateNotificationSettings(prisma, userId, { + quietHours: { start: "22:00", end: "07:00" }, + }); + const now = new Date("2026-07-01T21:30:00Z"); + await expect(loadNotificationPolicy(prisma, userId, now)).resolves.toMatchObject({ + quiet: false, + localDate: "2026-07-01", + }); + await prisma.userProfile.create({ data: { userId, timeZone: "Europe/Vienna" } }); + await expect(loadNotificationPolicy(prisma, userId, now)).resolves.toEqual({ + quiet: true, + nudges: true, + localDate: "2026-07-01", + }); + }); + + it("caps nudges per local day, also under concurrency", async () => { + const userId = await newUser(); + const claims = await Promise.all( + Array.from({ length: 6 }, () => claimDailyNudge(prisma, userId, "2026-07-01")), + ); + expect(claims.filter(Boolean)).toHaveLength(3); + await expect(claimDailyNudge(prisma, userId, "2026-07-01")).resolves.toBe(false); + await expect(claimDailyNudge(prisma, userId, "2026-07-02")).resolves.toBe(true); + const row = await prisma.userNotificationSettings.findUniqueOrThrow({ where: { userId } }); + expect(row.nudgeCount).toBe(1); + // Claiming never changes the user's settings. + await expect(getNotificationSettings(prisma, userId)).resolves.toEqual({ + quietHours: null, + nudges: true, + }); + }); + + it("keeps devices per user, moves a re-registered endpoint and caps the list", async () => { + const owner = await newUser(); + const other = await newUser(); + const endpoint = `https://push.example.test/${randomUUID()}`; + await savePushSubscription(prisma, owner, { kind: "webpush", endpoint, ...keys }); + await savePushSubscription(prisma, owner, { kind: "expo", endpoint: "ExponentPushToken[a]" }); + expect((await listPushSubscriptions(prisma, owner)).map((row) => row.kind)).toEqual([ + "webpush", + "expo", + ]); + + // Another user cannot remove the owner's device by naming its endpoint. + await removePushSubscription(prisma, other, endpoint); + expect(await listPushSubscriptions(prisma, owner)).toHaveLength(2); + + // The same browser signed into another account now delivers to that account. + await savePushSubscription(prisma, other, { kind: "webpush", endpoint, ...keys }); + expect((await listPushSubscriptions(prisma, owner)).map((row) => row.kind)).toEqual(["expo"]); + expect((await listPushSubscriptions(prisma, other)).map((row) => row.endpoint)).toEqual([ + endpoint, + ]); + + await removePushSubscriptionsOfKind(prisma, owner, "expo"); + expect(await listPushSubscriptions(prisma, owner)).toEqual([]); + + for (let index = 0; index <= PUSH_SUBSCRIPTION_LIMIT; index += 1) { + await savePushSubscription(prisma, owner, { + kind: "expo", + endpoint: `ExponentPushToken[${owner}-${index}]`, + }); + } + const kept = await listPushSubscriptions(prisma, owner); + expect(kept).toHaveLength(PUSH_SUBSCRIPTION_LIMIT); + expect(kept.map((row) => row.endpoint)).not.toContain(`ExponentPushToken[${owner}-0]`); + }); + + it("requires keys for web push and removes everything with the account", async () => { + const userId = await newUser(); + await expect( + prisma.pushSubscription.create({ + data: { userId, kind: "webpush", endpoint: `https://push.example.test/${userId}` }, + }), + ).rejects.toThrow(); + await expect( + prisma.pushSubscription.create({ + data: { userId, kind: "sms", endpoint: `sms-${userId}` }, + }), + ).rejects.toThrow(); + await savePushSubscription(prisma, userId, { + kind: "expo", + endpoint: `ExponentPushToken[${userId}]`, + }); + await updateNotificationSettings(prisma, userId, { nudges: false }); + + await prisma.user.delete({ where: { id: userId } }); + expect(await prisma.pushSubscription.count({ where: { userId } })).toBe(0); + expect(await prisma.userNotificationSettings.count({ where: { userId } })).toBe(0); + }); +}); diff --git a/packages/db/src/push-subscriptions.ts b/packages/db/src/push-subscriptions.ts index 885f337..879f705 100644 --- a/packages/db/src/push-subscriptions.ts +++ b/packages/db/src/push-subscriptions.ts @@ -86,3 +86,24 @@ export async function listPushSubscriptions( }); return rows.map((row) => ({ ...row, kind: row.kind as PushSubscriptionKind })); } + +export async function countPushSubscriptions( + prisma: PrismaClient, + userId: string, +): Promise { + return prisma.pushSubscription.count({ where: { userId } }); +} + +/** Records a successful delivery to a device. */ +export async function markPushSubscriptionUsed( + prisma: PrismaClient, + id: string, + at: Date, +): Promise { + await prisma.pushSubscription.updateMany({ where: { id }, data: { lastUsedAt: at } }); +} + +/** Drops a device its push service no longer knows. */ +export async function forgetPushSubscription(prisma: PrismaClient, id: string): Promise { + await prisma.pushSubscription.deleteMany({ where: { id } }); +} diff --git a/packages/testkit/src/cli/harness.ts b/packages/testkit/src/cli/harness.ts index 5305a76..38487bc 100644 --- a/packages/testkit/src/cli/harness.ts +++ b/packages/testkit/src/cli/harness.ts @@ -109,9 +109,11 @@ async function main() { "packages/testkit/src/delegation.postgres.test.ts", "packages/testkit/src/user-profile.postgres.test.ts", "packages/testkit/src/onboarding-discovery.postgres.test.ts", + "packages/testkit/src/notifications.postgres.test.ts", "packages/db/src/space-membership.postgres.test.ts", "packages/db/src/main-agent.postgres.test.ts", "packages/db/src/user-profile.postgres.test.ts", + "packages/db/src/notification-settings.postgres.test.ts", "packages/db/src/messaging.postgres.test.ts", "packages/memory/src/commit.postgres.test.ts", "packages/adapters/src/wakeup.postgres.test.ts", @@ -120,6 +122,7 @@ async function main() { "packages/adapters/src/cloud-agent.postgres.test.ts", "packages/adapters/src/builtin-memory-provider.postgres.test.ts", "packages/adapters/src/profile-consolidation.postgres.test.ts", + "packages/adapters/src/notify-user.postgres.test.ts", ]; // Each app reconciles all durable work in its database, including intentionally // unfinished fixture runs. Clone the pristine migrated schema so one suite diff --git a/packages/testkit/src/notifications.postgres.test.ts b/packages/testkit/src/notifications.postgres.test.ts new file mode 100644 index 0000000..9668046 --- /dev/null +++ b/packages/testkit/src/notifications.postgres.test.ts @@ -0,0 +1,198 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { ComposioEmulator } from "@rakazo/adapters"; +import type { NotificationSettings, WebPushConfig } from "@rakazo/contracts"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import type { createApp } from "../../../apps/api/src/app.ts"; +import { sessionCookieHeader } from "./index.js"; + +type App = { request: (input: string, init?: RequestInit) => Response | Promise }; +type AppHandles = Awaited>; + +process.env.WAKEUP_DRIVER = "memory"; +process.env.SANDBOX_PROVIDER = "fake"; +process.env.AGENT_RUNTIME = "scripted"; + +const hasDb = process.env.VERIFY_DATABASE === "1" && Boolean(process.env.DATABASE_URL); +const describeWithDatabase = hasDb ? describe : describe.skip; + +const keys = { p256dh: `B${"A".repeat(86)}`, auth: "A".repeat(22) }; + +describeWithDatabase("notifications through the API", () => { + let handles: AppHandles; + let app: App; + const stamp = `${Date.now()}-${Math.random().toString(36).slice(2)}`; + const dataDir = mkdtempSync(path.join(tmpdir(), "rakazo-notifications-")); + + beforeAll(async () => { + const { createApp } = await import("../../../apps/api/src/app.ts"); + handles = await createApp({ + databaseUrl: process.env.DATABASE_URL!, + dataDir, + sandboxProvider: "fake", + agentRuntime: "scripted", + wakeupDriver: "memory", + signupsEnabled: "true", + composio: new ComposioEmulator(), + }); + app = handles.app; + }); + + afterAll(async () => { + await handles?.stop(); + rmSync(dataDir, { recursive: true, force: true }); + }); + + async function devices(cookie: string) { + const me = await rpc<{ userId: string }>(app, cookie, "me"); + return handles.prisma.pushSubscription.findMany({ + where: { userId: me.userId }, + orderBy: { createdAt: "asc" }, + select: { kind: true, endpoint: true, userAgent: true }, + }); + } + + it("reads and updates the caller's own settings", async () => { + const owner = await signup(app, `settings-owner-${stamp}@rakazo.test`, "Settings Owner"); + const other = await signup(app, `settings-other-${stamp}@rakazo.test`, "Settings Other"); + + await expect( + rpc(app, owner, "notifications/settings/get"), + ).resolves.toEqual({ quietHours: null, nudges: true }); + await expect( + rpc(app, owner, "notifications/settings/update", { + quietHours: { start: "21:00", end: "06:30" }, + nudges: false, + }), + ).resolves.toEqual({ quietHours: { start: "21:00", end: "06:30" }, nudges: false }); + await expect( + rpc(app, other, "notifications/settings/get"), + ).resolves.toEqual({ quietHours: null, nudges: true }); + + for (const invalid of [ + { quietHours: { start: "9:00", end: "17:00" } }, + { quietHours: { start: "24:00", end: "06:00" } }, + { quietHours: { start: "08:00", end: "08:00" } }, + ]) { + const response = await raw(app, owner, "notifications/settings/update", invalid); + expect(response.status, JSON.stringify(invalid)).toBe(400); + } + await expect( + rpc(app, owner, "notifications/settings/get"), + ).resolves.toEqual({ quietHours: { start: "21:00", end: "06:30" }, nudges: false }); + }); + + it("hands browsers one stable public key", async () => { + const cookie = await signup(app, `push-config-${stamp}@rakazo.test`, "Push Config"); + const first = await rpc(app, cookie, "notifications/webPush/config"); + expect(first.available).toBe(true); + expect(first.publicKey).toMatch(/^[A-Za-z0-9_-]{87}$/); + await expect(rpc(app, cookie, "notifications/webPush/config")).resolves.toEqual( + first, + ); + const stored = await handles.prisma.integrationProviderConfig.findUniqueOrThrow({ + where: { id: "web-push-vapid" }, + }); + expect(stored.ciphertext).not.toContain(first.publicKey!); + }); + + it("registers browsers and phones per user, and only the owner removes them", async () => { + const owner = await signup(app, `push-owner-${stamp}@rakazo.test`, "Push Owner"); + const intruder = await signup(app, `push-intruder-${stamp}@rakazo.test`, "Push Intruder"); + const endpoint = `https://push.example.test/${stamp}`; + + await rpc(app, owner, "notifications/webPush/subscribe", { endpoint, keys }); + await rpc(app, owner, "notifications/registerPush", { token: `ExponentPushToken[${stamp}]` }); + expect(await devices(owner)).toEqual([ + { kind: "webpush", endpoint, userAgent: "" }, + { kind: "expo", endpoint: `ExponentPushToken[${stamp}]`, userAgent: "" }, + ]); + + await rpc(app, intruder, "notifications/webPush/unsubscribe", { endpoint }); + await rpc(app, intruder, "notifications/unregisterPush"); + expect(await devices(owner)).toHaveLength(2); + + await rpc(app, owner, "notifications/unregisterPush"); + expect(await devices(owner)).toEqual([{ kind: "webpush", endpoint, userAgent: "" }]); + await rpc(app, owner, "notifications/webPush/unsubscribe", { endpoint }); + expect(await devices(owner)).toEqual([]); + }); + + it("rejects endpoints that are not public https push services", async () => { + const cookie = await signup(app, `push-invalid-${stamp}@rakazo.test`, "Push Invalid"); + for (const endpoint of [ + "http://push.example.test/device", + "https://127.0.0.1/device", + "https://localhost/device", + "https://10.0.0.8/device", + "https://user:secret@push.example.test/device", + "not a url", + ]) { + const response = await raw(app, cookie, "notifications/webPush/subscribe", { + endpoint, + keys, + }); + expect(response.status, endpoint).toBe(400); + } + const badKeys = await raw(app, cookie, "notifications/webPush/subscribe", { + endpoint: `https://push.example.test/${stamp}-keys`, + keys: { p256dh: "short", auth: keys.auth }, + }); + expect(badKeys.status).toBe(400); + const badToken = await raw(app, cookie, "notifications/registerPush", { + token: "https://push.example.test/not-an-expo-token", + }); + expect(badToken.status).toBe(400); + expect(await devices(cookie)).toEqual([]); + }); + + it("refuses every notification procedure without a session", async () => { + for (const [procedure, body] of [ + ["notifications/settings/get", {}], + ["notifications/settings/update", { nudges: false }], + ["notifications/webPush/config", {}], + ["notifications/webPush/subscribe", { endpoint: `https://push.example.test/${stamp}`, keys }], + ["notifications/webPush/unsubscribe", { endpoint: `https://push.example.test/${stamp}` }], + ["notifications/registerPush", { token: "ExponentPushToken[anon]" }], + ["notifications/unregisterPush", {}], + ] as const) { + const response = await raw(app, "", procedure, body); + expect(response.status, procedure).toBe(401); + } + }); +}); + +async function signup(app: App, email: string, name: string) { + const response = await app.request("/api/auth/sign-up/email", { + method: "POST", + headers: { "content-type": "application/json", origin: "http://127.0.0.1:5173" }, + body: JSON.stringify({ email, password: "password12", name }), + }); + if (response.status >= 400) { + throw new Error(`signup failed ${response.status}: ${await response.text()}`); + } + return sessionCookieHeader(response); +} + +async function raw(app: App, cookie: string, procedure: string, body: unknown = {}) { + return app.request(`/rpc/${procedure}`, { + method: "POST", + headers: { + "content-type": "application/json", + ...(cookie ? { cookie } : {}), + origin: "http://127.0.0.1:5173", + }, + body: JSON.stringify({ json: body }), + }); +} + +async function rpc(app: App, cookie: string, procedure: string, body: unknown = {}) { + const response = await raw(app, cookie, procedure, body); + const text = await response.text(); + const payload = JSON.parse(text) as { json?: T; error?: { message?: string } }; + if (response.status >= 400 || payload.error) { + throw new Error(`${procedure} ${response.status}: ${payload.error?.message ?? text}`); + } + return payload.json as T; +}