From e782636c4b24d1761b811861935691004ed50270 Mon Sep 17 00:00:00 2001 From: Fusion Date: Tue, 28 Apr 2026 09:22:46 -0700 Subject: [PATCH] feat(FN-2866): wire provider-backed notification service into engine - Add notification service module with provider abstractions and ntfy provider implementation - Refactor NtfyNotifier into a compatibility wrapper that delegates task-event delivery to NotificationService - Initialize and stop NotificationService from ProjectEngine while preserving gridlock notifications via NtfyNotifier - Export notification APIs from engine index and add focused unit coverage for provider, service, and project-engine wiring --- .../__tests__/notification-service.test.ts | 131 +++++++++++ .../src/__tests__/ntfy-provider.test.ts | 94 ++++++++ .../src/__tests__/project-engine.test.ts | 8 + packages/engine/src/index.ts | 4 + packages/engine/src/notification/index.ts | 5 + .../src/notification/notification-service.ts | 206 ++++++++++++++++++ .../engine/src/notification/ntfy-provider.ts | 154 +++++++++++++ packages/engine/src/notifier.ts | 171 ++------------- packages/engine/src/project-engine.ts | 14 +- 9 files changed, 629 insertions(+), 158 deletions(-) create mode 100644 packages/engine/src/__tests__/notification-service.test.ts create mode 100644 packages/engine/src/__tests__/ntfy-provider.test.ts create mode 100644 packages/engine/src/notification/index.ts create mode 100644 packages/engine/src/notification/notification-service.ts create mode 100644 packages/engine/src/notification/ntfy-provider.ts diff --git a/packages/engine/src/__tests__/notification-service.test.ts b/packages/engine/src/__tests__/notification-service.test.ts new file mode 100644 index 000000000..c03111a6a --- /dev/null +++ b/packages/engine/src/__tests__/notification-service.test.ts @@ -0,0 +1,131 @@ +import { describe, expect, it, vi } from "vitest"; +import type { NotificationProvider, Settings, Task } from "@fusion/core"; +import { NotificationService } from "../notification/notification-service.js"; +import { NtfyNotificationProvider } from "../notification/ntfy-provider.js"; + +type Listener = (...args: any[]) => void | Promise; + +function createStore(settings: Partial = {}) { + const listeners = new Map>(); + const getBucket = (event: string) => listeners.get(event) ?? new Set(); + + return { + getSettings: vi.fn(async () => ({ ntfyEnabled: false, ...settings }) as Settings), + on: vi.fn((event: string, listener: Listener) => { + const bucket = getBucket(event); + bucket.add(listener); + listeners.set(event, bucket); + }), + off: vi.fn((event: string, listener: Listener) => { + getBucket(event).delete(listener); + }), + emit(event: string, payload: unknown) { + for (const listener of getBucket(event)) { + void listener(payload); + } + }, + }; +} + +function task(overrides: Partial = {}): Task { + return { + id: "FN-1", + title: "Task title", + description: "Task desc", + status: "todo", + column: "todo", + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + dependencies: [], + steps: [], + currentStep: 0, + log: [], + ...overrides, + } as Task; +} + +describe("NotificationService", () => { + it("dispatches in-review event to registered provider", async () => { + const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" }); + const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" })); + const provider: NotificationProvider = { + getProviderId: () => "mock", + isEventSupported: () => true, + sendNotification, + }; + + const service = new NotificationService(store as any); + service.registerProvider(provider); + await service.start(); + + store.emit("task:moved", { task: task(), from: "todo", to: "in-review" }); + await Promise.resolve(); + + expect(sendNotification).toHaveBeenCalledWith( + "in-review", + expect.objectContaining({ taskId: "FN-1", event: "in-review" }), + ); + }); + + it("deduplicates same task+event but not different event types", async () => { + const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" }); + const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" })); + const provider: NotificationProvider = { + getProviderId: () => "mock", + isEventSupported: () => true, + sendNotification, + }; + + const service = new NotificationService(store as any); + service.registerProvider(provider); + await service.start(); + + store.emit("task:moved", { task: task(), from: "todo", to: "in-review" }); + store.emit("task:moved", { task: task(), from: "todo", to: "in-review" }); + store.emit("task:updated", task({ status: "failed" })); + await Promise.resolve(); + + expect(sendNotification).toHaveBeenCalledTimes(2); + }); + + it("stop unsubscribes listeners", async () => { + const store = createStore({ ntfyEnabled: true, ntfyTopic: "topic" }); + const sendNotification = vi.fn(async () => ({ success: true, providerId: "mock" })); + const provider: NotificationProvider = { + getProviderId: () => "mock", + isEventSupported: () => true, + sendNotification, + }; + + const service = new NotificationService(store as any); + service.registerProvider(provider); + await service.start(); + await service.stop(); + + store.emit("task:moved", { task: task(), from: "todo", to: "in-review" }); + await Promise.resolve(); + expect(sendNotification).not.toHaveBeenCalled(); + }); + + it("auto-registers ntfy provider when enabled and topic set", async () => { + const store = createStore({ ntfyEnabled: true, ntfyTopic: "demo", ntfyDashboardHost: "http://x" }); + const initSpy = vi.spyOn(NtfyNotificationProvider.prototype, "initialize"); + + const service = new NotificationService(store as any, { projectId: "p1", ntfyBaseUrl: "https://n" }); + await service.start(); + + expect(initSpy).toHaveBeenCalledWith( + expect.objectContaining({ topic: "demo", projectId: "p1", ntfyBaseUrl: "https://n" }), + ); + initSpy.mockRestore(); + }); + + it("skips ntfy provider when disabled", async () => { + const store = createStore({ ntfyEnabled: false, ntfyTopic: "demo" }); + const initSpy = vi.spyOn(NtfyNotificationProvider.prototype, "initialize"); + const service = new NotificationService(store as any); + await service.start(); + expect(initSpy).not.toHaveBeenCalled(); + initSpy.mockRestore(); + }); +}); diff --git a/packages/engine/src/__tests__/ntfy-provider.test.ts b/packages/engine/src/__tests__/ntfy-provider.test.ts new file mode 100644 index 000000000..d527d8c13 --- /dev/null +++ b/packages/engine/src/__tests__/ntfy-provider.test.ts @@ -0,0 +1,94 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + sendNtfyNotification: vi.fn(async () => undefined), + buildNtfyClickUrl: vi.fn(() => "http://dash/?project=p1&task=FN-1"), +})); + +vi.mock("../notifier.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + sendNtfyNotification: mocks.sendNtfyNotification, + buildNtfyClickUrl: mocks.buildNtfyClickUrl, + }; +}); + +import { NtfyNotificationProvider } from "../notification/ntfy-provider.js"; + +describe("NtfyNotificationProvider", () => { + let provider: NtfyNotificationProvider; + + beforeEach(async () => { + mocks.sendNtfyNotification.mockClear(); + mocks.buildNtfyClickUrl.mockClear(); + provider = new NtfyNotificationProvider(); + await provider.initialize({ + topic: "topic-a", + ntfyBaseUrl: "https://ntfy.local", + dashboardHost: "http://dash", + projectId: "p1", + }); + }); + + it("returns provider id", () => { + expect(provider.getProviderId()).toBe("ntfy"); + }); + + it.each([ + ["in-review", "Task FN-1 completed", "is ready for review", "default"], + ["merged", "Task FN-1 merged", "has been merged to main", "default"], + ["failed", "Task FN-1 failed", "has failed and needs attention", "high"], + ["awaiting-approval", "Plan needs approval for FN-1", "needs your approval", "high"], + ["awaiting-user-review", "User review needed for FN-1", "needs human review", "high"], + ["planning-awaiting-input", "Planning input needed for FN-1", "awaiting your input", "high"], + ])("maps %s event correctly", async (event, expectedTitle, messagePart, priority) => { + await provider.sendNotification(event as any, { taskId: "FN-1", taskTitle: "T", event: event as any }); + + expect(mocks.sendNtfyNotification).toHaveBeenCalledWith( + expect.objectContaining({ + topic: "topic-a", + title: expectedTitle, + priority, + message: expect.stringContaining(messagePart), + }), + ); + }); + + it("supports known events and rejects unknown", () => { + expect(provider.isEventSupported("in-review" as any)).toBe(true); + expect(provider.isEventSupported("merged" as any)).toBe(true); + expect(provider.isEventSupported("failed" as any)).toBe(true); + expect(provider.isEventSupported("awaiting-approval" as any)).toBe(true); + expect(provider.isEventSupported("awaiting-user-review" as any)).toBe(true); + expect(provider.isEventSupported("planning-awaiting-input" as any)).toBe(true); + expect(provider.isEventSupported("custom-event" as any)).toBe(false); + }); + + it("shutdown aborts internal AbortController", async () => { + await provider.shutdown(); + await provider.sendNotification("in-review" as any, { taskId: "FN-1", taskTitle: "T", event: "in-review" as any }); + expect(mocks.sendNtfyNotification).toHaveBeenCalledWith(expect.objectContaining({ signal: undefined })); + }); + + it("uses fallback identifier from id+description when no title", async () => { + await provider.sendNotification("failed" as any, { + taskId: "FN-1", + taskDescription: "desc", + event: "failed" as any, + }); + + expect(mocks.sendNtfyNotification).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringContaining('Task "FN-1: desc"') }), + ); + }); + + it("builds click URL from config", async () => { + await provider.sendNotification("merged" as any, { taskId: "FN-1", taskTitle: "T", event: "merged" as any }); + expect(mocks.buildNtfyClickUrl).toHaveBeenCalledWith({ + dashboardHost: "http://dash", + projectId: "p1", + taskId: "FN-1", + }); + }); +}); diff --git a/packages/engine/src/__tests__/project-engine.test.ts b/packages/engine/src/__tests__/project-engine.test.ts index 3f366b19e..6a38ca07d 100644 --- a/packages/engine/src/__tests__/project-engine.test.ts +++ b/packages/engine/src/__tests__/project-engine.test.ts @@ -72,6 +72,14 @@ vi.mock("../notifier.js", () => ({ NtfyNotifier: vi.fn().mockImplementation(() => ({ start: vi.fn(async () => undefined), stop: vi.fn(), + notifyGridlock: vi.fn(), + })), +})); + +vi.mock("../notification/index.js", () => ({ + NotificationService: vi.fn().mockImplementation(() => ({ + start: vi.fn().mockResolvedValue(undefined), + stop: vi.fn(), })), })); diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index c6a24c499..1cd045b15 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -59,11 +59,15 @@ export { isNtfyEventEnabled, buildNtfyClickUrl, sendNtfyNotification, + formatTaskIdentifier, type NtfyNotifierOptions, type NtfyNotificationPriority, type NtfyNotificationConfigInput, type SendNtfyNotificationInput, } from "./notifier.js"; +// ── Notification Service ────────────────────────────────────── +export { NtfyNotificationProvider, NotificationService } from "./notification/index.js"; +export type { NtfyProviderConfig, NotificationServiceOptions } from "./notification/index.js"; export { CronRunner, type CronRunnerOptions, type AiPromptExecutor, createAiPromptExecutor } from "./cron-runner.js"; export { RoutineRunner, type RoutineRunnerOptions } from "./routine-runner.js"; export { RoutineScheduler, type RoutineSchedulerOptions } from "./routine-scheduler.js"; diff --git a/packages/engine/src/notification/index.ts b/packages/engine/src/notification/index.ts new file mode 100644 index 000000000..19079a7e6 --- /dev/null +++ b/packages/engine/src/notification/index.ts @@ -0,0 +1,5 @@ +export { NtfyNotificationProvider } from "./ntfy-provider.js"; +export type { NtfyProviderConfig } from "./ntfy-provider.js"; + +export { NotificationService } from "./notification-service.js"; +export type { NotificationServiceOptions } from "./notification-service.js"; diff --git a/packages/engine/src/notification/notification-service.ts b/packages/engine/src/notification/notification-service.ts new file mode 100644 index 000000000..8c0573aaf --- /dev/null +++ b/packages/engine/src/notification/notification-service.ts @@ -0,0 +1,206 @@ +import type { + Column, + MergeResult, + NotificationEvent, + NotificationPayload, + NotificationProvider, + Settings, + Task, +} from "@fusion/core"; +import { NotificationDispatcher } from "@fusion/core"; +import { DEFAULT_NTFY_EVENTS } from "../notifier.js"; +import { schedulerLog } from "../logger.js"; +import { NtfyNotificationProvider } from "./ntfy-provider.js"; + +export interface NotificationServiceOptions { + /** Project identifier for notification deep links */ + projectId?: string; + /** Base URL for ntfy.sh (backward compat with NtfyNotifierOptions) */ + ntfyBaseUrl?: string; +} + +interface NotificationServiceStore { + getSettings(): Promise | Settings; + on(event: string, listener: (...args: any[]) => void): void; + off(event: string, listener: (...args: any[]) => void): void; +} + +export class NotificationService { + private readonly dispatcher = new NotificationDispatcher(); + private readonly notifiedEvents = new Set(); + private started = false; + private ntfyEnabled = false; + private ntfyProvider?: NtfyNotificationProvider; + + constructor( + private readonly store: NotificationServiceStore, + private readonly options: NotificationServiceOptions = {}, + ) {} + + registerProvider(provider: NotificationProvider): void { + this.dispatcher.registerProvider(provider); + } + + async start(): Promise { + if (this.started) { + return; + } + + const settings = await this.store.getSettings(); + await this.syncNtfyProvider(settings); + + await this.dispatcher.initializeAll(); + + this.store.on("task:moved", this.handleTaskMoved); + this.store.on("task:updated", this.handleTaskUpdated); + this.store.on("task:merged", this.handleTaskMerged); + this.store.on("settings:updated", this.handleSettingsUpdated); + + this.started = true; + schedulerLog.log("NotificationService started"); + } + + async stop(): Promise { + if (!this.started) { + return; + } + + if (typeof this.store.off === "function") { + this.store.off("task:moved", this.handleTaskMoved); + this.store.off("task:updated", this.handleTaskUpdated); + this.store.off("task:merged", this.handleTaskMerged); + this.store.off("settings:updated", this.handleSettingsUpdated); + } + + await this.dispatcher.shutdownAll(); + this.started = false; + + schedulerLog.log("NotificationService stopped"); + } + + private handleTaskMoved = (data: { task: Task; from: Column; to: Column }): void => { + if (!this.ntfyEnabled || data.to !== "in-review") { + return; + } + + const payload = this.createTaskPayload(data.task, "in-review"); + this.maybeNotify(data.task.id, "in-review", payload); + }; + + private handleTaskUpdated = (task: Task): void => { + if (!this.ntfyEnabled) { + return; + } + + if (task.status === "failed") { + this.maybeNotify(task.id, "failed", this.createTaskPayload(task, "failed")); + } + + if (task.status === "awaiting-approval") { + this.maybeNotify( + task.id, + "awaiting-approval", + this.createTaskPayload(task, "awaiting-approval"), + ); + } + + if (task.status === "awaiting-user-review") { + this.maybeNotify( + task.id, + "awaiting-user-review", + this.createTaskPayload(task, "awaiting-user-review"), + ); + } + }; + + private handleTaskMerged = (result: MergeResult): void => { + if (!this.ntfyEnabled || !result.merged) { + return; + } + + this.maybeNotify( + result.task.id, + "merged", + this.createTaskPayload(result.task, "merged"), + ); + }; + + private handleSettingsUpdated = async (data: { settings: Settings; previous: Settings }): Promise => { + const { settings, previous } = data; + + if ( + settings.ntfyEnabled !== previous.ntfyEnabled || + settings.ntfyTopic !== previous.ntfyTopic || + settings.ntfyBaseUrl !== previous.ntfyBaseUrl || + settings.ntfyDashboardHost !== previous.ntfyDashboardHost || + JSON.stringify(settings.ntfyEvents) !== JSON.stringify(previous.ntfyEvents) + ) { + const wasEnabled = Boolean(previous.ntfyEnabled && previous.ntfyTopic); + const isEnabled = Boolean(settings.ntfyEnabled && settings.ntfyTopic); + + await this.syncNtfyProvider(settings); + + if (isEnabled && !wasEnabled) { + schedulerLog.log("NotificationService ntfy enabled"); + } else if (!isEnabled && wasEnabled) { + schedulerLog.log("NotificationService ntfy disabled"); + } else if (settings.ntfyTopic !== previous.ntfyTopic) { + schedulerLog.log("NotificationService ntfy topic updated"); + } else if (settings.ntfyBaseUrl !== previous.ntfyBaseUrl) { + schedulerLog.log("NotificationService ntfy base URL updated"); + } else if (settings.ntfyDashboardHost !== previous.ntfyDashboardHost) { + schedulerLog.log("NotificationService ntfy dashboard host updated"); + } else if (JSON.stringify(settings.ntfyEvents) !== JSON.stringify(previous.ntfyEvents)) { + schedulerLog.log("NotificationService ntfy events updated"); + } + } + }; + + private async syncNtfyProvider(settings: Settings): Promise { + const enabled = Boolean(settings.ntfyEnabled && settings.ntfyTopic); + this.ntfyEnabled = enabled; + + if (!enabled) { + if (this.ntfyProvider) { + await this.ntfyProvider.shutdown?.(); + this.dispatcher.unregisterProvider(this.ntfyProvider.getProviderId()); + this.ntfyProvider = undefined; + } + return; + } + + if (!this.ntfyProvider) { + this.ntfyProvider = new NtfyNotificationProvider(); + this.registerProvider(this.ntfyProvider); + } + + await this.ntfyProvider.initialize?.({ + topic: settings.ntfyTopic, + ntfyBaseUrl: settings.ntfyBaseUrl ?? this.options.ntfyBaseUrl, + dashboardHost: settings.ntfyDashboardHost, + events: settings.ntfyEvents ?? [...DEFAULT_NTFY_EVENTS], + projectId: this.options.projectId, + }); + } + + private createTaskPayload(task: Task, event: NotificationEvent): NotificationPayload { + return { + taskId: task.id, + taskTitle: task.title, + taskDescription: task.description, + event, + }; + } + + private maybeNotify(taskId: string, eventType: NotificationEvent, payload: NotificationPayload): void { + const key = `${taskId}:${eventType}`; + if (this.notifiedEvents.has(key)) { + return; + } + + this.notifiedEvents.add(key); + this.dispatcher.dispatch(eventType, payload).catch(() => { + // best effort dispatch + }); + } +} diff --git a/packages/engine/src/notification/ntfy-provider.ts b/packages/engine/src/notification/ntfy-provider.ts new file mode 100644 index 000000000..a0dbb0d1d --- /dev/null +++ b/packages/engine/src/notification/ntfy-provider.ts @@ -0,0 +1,154 @@ +import type { + NotificationEvent, + NotificationPayload, + NotificationProvider, + NotificationResult, + NtfyNotificationEvent, + Task, +} from "@fusion/core"; +import { + DEFAULT_NTFY_EVENTS, + buildNtfyClickUrl, + formatTaskIdentifier, + resolveNtfyEvents, + sendNtfyNotification, +} from "../notifier.js"; + +export interface NtfyProviderConfig { + /** ntfy topic name */ + topic: string; + /** ntfy server base URL (default: https://ntfy.sh) */ + ntfyBaseUrl?: string; + /** Dashboard host for click-through deep links */ + dashboardHost?: string; + /** Project identifier for deep links */ + projectId?: string; + /** Events to enable (default: DEFAULT_NTFY_EVENTS) */ + events?: NtfyNotificationEvent[]; +} + +type SupportedNtfyEvent = + | "in-review" + | "merged" + | "failed" + | "awaiting-approval" + | "awaiting-user-review" + | "planning-awaiting-input"; + +const SUPPORTED_EVENTS = new Set([ + "in-review", + "merged", + "failed", + "awaiting-approval", + "awaiting-user-review", + "planning-awaiting-input", +]); + +export class NtfyNotificationProvider implements NotificationProvider { + private config?: NtfyProviderConfig; + private abortController: AbortController | null = null; + + getProviderId(): string { + return "ntfy"; + } + + async initialize(config: Record): Promise { + if (typeof config.topic !== "string" || config.topic.trim() === "") { + return; + } + + this.config = config as unknown as NtfyProviderConfig; + this.config.events = resolveNtfyEvents(this.config.events); + this.abortController = new AbortController(); + } + + async shutdown(): Promise { + this.abortController?.abort(); + this.abortController = null; + } + + isEventSupported(event: NotificationEvent): boolean { + if (!SUPPORTED_EVENTS.has(event as SupportedNtfyEvent)) { + return false; + } + + const enabledEvents = this.config?.events ?? [...DEFAULT_NTFY_EVENTS]; + return enabledEvents.includes(event as NtfyNotificationEvent); + } + + async sendNotification( + event: NotificationEvent, + payload: NotificationPayload, + ): Promise { + if (!this.config?.topic) { + return { success: false, providerId: this.getProviderId(), error: "ntfy topic not configured" }; + } + + if (!this.isEventSupported(event)) { + return { + success: false, + providerId: this.getProviderId(), + error: `unsupported event: ${event}`, + }; + } + + const taskLike = { + id: payload.taskId, + title: payload.taskTitle, + description: payload.taskDescription ?? "", + } as Pick as Task; + + const identifier = formatTaskIdentifier(taskLike); + const clickUrl = buildNtfyClickUrl({ + dashboardHost: this.config.dashboardHost, + projectId: this.config.projectId, + taskId: payload.taskId, + }); + + const contentByEvent: Record = { + "in-review": { + title: `Task ${payload.taskId} completed`, + message: `Task "${identifier}" is ready for review`, + priority: "default", + }, + merged: { + title: `Task ${payload.taskId} merged`, + message: `Task "${identifier}" has been merged to main`, + priority: "default", + }, + failed: { + title: `Task ${payload.taskId} failed`, + message: `Task "${identifier}" has failed and needs attention`, + priority: "high", + }, + "awaiting-approval": { + title: `Plan needs approval for ${payload.taskId}`, + message: `Task "${identifier}" needs your approval before it can proceed`, + priority: "high", + }, + "awaiting-user-review": { + title: `User review needed for ${payload.taskId}`, + message: `Task "${identifier}" needs human review before it can proceed`, + priority: "high", + }, + "planning-awaiting-input": { + title: `Planning input needed for ${payload.taskId}`, + message: `Task "${identifier}" is awaiting your input during planning`, + priority: "high", + }, + }; + + const content = contentByEvent[event as SupportedNtfyEvent]; + await sendNtfyNotification({ + ntfyBaseUrl: this.config.ntfyBaseUrl, + topic: this.config.topic, + title: content.title, + message: content.message, + priority: content.priority, + clickUrl, + signal: this.abortController?.signal, + }); + + return { success: true, providerId: this.getProviderId() }; + } +} diff --git a/packages/engine/src/notifier.ts b/packages/engine/src/notifier.ts index df340e541..5a5188922 100644 --- a/packages/engine/src/notifier.ts +++ b/packages/engine/src/notifier.ts @@ -1,6 +1,7 @@ -import type { Task, Column, Settings, MergeResult, NtfyNotificationEvent } from "@fusion/core"; +import type { Task, Settings, NtfyNotificationEvent } from "@fusion/core"; import type { GridlockEvent } from "./gridlock-detector.js"; import { schedulerLog } from "./logger.js"; +import { NotificationService } from "./notification/index.js"; export interface NtfyNotifierOptions { /** Base URL for ntfy.sh. Default: https://ntfy.sh */ @@ -58,7 +59,7 @@ type AnyNotificationEvent = TaskNotificationEvent | "gridlock"; * - If title exists: returns "{title}" * - If no title: returns "{id}: {first 200 chars of description}" (truncated with "..." if > 200) */ -function formatTaskIdentifier(task: Task): string { +export function formatTaskIdentifier(task: Task): string { if (task.title) { return task.title; } @@ -160,9 +161,9 @@ export async function sendNtfyNotification({ } /** - * NtfyNotifier sends push notifications via ntfy.sh when tasks complete - * or fail. It listens to TaskStore events and sends HTTP POST requests - * to the configured ntfy topic. + * NtfyNotifier is a backward-compatible wrapper around NotificationService. + * It keeps legacy APIs (getConfig, notifyGridlock) while delegating task event + * notifications to the pluggable provider-based notification module. */ export class NtfyNotifier { private config: NtfyConfig = { @@ -171,9 +172,10 @@ export class NtfyNotifier { dashboardHost: undefined, events: [...DEFAULT_NTFY_EVENTS], }; + private readonly notificationService: NotificationService; private ntfyBaseUrl: string; private readonly defaultNtfyBaseUrl: string; - private projectId?: string; + private readonly projectId?: string; private notifiedEvents: Set = new Set(); private abortController: AbortController | null = null; @@ -184,27 +186,23 @@ export class NtfyNotifier { this.defaultNtfyBaseUrl = resolveNtfyBaseUrl(options.ntfyBaseUrl); this.ntfyBaseUrl = this.defaultNtfyBaseUrl; this.projectId = options.projectId; + this.notificationService = new NotificationService(store, { + projectId: this.projectId, + ntfyBaseUrl: options.ntfyBaseUrl, + }); } async start(): Promise { this.abortController = new AbortController(); - const settings = await this.store.getSettings(); this.loadConfig(settings); - - this.store.on("task:moved", this.handleTaskMoved); - this.store.on("task:updated", this.handleTaskUpdated); - this.store.on("task:merged", this.handleTaskMerged); this.store.on("settings:updated", this.handleSettingsUpdated); - + await this.notificationService.start(); schedulerLog.log("NtfyNotifier started"); } stop(): void { if (typeof this.store.off === "function") { - this.store.off("task:moved", this.handleTaskMoved); - this.store.off("task:updated", this.handleTaskUpdated); - this.store.off("task:merged", this.handleTaskMerged); this.store.off("settings:updated", this.handleSettingsUpdated); } @@ -213,143 +211,12 @@ export class NtfyNotifier { this.abortController = null; } + void this.notificationService.stop(); schedulerLog.log("NtfyNotifier stopped"); } - private handleTaskMoved = (data: { task: Task; from: Column; to: Column }): void => { - if (!this.config.enabled || !this.config.topic) return; - - const { task, to } = data; - - if (to === "in-review" && this.isEventEnabled("in-review")) { - const clickUrl = buildNtfyClickUrl({ - dashboardHost: this.config.dashboardHost, - projectId: this.projectId, - taskId: task.id, - }); - this.maybeNotify(task.id, "in-review", () => - sendNtfyNotification({ - ntfyBaseUrl: this.ntfyBaseUrl, - topic: this.config.topic!, - title: `Task ${task.id} completed`, - message: `Task "${formatTaskIdentifier(task)}" is ready for review`, - priority: "default", - clickUrl, - signal: this.abortController?.signal, - }), - ); - } - }; - - private handleTaskUpdated = (task: Task): void => { - if (!this.config.enabled || !this.config.topic) return; - - if (task.status === "failed" && this.isEventEnabled("failed")) { - const clickUrl = buildNtfyClickUrl({ - dashboardHost: this.config.dashboardHost, - projectId: this.projectId, - taskId: task.id, - }); - this.maybeNotify(task.id, "failed", () => - sendNtfyNotification({ - ntfyBaseUrl: this.ntfyBaseUrl, - topic: this.config.topic!, - title: `Task ${task.id} failed`, - message: `Task "${formatTaskIdentifier(task)}" has failed and needs attention`, - priority: "high", - clickUrl, - signal: this.abortController?.signal, - }), - ); - } - - if (task.status === "awaiting-approval" && this.isEventEnabled("awaiting-approval")) { - const clickUrl = buildNtfyClickUrl({ - dashboardHost: this.config.dashboardHost, - projectId: this.projectId, - taskId: task.id, - }); - this.maybeNotify(task.id, "awaiting-approval", () => - sendNtfyNotification({ - ntfyBaseUrl: this.ntfyBaseUrl, - topic: this.config.topic!, - title: `Plan needs approval for ${task.id}`, - message: `Task "${formatTaskIdentifier(task)}" needs your approval before it can proceed`, - priority: "high", - clickUrl, - signal: this.abortController?.signal, - }), - ); - } - - if (task.status === "awaiting-user-review" && this.isEventEnabled("awaiting-user-review")) { - const clickUrl = buildNtfyClickUrl({ - dashboardHost: this.config.dashboardHost, - projectId: this.projectId, - taskId: task.id, - }); - this.maybeNotify(task.id, "awaiting-user-review", () => - sendNtfyNotification({ - ntfyBaseUrl: this.ntfyBaseUrl, - topic: this.config.topic!, - title: `User review needed for ${task.id}`, - message: `Task "${formatTaskIdentifier(task)}" needs human review before it can proceed`, - priority: "high", - clickUrl, - signal: this.abortController?.signal, - }), - ); - } - }; - - private handleTaskMerged = (result: MergeResult): void => { - if (!this.config.enabled || !this.config.topic) return; - - if (result.merged && this.isEventEnabled("merged")) { - const clickUrl = buildNtfyClickUrl({ - dashboardHost: this.config.dashboardHost, - projectId: this.projectId, - taskId: result.task.id, - }); - this.maybeNotify(result.task.id, "merged", () => - sendNtfyNotification({ - ntfyBaseUrl: this.ntfyBaseUrl, - topic: this.config.topic!, - title: `Task ${result.task.id} merged`, - message: `Task "${formatTaskIdentifier(result.task)}" has been merged to main`, - priority: "default", - clickUrl, - signal: this.abortController?.signal, - }), - ); - } - }; - private handleSettingsUpdated = (data: { settings: Settings; previous: Settings }): void => { - const { settings, previous } = data; - - if (settings.ntfyEnabled !== previous.ntfyEnabled || - settings.ntfyTopic !== previous.ntfyTopic || - settings.ntfyBaseUrl !== previous.ntfyBaseUrl || - settings.ntfyDashboardHost !== previous.ntfyDashboardHost || - JSON.stringify(settings.ntfyEvents) !== JSON.stringify(previous.ntfyEvents)) { - const wasEnabled = this.config.enabled; - this.loadConfig(settings); - - if (this.config.enabled && !wasEnabled) { - schedulerLog.log("NtfyNotifier enabled"); - } else if (!this.config.enabled && wasEnabled) { - schedulerLog.log("NtfyNotifier disabled"); - } else if (this.config.topic !== previous.ntfyTopic) { - schedulerLog.log("NtfyNotifier topic updated"); - } else if (this.ntfyBaseUrl !== resolveNtfyBaseUrl(previous.ntfyBaseUrl)) { - schedulerLog.log("NtfyNotifier base URL updated"); - } else if (this.config.dashboardHost !== previous.ntfyDashboardHost) { - schedulerLog.log("NtfyNotifier dashboard host updated"); - } else if (JSON.stringify(this.config.events) !== JSON.stringify(previous.ntfyEvents)) { - schedulerLog.log("NtfyNotifier events updated"); - } - } + this.loadConfig(data.settings); }; private loadConfig(settings: Settings): void { @@ -398,14 +265,6 @@ export class NtfyNotifier { return isNtfyEventEnabled(this.config.events, event); } - private maybeNotify( - taskId: string, - eventType: TaskNotificationEvent, - notifyFn: () => Promise, - ): void { - this.maybeNotifyByKey(`${taskId}:${eventType}`, notifyFn); - } - private maybeNotifyByKey(key: string, notifyFn: () => Promise): void { if (this.notifiedEvents.has(key)) { return; diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index f4444fe57..d597da1df 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -16,6 +16,7 @@ import type { ProjectRuntimeConfig } from "./project-runtime.js"; import { PrMonitor } from "./pr-monitor.js"; import { PrCommentHandler } from "./pr-comment-handler.js"; import { NtfyNotifier } from "./notifier.js"; +import { NotificationService } from "./notification/index.js"; import { GridlockDetector } from "./gridlock-detector.js"; import { CronRunner, createAiPromptExecutor } from "./cron-runner.js"; import type { RoutineRunner } from "./routine-runner.js"; @@ -115,7 +116,7 @@ export interface ProjectEngineOptions { * * - **Auto-merge queue** — serialized merge with conflict retry, semaphore gating * - **PrMonitor + PrCommentHandler** — GitHub PR feedback loop - * - **NtfyNotifier** — push notifications + * - **NotificationService** — provider-driven push notifications * - **CronRunner + AutomationStore** — scheduled automations * - **Settings event listeners** — dynamic reconfiguration * @@ -128,6 +129,7 @@ export class ProjectEngine { private prMonitor?: PrMonitor; private prCommentHandler?: PrCommentHandler; private notifier?: NtfyNotifier; + private notificationService?: NotificationService; private gridlockDetector?: GridlockDetector; private cronRunner?: CronRunner; private automationStore?: AutomationStoreType; @@ -218,8 +220,15 @@ export class ProjectEngine { this.prCommentHandler!.handleNewComments(taskId, prInfo, comments), ); - // 3. Initialize NtfyNotifier (unless caller manages it externally) + // 3. Initialize notification services (unless caller manages them externally) if (!this.options.skipNotifier) { + this.notificationService = new NotificationService(store, { + projectId: this.options.projectId, + ntfyBaseUrl: this.options.ntfyBaseUrl, + }); + await this.notificationService.start(); + + // Backward-compatibility shim for gridlock notifications. this.notifier = new NtfyNotifier(store, { projectId: this.options.projectId, ntfyBaseUrl: this.options.ntfyBaseUrl, @@ -394,6 +403,7 @@ export class ProjectEngine { } // Stop auxiliary subsystems + this.notificationService?.stop(); this.notifier?.stop(); this.gridlockDetector?.stop(); this.cronRunner?.stop();