/** pi-ntfy payload building and dispatch. */ import { appendFileSync, renameSync, statSync } from "node:fs"; import { hostname } from "node:os"; import { basename } from "node:path"; import { channelName, sendToChannel } from "./channels/index.ts"; import { logPath } from "./config.ts"; import { toChannelResult } from "./http.ts"; import { formatDuration, render } from "./render.ts"; import type { ChannelConfig, ChannelResult, NotifyConfig, NotifyEvent, NotifyPayload, } from "./types.ts"; const LOG_ROTATE_BYTES = 256 * 1024; export interface PayloadInput { event: NotifyEvent; status: string; reason: string; cwd?: string; durationMs?: number; session?: string; sessionId?: string; model?: string; mode?: string; thinking?: string; task?: string; summary?: string; tools?: string; } export function buildPayload( config: NotifyConfig, input: PayloadInput, ): NotifyPayload { const now = new Date(); const cwd = input.cwd ?? process.cwd(); const vars: Record = { event: input.event, status: input.status, reason: input.reason, cwd, project: basename(cwd) || cwd, duration: input.durationMs === undefined ? "" : formatDuration(input.durationMs), session: input.session ?? "", sessionId: input.sessionId ?? "", model: input.model ?? "", mode: input.mode ?? "", thinking: input.thinking ?? "", task: input.task ?? "", summary: input.summary ?? "", tools: input.tools ?? "", host: hostname(), time: now.toTimeString().slice(0, 8), date: now.toISOString().slice(0, 10), }; const title = render(config.titleTemplate, vars).trim() || "pi"; let text = render(config.template, vars).trim() || title; if (text.length > config.maxTextChars) text = `${text.slice(0, config.maxTextChars - 1)}…`; // NTFY accepts plain text bodies; metadata is carried in headers. vars.title = title; vars.text = text; return { event: input.event, status: input.status, reason: input.reason, title, text, vars, }; } interface Target { channel: ChannelConfig; name: string; } export function resolveTargets( config: NotifyConfig, event: NotifyEvent, ): Target[] { const targets: Target[] = []; config.channels.forEach((channel, index) => { if (channel.enabled === false) return; if (event !== "test") { const override = channel.events?.[event]; const allowed = override ?? config.events[event] ?? false; if (!allowed) return; } targets.push({ channel, name: channelName(channel, index) }); }); return targets; } function writeLog(config: NotifyConfig, line: string): void { const path = logPath(); try { const stat = statSync(path, { throwIfNoEntry: false }); if (stat && stat.size > LOG_ROTATE_BYTES) renameSync(path, `${path}.1`); } catch { // rotation is best-effort } try { appendFileSync(path, `${new Date().toISOString()} ${line}\n`, "utf-8"); } catch { // never let logging break a notification } void config; } function capPromise( promise: Promise, ms: number, fallback: T, ): Promise { return Promise.race([ promise, new Promise((resolve) => { const timer = setTimeout(() => resolve(fallback), ms); // Do not hold the event loop open while pi is exiting. (timer as { unref?: () => void }).unref?.(); }), ]); } export interface Notifier { /** Non-blocking send; failures are logged, never thrown. */ fire(config: NotifyConfig, input: PayloadInput): void; /** Awaited send with a hard cap — used on the shutdown path. */ fireBlocking( config: NotifyConfig, input: PayloadInput, capMs: number, ): Promise; /** Ignores event switches and pushes to every enabled channel. */ sendTest(config: NotifyConfig, input: PayloadInput): Promise; /** Wait for in-flight sends (bounded). */ flush(capMs: number): Promise; lastResults(): | { at: number; event: NotifyEvent; results: ChannelResult[] } | undefined; } export function createNotifier(): Notifier { const inflight = new Set>(); let lastKey = ""; let lastAt = 0; let last: | { at: number; event: NotifyEvent; results: ChannelResult[] } | undefined; async function dispatch( config: NotifyConfig, payload: NotifyPayload, targets: Target[], ): Promise { const results = await Promise.all( targets.map(async ({ channel, name }) => { try { const result = await sendToChannel( channel, payload, channel.timeoutMs ?? config.timeoutMs, ); return toChannelResult(name, result); } catch (error) { return { channel: name, ok: false, status: 0, error: error instanceof Error ? error.message : String(error), ms: 0, } satisfies ChannelResult; } }), ); last = { at: Date.now(), event: payload.event, results }; for (const result of results) { if (result.ok && !config.debug) continue; const verdict = result.ok ? "ok" : "FAIL"; const detail = result.ok ? `status=${result.status}` : (result.error ?? "unknown"); writeLog( config, `[${payload.event}] ${result.channel} ${verdict} ${detail} (${result.ms}ms)`, ); } return results; } function prepare( config: NotifyConfig, input: PayloadInput, ignoreSwitches: boolean, ): { payload: NotifyPayload; targets: Target[] } | undefined { if (!config.enabled) return undefined; const targets = resolveTargets(config, ignoreSwitches ? "test" : input.event); if (targets.length === 0) return undefined; const payload = buildPayload(config, input); if (!ignoreSwitches && config.dedupeMs > 0) { const key = `${payload.event}|${payload.text}`; const now = Date.now(); if (key === lastKey && now - lastAt < config.dedupeMs) return undefined; lastKey = key; lastAt = now; } return { payload, targets }; } function track(promise: Promise): Promise { const tracked = promise.finally(() => { inflight.delete(tracked); }); inflight.add(tracked); return tracked; } return { fire(config, input) { const prepared = prepare(config, input, false); if (!prepared) return; void track( dispatch(config, prepared.payload, prepared.targets).catch( () => [] as ChannelResult[], ), ); }, async fireBlocking(config, input, capMs) { const prepared = prepare(config, input, false); if (!prepared) return []; const run = track( dispatch(config, prepared.payload, prepared.targets).catch( () => [] as ChannelResult[], ), ); return capPromise(run, capMs, []); }, async sendTest(config, input) { const prepared = prepare(config, input, true); if (!prepared) return []; return track( dispatch(config, prepared.payload, prepared.targets).catch( () => [] as ChannelResult[], ), ); }, async flush(capMs) { if (inflight.size === 0) return; await capPromise( Promise.allSettled([...inflight]).then(() => undefined), capMs, undefined, ); }, lastResults() { return last; }, }; }