From a094c23b47c6b711a0a57e86dc927355b7ddd86b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 3 Oct 2026 13:19:12 +0000 Subject: [PATCH] feat(notifications): fan push out to every device, honour quiet hours, add notify_user Run notifications reached at most one phone and nothing could hold them back at night. - PushNotificationProvider sends each notification to all of the user's registered devices (web push and Expo), skips everything during quiet hours in the profile time zone, drops devices the push service reports gone (404/410, DeviceNotRegistered), records lastUsedAt and links each push to its conversation. - Expo tokens move from DATA_DIR/push-tokens files into push_subscriptions: the API imports and deletes the legacy files at start; notifications.registerPush/unregisterPush keep their contract for the mobile app (tokens must look like Expo tokens). - NotificationMessage gains the nudge kind, and the main bot gets notify_user: a short push to the main conversation that respects the nudge switch, quiet hours and a cap of three per day. It is withheld from shared messaging channels and from runs answering another owner's agent. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u --- apps/api/src/app.ts | 16 +- apps/api/src/router.ts | 12 +- apps/worker/src/index.ts | 11 +- packages/adapter-kit/src/types.ts | 2 +- packages/adapters/src/builtin-tools.ts | 22 ++ packages/adapters/src/executor-tools.test.ts | 6 +- packages/adapters/src/executor-tools.ts | 7 +- packages/adapters/src/executor.ts | 10 + packages/adapters/src/expo-push.test.ts | 298 ++++++++----------- packages/adapters/src/expo-push.ts | 137 +++++---- packages/adapters/src/index.ts | 1 + packages/adapters/src/notify-user-tool.ts | 85 ++++++ packages/adapters/src/push-notifications.ts | 89 ++++++ packages/contracts/src/notifications.ts | 7 + packages/contracts/src/rpc.ts | 3 +- packages/db/src/push-subscriptions.ts | 21 ++ 16 files changed, 469 insertions(+), 258 deletions(-) create mode 100644 packages/adapters/src/notify-user-tool.ts create mode 100644 packages/adapters/src/push-notifications.ts 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 b6c7bc6..e08003f 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -44,7 +44,6 @@ import { computerSupportsUpdate, computerUpdateView, createVoiceProvider, - deletePushToken, deploymentAutoReviewDefault, destroyBot, displayBotWorkspacePath, @@ -76,7 +75,6 @@ import { resolveAutoReviewChecker, resolveBotWorkspacePath, sanitizeComposioError, - savePushToken, scheduleComputerControlExpiry, scheduleComputerSleep, screenLeaseIdForRun, @@ -139,6 +137,7 @@ import { projectWorkspaceDto, releaseSpaceDeletionClaim, removePushSubscription, + removePushSubscriptionsOfKind, renewSpaceDeletionClaim, requireMembership, SPACE_DELETION_CLAIM_TIMEOUT_MS, @@ -4763,11 +4762,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/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 d77cb5e..31b0a7a 100644 --- a/packages/adapter-kit/src/types.ts +++ b/packages/adapter-kit/src/types.ts @@ -553,7 +553,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 0ff52fa..b9726f2 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."; @@ -422,6 +423,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 5cb9181..7ec57e1 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, @@ -2573,6 +2574,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..c42f8a1 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 "./web-push.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..c22f416 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 "./web-push.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 ecf21b9..c4393d1 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -100,6 +100,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/push-notifications.ts b/packages/adapters/src/push-notifications.ts new file mode 100644 index 0000000..74f5956 --- /dev/null +++ b/packages/adapters/src/push-notifications.ts @@ -0,0 +1,89 @@ +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"; +import type { PushOutcome, PushPayload, PushTarget } from "./web-push.js"; + +/** Lock screens show a line or two; the rest lives in the conversation. */ +const PUSH_BODY_MAX_LENGTH = 240; + +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/contracts/src/notifications.ts b/packages/contracts/src/notifications.ts index 5cb6287..b82c110 100644 --- a/packages/contracts/src/notifications.ts +++ b/packages/contracts/src/notifications.ts @@ -58,6 +58,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 1dc59cb..d22607a 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, @@ -807,7 +808,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/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 } }); +}