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 } }); +}