Merge web push, quiet hours and nudges
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u
This commit is contained in:
Claude
commit
245efc88b1
@ -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=
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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: {
|
||||
|
||||
@ -242,6 +242,7 @@ export async function openUserSettings(
|
||||
| "memory"
|
||||
| "profile"
|
||||
| "voice"
|
||||
| "notifications"
|
||||
| "usage"
|
||||
| "computer"
|
||||
| "updates",
|
||||
|
||||
129
apps/web/e2e/notification-settings.spec.ts
Normal file
129
apps/web/e2e/notification-settings.spec.ts
Normal file
@ -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();
|
||||
});
|
||||
197
apps/web/src/components/NotificationSettingsPanel.tsx
Normal file
197
apps/web/src/components/NotificationSettingsPanel.tsx
Normal file
@ -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 (
|
||||
<li className="py-3 first:pt-0 last:pb-0">
|
||||
<div className="flex items-center justify-between gap-4">
|
||||
<span id={labelId} className="text-ui-body text-foreground">
|
||||
{label}
|
||||
</span>
|
||||
<Switch
|
||||
data-testid={testId}
|
||||
aria-labelledby={labelId}
|
||||
checked={checked}
|
||||
disabled={disabled}
|
||||
onCheckedChange={onCheckedChange}
|
||||
/>
|
||||
</div>
|
||||
{children}
|
||||
</li>
|
||||
);
|
||||
}
|
||||
|
||||
/** 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<NotificationSettings | null>(null);
|
||||
const [push, setPush] = useState<PushState>("loading");
|
||||
const [pushBusy, setPushBusy] = useState(false);
|
||||
const [error, setError] = useState<string | null>(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 (
|
||||
<div data-testid="notification-settings" className="space-y-5">
|
||||
{error ? (
|
||||
<p role="alert" className="text-ui-body-sm text-destructive">
|
||||
{error}
|
||||
</p>
|
||||
) : null}
|
||||
{settings ? (
|
||||
<section className="rounded-xl border border-border px-4 py-4">
|
||||
<ul className="divide-y divide-border">
|
||||
{pushSupported ? (
|
||||
<SettingRow
|
||||
label={<Trans>Push on this device</Trans>}
|
||||
testId="notification-push-switch"
|
||||
checked={push === "on"}
|
||||
disabled={push === "loading" || push === "blocked" || pushBusy}
|
||||
onCheckedChange={(next) => void changePush(next)}
|
||||
>
|
||||
{push === "blocked" ? (
|
||||
<p className="mt-1 text-ui-body-sm text-muted-foreground">
|
||||
<Trans>Blocked in your browser settings</Trans>
|
||||
</p>
|
||||
) : null}
|
||||
</SettingRow>
|
||||
) : null}
|
||||
<SettingRow
|
||||
label={<Trans>Quiet hours</Trans>}
|
||||
testId="notification-quiet-hours-switch"
|
||||
checked={settings.quietHours !== null}
|
||||
onCheckedChange={(next) => save({ quietHours: next ? quietDraft : null })}
|
||||
>
|
||||
{settings.quietHours ? (
|
||||
<div className="mt-3 flex items-center gap-2">
|
||||
<Input
|
||||
type="time"
|
||||
data-testid="notification-quiet-hours-start"
|
||||
aria-label={t`Quiet hours start`}
|
||||
value={quietDraft.start}
|
||||
onChange={(event) => changeQuietTime("start", event.target.value)}
|
||||
className="w-36"
|
||||
/>
|
||||
<span aria-hidden="true" className="text-muted-foreground">
|
||||
–
|
||||
</span>
|
||||
<Input
|
||||
type="time"
|
||||
data-testid="notification-quiet-hours-end"
|
||||
aria-label={t`Quiet hours end`}
|
||||
value={quietDraft.end}
|
||||
onChange={(event) => changeQuietTime("end", event.target.value)}
|
||||
className="w-36"
|
||||
/>
|
||||
</div>
|
||||
) : null}
|
||||
</SettingRow>
|
||||
<SettingRow
|
||||
label={<Trans>Nudges from your Oo</Trans>}
|
||||
testId="notification-nudges-switch"
|
||||
checked={settings.nudges}
|
||||
onCheckedChange={(next) => save({ nudges: next })}
|
||||
/>
|
||||
</ul>
|
||||
</section>
|
||||
) : null}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@ -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" ? <UserProfileSettings /> : null}
|
||||
{section === "notifications" ? <NotificationSettingsPanel /> : null}
|
||||
{section === "usage" ? (
|
||||
<UsageSettingsPanel usage={usage} panelRef={usageRef} />
|
||||
) : null}
|
||||
|
||||
@ -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("/");
|
||||
})
|
||||
}
|
||||
>
|
||||
<LogOut className="text-muted-foreground" strokeWidth={1.75} />
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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;
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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<string> = new Set<string>([
|
||||
"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_"))),
|
||||
);
|
||||
|
||||
@ -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));
|
||||
}
|
||||
|
||||
@ -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> = {}): 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<Uint8Array>({
|
||||
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"]);
|
||||
});
|
||||
});
|
||||
|
||||
@ -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<string | undefined> {
|
||||
async function readLegacyToken(file: string): Promise<string | undefined> {
|
||||
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<st
|
||||
}
|
||||
}
|
||||
|
||||
export async function savePushToken(dataDir: string, userId: string, token: string): Promise<void> {
|
||||
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<number> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<PushOutcome> {
|
||||
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";
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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";
|
||||
|
||||
85
packages/adapters/src/notify-user-tool.ts
Normal file
85
packages/adapters/src/notify-user-tool.ts
Normal file
@ -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<string, unknown>,
|
||||
): Promise<NotifyUserResult> {
|
||||
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 };
|
||||
}
|
||||
219
packages/adapters/src/notify-user.postgres.test.ts
Normal file
219
packages/adapters/src/notify-user.postgres.test.ts
Normal file
@ -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<void>;
|
||||
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<ReturnType<typeof newUser>>, 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) };
|
||||
}
|
||||
183
packages/adapters/src/push-notifications.test.ts
Normal file
183
packages/adapters/src/push-notifications.test.ts
Normal file
@ -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);
|
||||
});
|
||||
});
|
||||
102
packages/adapters/src/push-notifications.ts
Normal file
102
packages/adapters/src/push-notifications.ts
Normal file
@ -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<PushOutcome>;
|
||||
}
|
||||
|
||||
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<Record<PushSubscriptionKind, PushSender>>,
|
||||
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<void> {
|
||||
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<string> {
|
||||
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)}`;
|
||||
}
|
||||
}
|
||||
156
packages/adapters/src/web-push.test.ts
Normal file
156
packages/adapters/src/web-push.test.ts
Normal file
@ -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<string, string>();
|
||||
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");
|
||||
});
|
||||
});
|
||||
@ -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;
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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: {
|
||||
|
||||
62
packages/core/src/quiet-hours.test.ts
Normal file
62
packages/core/src/quiet-hours.test.ts
Normal file
@ -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);
|
||||
});
|
||||
});
|
||||
@ -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"
|
||||
)
|
||||
|
||||
184
packages/db/src/notification-settings.postgres.test.ts
Normal file
184
packages/db/src/notification-settings.postgres.test.ts
Normal file
@ -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<void>;
|
||||
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);
|
||||
});
|
||||
});
|
||||
@ -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<number> {
|
||||
return prisma.pushSubscription.count({ where: { userId } });
|
||||
}
|
||||
|
||||
/** Records a successful delivery to a device. */
|
||||
export async function markPushSubscriptionUsed(
|
||||
prisma: PrismaClient,
|
||||
id: string,
|
||||
at: Date,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
await prisma.pushSubscription.deleteMany({ where: { id } });
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
198
packages/testkit/src/notifications.postgres.test.ts
Normal file
198
packages/testkit/src/notifications.postgres.test.ts
Normal file
@ -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<Response> };
|
||||
type AppHandles = Awaited<ReturnType<typeof createApp>>;
|
||||
|
||||
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<NotificationSettings>(app, owner, "notifications/settings/get"),
|
||||
).resolves.toEqual({ quietHours: null, nudges: true });
|
||||
await expect(
|
||||
rpc<NotificationSettings>(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<NotificationSettings>(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<NotificationSettings>(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<WebPushConfig>(app, cookie, "notifications/webPush/config");
|
||||
expect(first.available).toBe(true);
|
||||
expect(first.publicKey).toMatch(/^[A-Za-z0-9_-]{87}$/);
|
||||
await expect(rpc<WebPushConfig>(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<T>(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;
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user