feat: push selected events to a phone over ntfy

Adds a Notifications tab to Settings so an operator can pick which events
reach their phone, published to a single self-hosted ntfy topic.

The notifier subscribes to the realtime event hub the dashboard already
consumes, so every notifiable event keeps exactly one publisher. Inbound
media had no event, so `media.received` is added and published from the
existing durable `gotImage` gate in the chat service — one notification
per conversation, never repeated when the room re-syncs. It feeds the
in-app notification bell as well.

- Best-effort delivery: a rejecting or unreachable ntfy server is logged
  and never propagated into the path that produced the event.
- Events buffered before the notifier attaches are skipped, so a restart
  cannot replay a backlog to somebody's phone.
- Settings are applied field by field onto the live config object, so a
  save takes effect without a restart.
- The access token is environment-owned (NTFY_AUTH_TOKEN) like the other
  secrets: stripped before the settings record is written, never returned
  by the API, which reports only `hasAuthToken`.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Jamie BruntonandClaude Opus 5 committed 2026-08-10 13:55:00 +01:00
1 parent 827eb3a1e0
commit 38d175e4a1
34 files changed
+1728 -8

No files matched your search

+2
View File
@@ -7,6 +7,8 @@ export type {
SyncPhase,
SyncCompletion,
SyncProgressListener,
NewReplyListener,
MediaReceivedListener,
ConversationStartResult,
} from './service.js';
export type {
+28 -2
View File
@@ -135,6 +135,13 @@ export type NewReplyListener = (info: {
replySnippet: string;
}) => void;
/** Fired the first time a conversation we opened receives inbound media. */
export type MediaReceivedListener = (info: {
conversationId: string;
username: string;
count: number;
}) => void;
export class ChatService {
private cache: ChatCache;
private queue: Promise<unknown> = Promise.resolve();
@@ -147,6 +154,7 @@ export class ChatService {
private progress: SyncProgress | null = null;
private progressListener: SyncProgressListener | null = null;
private newReplyListener: NewReplyListener | null = null;
private mediaReceivedListener: MediaReceivedListener | null = null;
private invitationCooldown: InvitationCooldown | null = null;
constructor(
@@ -535,6 +543,11 @@ export class ChatService {
this.newReplyListener = listener;
}
/** Subscribe to inbound-media notifications (fires once per chat, on the first image). */
setMediaReceivedListener(listener: MediaReceivedListener | null): void {
this.mediaReceivedListener = listener;
}
private beginSync(kind: 'full' | 'delta'): void {
this.syncingKind = kind;
this.progress = {
@@ -1167,12 +1180,25 @@ export class ChatService {
);
const incomingMessages = conversationMessages.filter((message) => !message.isSelf);
const hasReply = incomingMessages.length > 0;
const hasReceivedImage = incomingMessages.some(
(message) => (message.media?.length ?? 0) > 0,
const incomingMediaCount = incomingMessages.reduce(
(total, message) => total + (message.media?.length ?? 0),
0,
);
const hasReceivedImage = incomingMediaCount > 0;
const firstAttributedReply = hasReply && !performanceMessage.gotResponse;
if (hasReceivedImage && !performanceMessage.gotImage) {
this.messagesRepo.markGotImage(performanceMessage.id);
// Guarded by the durable `gotImage` flag, so re-syncing a conversation
// cannot re-notify for media that was already reported.
try {
this.mediaReceivedListener?.({
conversationId: chat.roomId,
username: chat.username,
count: incomingMediaCount,
});
} catch (err) {
this.logger.warn('[chat] mediaReceivedListener threw:', err);
}
}
if (hasReply && this.successfulMessagesRepo && this.outcomesRepo) {
+217
View File
@@ -0,0 +1,217 @@
import type { Logger } from '../utils/logger.js';
import type { Config } from '../config/index.js';
import type { RealtimeEvent } from '../web/events/event-types.js';
export type PushNotificationSettings = Config['pushNotifications'];
export type PushNotificationEventKey = keyof PushNotificationSettings['events'];
export interface PushNotificationDraft {
event: PushNotificationEventKey;
title: string;
body: string;
/** ntfy emoji shortcodes rendered alongside the notification. */
tags: string[];
/** Overrides the configured priority when an event is more urgent. */
priority?: PushNotificationSettings['priority'];
}
export interface PushDeliveryResult {
delivered: boolean;
reason?: string;
}
/** Minimal surface of the realtime hub the notifier needs. */
export interface PushEventSource {
subscribe(listener: (event: RealtimeEvent) => void): { unsubscribe(): void };
}
type Fetcher = (input: string, init: RequestInit) => Promise<{ ok: boolean; status: number; statusText: string }>;
const REQUEST_TIMEOUT_MS = 10_000;
const MAX_BODY_LENGTH = 400;
/**
* Maps a realtime event onto the push notification an operator would want on
* their phone, or null when the event is not notification-worthy. Pure, so the
* copy and the event selection are testable without any transport.
*/
export function pushNotificationForEvent(event: RealtimeEvent): PushNotificationDraft | null {
switch (event.type) {
case 'media.received':
return {
event: 'imageReceived',
title: `Image from u/${event.payload.sender}`,
body: event.payload.count > 1
? `${event.payload.count} images are attached to the conversation.`
: 'An image is attached to the conversation.',
tags: ['camera'],
};
case 'reply.received':
return {
event: 'replyReceived',
title: `Reply from u/${event.payload.sender}`,
body: event.payload.preview?.trim() || 'A new message arrived in the conversation.',
tags: ['speech_balloon'],
};
case 'message.failed':
return {
event: 'messageFailed',
title: `Message to u/${event.payload.recipient} failed`,
body: event.payload.reason,
tags: ['warning'],
};
case 'bot.status.changed':
if (event.payload.status !== 'stopped') return null;
return {
event: 'botStopped',
title: 'Bot stopped',
body: event.payload.reason ?? 'The bot is no longer processing candidates.',
tags: ['octagonal_sign'],
};
case 'rate-limit.changed':
if (!event.payload.limited) return null;
return {
event: 'rateLimited',
title: 'Rate limit reached',
body: event.payload.reason ?? 'Sending is paused until the limit resets.',
tags: ['hourglass'],
};
case 'reddit.session.changed':
if (!['expired', 'logged-out', 'error'].includes(event.payload.status)) return null;
return {
event: 'sessionExpired',
title: 'Reddit account needs attention',
body: event.payload.reason
?? (event.payload.username
? `The session for u/${event.payload.username} is ${event.payload.status}.`
: `The Reddit session is ${event.payload.status}.`),
tags: ['rotating_light'],
priority: 'high',
};
case 'sync.failed':
return {
event: 'syncFailed',
title: `${event.payload.kind} sync failed`,
body: event.payload.reason,
tags: ['warning'],
};
default:
return null;
}
}
/**
* Publishes selected application events to a single ntfy topic so they reach
* the operator's phone. Delivery is best-effort: a failing or unreachable ntfy
* server is logged and never propagated into the calling path.
*/
export class PushNotifier {
private subscription: { unsubscribe(): void } | null = null;
private attachedAt = 0;
constructor(
private readonly getSettings: () => PushNotificationSettings,
private readonly logger: Logger,
private readonly fetcher: Fetcher = globalThis.fetch as unknown as Fetcher,
private readonly now: () => number = Date.now,
) {}
/**
* Subscribes to the realtime hub. Events buffered before this call are
* skipped: the hub replays its buffer to every new subscriber, and replaying
* a restart's worth of history to somebody's phone is not useful.
*/
attach(source: PushEventSource): void {
if (this.subscription) return;
this.attachedAt = this.now();
this.subscription = source.subscribe((event) => {
if (Date.parse(event.occurredAt) < this.attachedAt) return;
const draft = pushNotificationForEvent(event);
if (!draft) return;
void this.publish(draft);
});
}
detach(): void {
this.subscription?.unsubscribe();
this.subscription = null;
}
/** Sends a notification if the event is enabled and the topic is configured. */
async publish(draft: PushNotificationDraft): Promise<PushDeliveryResult> {
const settings = this.getSettings();
if (!settings.enabled) return { delivered: false, reason: 'Push notifications are disabled.' };
if (!settings.events[draft.event]) {
return { delivered: false, reason: `The ${draft.event} event is not selected.` };
}
return this.deliver(draft, settings);
}
/** Sends a fixed notification regardless of the per-event selection. */
sendTest(): Promise<PushDeliveryResult> {
const settings = this.getSettings();
if (!settings.enabled) return Promise.resolve({ delivered: false, reason: 'Push notifications are disabled.' });
return this.deliver({
event: 'imageReceived',
title: 'RedditBot test notification',
body: 'Push notifications are configured correctly.',
tags: ['white_check_mark'],
}, settings);
}
private async deliver(
draft: PushNotificationDraft,
settings: PushNotificationSettings,
): Promise<PushDeliveryResult> {
const topic = settings.topic.trim();
if (topic === '') return { delivered: false, reason: 'No ntfy topic is configured.' };
let endpoint: string;
try {
endpoint = new URL(topic, `${settings.serverUrl.trim().replace(/\/+$/, '')}/`).toString();
} catch {
return { delivered: false, reason: `Invalid ntfy server URL: ${settings.serverUrl}` };
}
const headers: Record<string, string> = {
'Content-Type': 'text/plain; charset=utf-8',
Title: headerSafe(draft.title),
Priority: draft.priority ?? settings.priority,
Tags: draft.tags.join(','),
};
if (settings.authToken) headers.Authorization = `Bearer ${settings.authToken}`;
try {
const response = await this.fetcher(endpoint, {
method: 'POST',
headers,
body: truncate(draft.body, MAX_BODY_LENGTH),
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
});
if (!response.ok) {
const reason = `ntfy responded ${response.status} ${response.statusText}`.trim();
this.logger.warn(`[push] ${draft.event} notification rejected: ${reason}`);
return { delivered: false, reason };
}
return { delivered: true };
} catch (error) {
const reason = error instanceof Error ? error.message : String(error);
this.logger.warn(`[push] ${draft.event} notification failed: ${reason}`);
return { delivered: false, reason };
}
}
}
/**
* Titles travel as an HTTP header, so newlines (which would split the header)
* and non-ASCII bytes are removed. The body keeps its UTF-8 text.
*/
function headerSafe(value: string): string {
// eslint-disable-next-line no-control-regex
const ascii = value.replace(/[\r\n]+/g, ' ').replace(/[^\x20-\x7E]/g, '').trim();
return ascii.length <= 120 ? ascii : `${ascii.slice(0, 117)}...`;
}
function truncate(value: string, limit: number): string {
return value.length <= limit ? value : `${value.slice(0, limit - 1)}…`;
}