From 444e265275942532593d85cd24de5e56288da322 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 10 May 2026 14:49:38 +0800 Subject: [PATCH] refactor(server): inject push notification sender (#872) --- packages/server/src/server/bootstrap.ts | 3 + .../src/server/client-activity.e2e.test.ts | 41 +++++----- .../server/src/server/push/notifications.ts | 29 +++++++ .../server/src/server/push/push-service.ts | 2 +- .../src/server/test-utils/paseo-daemon.ts | 3 + .../websocket-server.notifications.test.ts | 75 ++++++++++--------- .../server/src/server/websocket-server.ts | 16 ++-- 7 files changed, 104 insertions(+), 65 deletions(-) create mode 100644 packages/server/src/server/push/notifications.ts diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index c62c8d05e..37134e0d8 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -119,6 +119,7 @@ import { createConfiguredTerminalManager } from "../terminal/terminal-manager-fa import { createConnectionOfferV2, encodeOfferToFragmentUrl } from "./connection-offer.js"; import { loadOrCreateDaemonKeyPair } from "./daemon-keypair.js"; import { startRelayTransport, type RelayTransportController } from "./relay-transport.js"; +import type { PushNotificationSender } from "./push/notifications.js"; import { getOrCreateServerId } from "./server-id.js"; import { resolveDaemonVersion } from "./daemon-version.js"; import type { AgentClient, AgentProvider } from "./agent/agent-sdk-types.js"; @@ -205,6 +206,7 @@ export interface PaseoDaemonConfig { providerOverrides?: Record; log?: PersistedConfig["log"]; onLifecycleIntent?: (intent: DaemonLifecycleIntent) => void; + pushNotificationSender?: PushNotificationSender; } export interface PaseoDaemon { @@ -842,6 +844,7 @@ export async function createPaseoDaemon( (hostname) => scriptHealthMonitor.getHealthForHostname(hostname), workspaceGitService, github, + config.pushNotificationSender, ); if (relayEnabled) { diff --git a/packages/server/src/server/client-activity.e2e.test.ts b/packages/server/src/server/client-activity.e2e.test.ts index 77770e677..15d65a56c 100644 --- a/packages/server/src/server/client-activity.e2e.test.ts +++ b/packages/server/src/server/client-activity.e2e.test.ts @@ -1,12 +1,19 @@ -import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, test } from "vitest"; import { createTestPaseoDaemon, type TestPaseoDaemon } from "./test-utils/paseo-daemon.js"; import { DaemonClient } from "./test-utils/daemon-client.js"; import type { AgentStreamEventPayload } from "../shared/messages.js"; import type { AgentSnapshotPayload } from "./messages.js"; -import { PushService } from "./push/push-service.js"; -import { PushTokenStore } from "./push/token-store.js"; +import type { PushNotificationSender, PushPayload } from "./push/notifications.js"; import { PRESENCE_THRESHOLD_MS } from "./agent-attention-policy.js"; +class RecordingPushNotificationSender implements PushNotificationSender { + readonly sent: PushPayload[] = []; + + async send(payload: PushPayload): Promise { + this.sent.push(payload); + } +} + /** * Tests for client activity tracking and smart notifications. * @@ -31,23 +38,17 @@ describe("client activity tracking", () => { let daemon: TestPaseoDaemon; let client1: DaemonClient; let client2: DaemonClient; - let sendPushSpy: ReturnType; - let getAllTokensSpy: ReturnType; + let pushNotifications: RecordingPushNotificationSender; beforeEach(async () => { - sendPushSpy = vi.spyOn(PushService.prototype, "sendPush").mockResolvedValue(undefined); - getAllTokensSpy = vi - .spyOn(PushTokenStore.prototype, "getAllTokens") - .mockReturnValue(["ExponentPushToken[activity-test]"]); - daemon = await createTestPaseoDaemon(); + pushNotifications = new RecordingPushNotificationSender(); + daemon = await createTestPaseoDaemon({ pushNotificationSender: pushNotifications }); }); afterEach(async () => { if (client1) await client1.close().catch(() => {}); if (client2) await client2.close().catch(() => {}); await daemon.close(); - sendPushSpy.mockRestore(); - getAllTokensSpy.mockRestore(); }, 30000); async function createClient(): Promise { @@ -205,7 +206,7 @@ describe("client activity tracking", () => { expect(attention.reason).toBe("finished"); expect(attention.shouldNotify).toBe(false); - expect(sendPushSpy).toHaveBeenCalledTimes(1); + expect(pushNotifications.sent).toHaveLength(1); }, 120000); test("notification when no heartbeat received (legacy/new client)", async () => { @@ -225,7 +226,7 @@ describe("client activity tracking", () => { expect(attention.reason).toBe("finished"); expect(attention.shouldNotify).toBe(false); - expect(sendPushSpy).toHaveBeenCalledTimes(1); + expect(pushNotifications.sent).toHaveLength(1); }, 120000); }); @@ -309,7 +310,7 @@ describe("client activity tracking", () => { // No stale client is selected for in-app; push handles the fallback. expect(attention1.shouldNotify).toBe(false); expect(attention2.shouldNotify).toBe(false); - expect(sendPushSpy).toHaveBeenCalledTimes(1); + expect(pushNotifications.sent).toHaveLength(1); }, 120000); test("notifies only the present Electron-style web client when Firefox is stale", async () => { @@ -345,7 +346,7 @@ describe("client activity tracking", () => { expect(attention1.shouldNotify).toBe(false); expect(attention2.shouldNotify).toBe(true); - expect(sendPushSpy).not.toHaveBeenCalled(); + expect(pushNotifications.sent).toEqual([]); }, 120000); }); @@ -465,7 +466,7 @@ describe("client activity tracking", () => { expect(attention1.shouldNotify).toBe(false); expect(attention2.shouldNotify).toBe(true); - expect(sendPushSpy).not.toHaveBeenCalled(); + expect(pushNotifications.sent).toEqual([]); }, 120000); test("notify web when user active on web but looking at different agent", async () => { @@ -543,7 +544,7 @@ describe("client activity tracking", () => { expect(attention1.shouldNotify).toBe(false); expect(attention2.shouldNotify).toBe(false); - expect(sendPushSpy).toHaveBeenCalledTimes(1); + expect(pushNotifications.sent).toHaveLength(1); }, 120000); }); @@ -581,7 +582,7 @@ describe("client activity tracking", () => { expect(attention1.shouldNotify).toBe(false); expect(attention2.shouldNotify).toBe(false); - expect(sendPushSpy).toHaveBeenCalledTimes(1); + expect(pushNotifications.sent).toHaveLength(1); }, 120000); test("notification when app not visible but activity is recent", async () => { @@ -647,7 +648,7 @@ describe("client activity tracking", () => { expect(attention1.shouldNotify).toBe(true); expect(attention2.shouldNotify).toBe(false); - expect(sendPushSpy).not.toHaveBeenCalled(); + expect(pushNotifications.sent).toEqual([]); }, 120000); }); }); diff --git a/packages/server/src/server/push/notifications.ts b/packages/server/src/server/push/notifications.ts new file mode 100644 index 000000000..07c3a5bcd --- /dev/null +++ b/packages/server/src/server/push/notifications.ts @@ -0,0 +1,29 @@ +import type pino from "pino"; + +import { PushService, type PushPayload } from "./push-service.js"; +import type { PushTokenStore } from "./token-store.js"; + +export type { PushPayload }; + +export interface PushNotificationSender { + send(payload: PushPayload): Promise; +} + +export function createPushNotificationSender( + logger: pino.Logger, + tokenStore: PushTokenStore, +): PushNotificationSender { + const pushService = new PushService(logger, tokenStore); + + return { + async send(payload) { + const tokens = tokenStore.getAllTokens(); + logger.info({ tokenCount: tokens.length }, "Sending push notification"); + if (tokens.length === 0) { + return; + } + + await pushService.sendPush(tokens, payload); + }, + }; +} diff --git a/packages/server/src/server/push/push-service.ts b/packages/server/src/server/push/push-service.ts index 8a792862a..dcfadc122 100644 --- a/packages/server/src/server/push/push-service.ts +++ b/packages/server/src/server/push/push-service.ts @@ -1,7 +1,7 @@ import type { PushTokenStore } from "./token-store.js"; import type pino from "pino"; -interface PushPayload { +export interface PushPayload { title: string; body: string; data?: Record; diff --git a/packages/server/src/server/test-utils/paseo-daemon.ts b/packages/server/src/server/test-utils/paseo-daemon.ts index 59a78ce08..58d347b65 100644 --- a/packages/server/src/server/test-utils/paseo-daemon.ts +++ b/packages/server/src/server/test-utils/paseo-daemon.ts @@ -11,6 +11,7 @@ import { } from "../bootstrap.js"; import type { AgentClient, AgentProvider } from "../agent/agent-sdk-types.js"; import { createTestAgentClients } from "./fake-agent-client.js"; +import type { PushNotificationSender } from "../push/notifications.js"; interface TestPaseoDaemonOptions { downloadTokenTtlMs?: number; @@ -30,6 +31,7 @@ interface TestPaseoDaemonOptions { voiceLlmModel?: string | null; dictationFinalTimeoutMs?: number; auth?: PaseoDaemonConfig["auth"]; + pushNotificationSender?: PushNotificationSender; } export interface TestPaseoDaemon { @@ -157,6 +159,7 @@ async function prepareTestDaemonConfig( relayEndpoint: options.relayEndpoint ?? "relay.paseo.sh:443", appBaseUrl: "https://app.paseo.sh", auth: options.auth, + pushNotificationSender: options.pushNotificationSender, openai: options.openai, speech: options.speech, voiceLlmProvider: options.voiceLlmProvider ?? null, diff --git a/packages/server/src/server/websocket-server.notifications.test.ts b/packages/server/src/server/websocket-server.notifications.test.ts index 9179cf889..0cbb1748e 100644 --- a/packages/server/src/server/websocket-server.notifications.test.ts +++ b/packages/server/src/server/websocket-server.notifications.test.ts @@ -10,6 +10,7 @@ import type { LoopService } from "./loop-service.js"; import type { ScheduleService } from "./schedule/service.js"; import type { CheckoutDiffManager } from "./checkout-diff-manager.js"; import { asInternals, createStub } from "./test-utils/class-mocks.js"; +import type { PushNotificationSender, PushPayload } from "./push/notifications.js"; const wsModuleMock = vi.hoisted(() => { class MockWebSocketServer { @@ -28,11 +29,6 @@ const wsModuleMock = vi.hoisted(() => { return { MockWebSocketServer }; }); -const pushMocks = vi.hoisted(() => ({ - getAllTokens: vi.fn(() => ["ExponentPushToken[token-1]"]), - sendPush: vi.fn(async () => {}), -})); - vi.mock("ws", () => ({ WebSocketServer: wsModuleMock.MockWebSocketServer, })); @@ -43,19 +39,6 @@ vi.mock("./session.js", () => ({ }, })); -vi.mock("./push/token-store.js", () => ({ - PushTokenStore: class { - getAllTokens = pushMocks.getAllTokens; - removeToken = vi.fn(); - }, -})); - -vi.mock("./push/push-service.js", () => ({ - PushService: class { - sendPush = pushMocks.sendPush; - }, -})); - import { VoiceAssistantWebSocketServer } from "./websocket-server.js"; interface WebSocketServerInternals { @@ -81,7 +64,16 @@ function createLogger() { return logger; } +class RecordingPushNotificationSender implements PushNotificationSender { + readonly sent: PushPayload[] = []; + + async send(payload: PushPayload): Promise { + this.sent.push(payload); + } +} + function createServer(agentManagerOverrides?: Record) { + const pushNotifications = new RecordingPushNotificationSender(); const agentManager = { setAgentAttentionCallback: vi.fn(), getAgent: vi.fn(() => null), @@ -137,9 +129,18 @@ function createServer(agentManagerOverrides?: Record) { })), dispose: vi.fn(), }), + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + pushNotifications, ); - return { server, agentManager }; + return { server, agentManager, pushNotifications }; } function createOpenSocket() { @@ -208,7 +209,7 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => { const getLastAssistantMessage = vi.fn( async () => "**Done**. Updated `README.md` and [link](https://example.com).", ); - const { server } = createServer({ + const { server, pushNotifications } = createServer({ getAgent: vi.fn(() => ({ config: { title: null }, cwd: "/tmp/worktree", @@ -223,21 +224,23 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => { reason: "finished", }); - expect(pushMocks.sendPush).toHaveBeenCalledWith(["ExponentPushToken[token-1]"], { - title: "Agent finished", - body: "Done. Updated README.md and link.", - data: { - serverId: "srv-test", - agentId: "agent-1", - reason: "finished", + expect(pushNotifications.sent).toEqual([ + { + title: "Agent finished", + body: "Done. Updated README.md and link.", + data: { + serverId: "srv-test", + agentId: "agent-1", + reason: "finished", + }, }, - }); + ]); expect(getLastAssistantMessage).toHaveBeenCalledWith("agent-1"); }); it("sends push notifications regardless of UI label presence", async () => { const getLastAssistantMessage = vi.fn(async () => "Done."); - const { server } = createServer({ + const { server, pushNotifications } = createServer({ getAgent: vi.fn(() => ({ config: { title: null }, cwd: "/tmp/worktree", @@ -253,12 +256,12 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => { reason: "finished", }); - expect(pushMocks.sendPush).toHaveBeenCalledTimes(1); + expect(pushNotifications.sent).toHaveLength(1); expect(getLastAssistantMessage).toHaveBeenCalledWith("agent-2"); }); it("routes a hidden stale focused browser tab's notification to the present Electron web client", async () => { - const { server } = createServer(); + const { server, pushNotifications } = createServer(); const nowMs = Date.now(); const electronWs = connectClient(server, { deviceType: "web", @@ -281,11 +284,11 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => { expect(readAttentionRequiredMessage(electronWs).shouldNotify).toBe(true); expect(readAttentionRequiredMessage(firefoxWs).shouldNotify).toBe(false); - expect(pushMocks.sendPush).not.toHaveBeenCalled(); + expect(pushNotifications.sent).toEqual([]); }); it("pushes non-error attention when the only connected client has never sent a heartbeat", async () => { - const { server } = createServer(); + const { server, pushNotifications } = createServer(); const ws = connectClient(server, null); await asInternals(server).broadcastAgentAttention({ @@ -295,11 +298,11 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => { }); expect(readAttentionRequiredMessage(ws).shouldNotify).toBe(false); - expect(pushMocks.sendPush).toHaveBeenCalledTimes(1); + expect(pushNotifications.sent).toHaveLength(1); }); it("does not push error attention when the only connected client has never sent a heartbeat", async () => { - const { server } = createServer(); + const { server, pushNotifications } = createServer(); const ws = connectClient(server, null); await asInternals(server).broadcastAgentAttention({ @@ -309,6 +312,6 @@ describe("VoiceAssistantWebSocketServer notification payloads", () => { }); expect(readAttentionRequiredMessage(ws).shouldNotify).toBe(false); - expect(pushMocks.sendPush).not.toHaveBeenCalled(); + expect(pushNotifications.sent).toEqual([]); }); }); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index d4ef8bee1..17fe30806 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -40,7 +40,7 @@ import { buildProviderRegistry, createClientsFromRegistry } from "./agent/provid import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js"; import { buildWorkspaceGitMetadataFromSnapshot } from "./workspace-git-metadata.js"; import { PushTokenStore } from "./push/token-store.js"; -import { PushService } from "./push/push-service.js"; +import { createPushNotificationSender, type PushNotificationSender } from "./push/notifications.js"; import type { ScriptHealthState } from "./script-health-monitor.js"; import type { ScriptRouteStore } from "./script-proxy.js"; import type { WorkspaceScriptRuntimeStore } from "./workspace-script-runtime-store.js"; @@ -358,7 +358,7 @@ export class VoiceAssistantWebSocketServer { private readonly paseoHome: string; private readonly daemonConfigStore: DaemonConfigStore; private readonly pushTokenStore: PushTokenStore; - private readonly pushService: PushService; + private readonly pushNotificationSender: PushNotificationSender; private readonly mcpBaseUrl: string | null; private speech!: SpeechService | null; private terminalManager!: TerminalManager | null; @@ -453,6 +453,7 @@ export class VoiceAssistantWebSocketServer { resolveScriptHealth?: (hostname: string) => ScriptHealthState | null, workspaceGitService?: WorkspaceGitService, github?: GitHubService, + pushNotificationSender?: PushNotificationSender, ) { this.logger = logger.child({ module: "websocket-server" }); this.serverId = serverId; @@ -529,7 +530,8 @@ export class VoiceAssistantWebSocketServer { const pushLogger = this.logger.child({ module: "push" }); this.pushTokenStore = new PushTokenStore(pushLogger, join(paseoHome, "push-tokens.json")); - this.pushService = new PushService(pushLogger, this.pushTokenStore); + this.pushNotificationSender = + pushNotificationSender ?? createPushNotificationSender(pushLogger, this.pushTokenStore); this.agentManager.setAgentAttentionCallback((params) => { void this.broadcastAgentAttention(params).catch((err) => { @@ -1819,11 +1821,9 @@ export class VoiceAssistantWebSocketServer { }); if (plan.shouldPush) { - const tokens = this.pushTokenStore.getAllTokens(); - this.logger.info({ tokenCount: tokens.length }, "Sending push notification"); - if (tokens.length > 0) { - void this.pushService.sendPush(tokens, notification); - } + void this.pushNotificationSender.send(notification).catch((err) => { + this.logger.warn({ err, agentId: params.agentId }, "Failed to send push notification"); + }); } for (const [clientIndex, { ws }] of clientEntries.entries()) {