import { BACKUP_COMMAND_CHANNEL, type BackupCommand } from '@dorfteich/shared'; import { Client } from 'pg'; import type { RemoteLogger } from './remote.js'; /** * Listens for api-issued backup commands ("back up now", "restore set X", * issue #103) on the {@link BACKUP_COMMAND_CHANNEL} — the same dedicated- * connection LISTEN/NOTIFY pattern as the collab server's listeners. The * handler is invoked fire-and-forget; serialization against the nightly * schedule happens in the caller's queue (index.ts). */ export interface CommandListenerDeps { createClient(): Client; onCommand(command: BackupCommand): void; log: RemoteLogger; reconnectDelayMs?: number; } export interface CommandListener { start(): Promise; stop(): Promise; } const DEFAULT_RECONNECT_DELAY_MS = 1000; export function parseCommand(payload: string): BackupCommand | null { let parsed: unknown; try { parsed = JSON.parse(payload); } catch { return null; } const command = parsed as Partial; if (command.kind === 'run') { return { kind: 'run', requestedBy: command.requestedBy ?? null }; } if ( command.kind === 'restore' && (command.source === 'local' || command.source === 'remote') && typeof command.backupId === 'string' && /^\d{8}-\d{6}$/.test(command.backupId) ) { return { kind: 'restore', source: command.source, backupId: command.backupId, requestedBy: command.requestedBy ?? null, }; } return null; } export function createCommandListener(deps: CommandListenerDeps): CommandListener { const reconnectDelayMs = deps.reconnectDelayMs ?? DEFAULT_RECONNECT_DELAY_MS; let client: Client | null = null; let stopped = false; let reconnectTimer: NodeJS.Timeout | null = null; function scheduleReconnect(): void { if (stopped || reconnectTimer) return; reconnectTimer = setTimeout(() => { reconnectTimer = null; void connect(); }, reconnectDelayMs); reconnectTimer.unref?.(); } async function connect(): Promise { if (stopped) return; const next = deps.createClient(); next.on('error', (error) => { deps.log.warn({ error: error.message }, 'command listener connection error; will reconnect'); if (client === next) client = null; scheduleReconnect(); }); next.on('end', () => { // The restore path terminates every other connection, including this // one — reconnect quietly so the next command still arrives. if (client === next) client = null; scheduleReconnect(); }); next.on('notification', (message) => { if (message.channel !== BACKUP_COMMAND_CHANNEL || !message.payload) return; const command = parseCommand(message.payload); if (!command) { deps.log.warn({ payload: message.payload }, 'ignoring malformed backup command'); return; } deps.onCommand(command); }); try { await next.connect(); await next.query(`LISTEN ${BACKUP_COMMAND_CHANNEL}`); client = next; deps.log.info({ channel: BACKUP_COMMAND_CHANNEL }, 'listening for backup commands'); } catch (error) { deps.log.warn( { error: (error as Error).message }, 'could not start command listener; will retry', ); await next.end().catch(() => undefined); scheduleReconnect(); } } return { async start(): Promise { stopped = false; await connect(); }, async stop(): Promise { stopped = true; if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } const current = client; client = null; if (current) await current.end().catch(() => undefined); }, }; }