diff --git a/dashboard/jest.setup.cjs b/dashboard/jest.setup.cjs index edc7f6fb..0a18b4aa 100644 --- a/dashboard/jest.setup.cjs +++ b/dashboard/jest.setup.cjs @@ -1,7 +1,5 @@ require('@testing-library/jest-dom'); -const { jest } = require('@types/jest'); - const originalFetch = global.fetch; const originalSetTimeout = global.setTimeout; const originalSetInterval = global.setInterval; @@ -136,7 +134,7 @@ function failureInjectionFetch(url, init) { return originalFetch(url, init); } -fobject.defineProperty(HTMLElement.prototype, 'clientHeight', { +Object.defineProperty(HTMLElement.prototype, 'clientHeight', { configurable: true, get() { return 600; diff --git a/dashboard/src/hooks/useTheme.ts b/dashboard/src/hooks/useTheme.ts index ef094294..070e7827 100644 --- a/dashboard/src/hooks/useTheme.ts +++ b/dashboard/src/hooks/useTheme.ts @@ -27,7 +27,11 @@ function applyTheme(theme: Theme): void { * and applies a `data-theme` attribute to `` for CSS consumption. */ export function useTheme() { - const [theme, setThemeState] = useState(getInitialTheme); + const [theme, setThemeState] = useState(() => { + const initial = getInitialTheme(); + if (typeof document !== 'undefined') applyTheme(initial); + return initial; + }); useEffect(() => { applyTheme(theme); diff --git a/dashboard/src/index.css b/dashboard/src/index.css index 64a4347e..1f674216 100644 --- a/dashboard/src/index.css +++ b/dashboard/src/index.css @@ -5,6 +5,9 @@ font-weight: 400; color: #e8eaed; background: #0b0d12; + --text-muted: #9aa0a6; + --text-accent: #fb923c; + --text-danger: #f28b82; /* ── Heatmap design tokens (dark) ── */ --hm-bg: rgba(255, 255, 255, 0.02); @@ -17,7 +20,7 @@ --hm-cell-hover-ring: rgba(255, 255, 255, 0.25); --hm-text-primary: #e8eaed; --hm-text-secondary: #9aa0a6; - --hm-text-tertiary: #6b7280; + --hm-text-tertiary: #7d8694; --hm-tooltip-bg: #1e2128; --hm-tooltip-border: rgba(255, 255, 255, 0.14); --hm-input-bg: #12151c; @@ -58,6 +61,9 @@ color-scheme: light; color: #1a1a2e; background: #f5f5f9; + --text-muted: #4b5563; + --text-accent: #9a3412; + --text-danger: #b91c1c; --hm-bg: rgba(0, 0, 0, 0.02); --hm-border: rgba(0, 0, 0, 0.1); @@ -69,7 +75,7 @@ --hm-cell-hover-ring: rgba(0, 0, 0, 0.2); --hm-text-primary: #1a1a2e; --hm-text-secondary: #4b5563; - --hm-text-tertiary: #9ca3af; + --hm-text-tertiary: #5b6570; --hm-tooltip-bg: #ffffff; --hm-tooltip-border: rgba(0, 0, 0, 0.12); --hm-input-bg: #ffffff; @@ -123,10 +129,6 @@ color: var(--hm-accent); } -[data-theme="light"] .event-explorer__lead { - color: #4b5563; -} - [data-theme="light"] .event-explorer__table-wrapper { border-color: rgba(0, 0, 0, 0.1); background: rgba(255, 255, 255, 0.7); @@ -2906,7 +2908,7 @@ body { .event-explorer__lead { margin: 0; max-width: 700px; - color: #cbd5e1; + color: var(--text-muted); line-height: 1.75; } @@ -4018,7 +4020,7 @@ body.event-explorer--resizing { .timeline__entry-attempt { font-size: 0.8rem; - color: #9aa0a6; + color: var(--text-muted); background: rgba(255,255,255,0.07); padding: 2px 6px; border-radius: 4px; @@ -4027,13 +4029,13 @@ body.event-explorer--resizing { .timeline__entry-time { margin-left: auto; font-size: 0.8rem; - color: #9aa0a6; + color: var(--text-muted); } .timeline__entry-error { margin: 2px 0 0; font-size: 0.85rem; - color: #f28b82; + color: var(--text-danger); } .timeline__entry-duration { @@ -4226,7 +4228,7 @@ body.event-explorer--resizing { .activity-feed__subtitle { margin: 0; - color: #9aa0a6; + color: var(--text-muted); } .activity-feed__toggle-realtime { @@ -4460,7 +4462,7 @@ body.event-explorer--resizing { font-size: 0.75rem; letter-spacing: 0.1em; text-transform: uppercase; - color: #fb923c; + color: var(--text-accent); } .webhook-dashboard__title { @@ -4471,7 +4473,7 @@ body.event-explorer--resizing { .webhook-dashboard__lead { margin: 0; max-width: 680px; - color: #cbd5e1; + color: var(--text-muted); line-height: 1.75; } diff --git a/dashboard/src/pages/EventExplorerPage.tsx b/dashboard/src/pages/EventExplorerPage.tsx index 788444db..cf1f85e1 100644 --- a/dashboard/src/pages/EventExplorerPage.tsx +++ b/dashboard/src/pages/EventExplorerPage.tsx @@ -22,6 +22,7 @@ import { isListenerApiTimeoutError, type ContractStatus, } from '../services/eventsApi'; +import { openEventStream, toEventStreamUrl } from '../services/eventStream'; import { resolveIndexingHealthUrl } from '../services/indexingHealthApi'; import { resolveNotificationHealthUrl } from '../services/notificationHealthApi'; import { generateMockEvents } from '../utils/eventData'; @@ -34,6 +35,7 @@ const DEFAULT_EVENT_COUNT = 5000; const DEFAULT_LIMIT = 12; const EVENT_ID_PARAM = 'eventId'; const API_URL = import.meta.env.VITE_EVENTS_API_URL ?? 'http://localhost:8787/api/events'; +const STREAM_URL = toEventStreamUrl(API_URL); const POLL_INTERVAL_MS = 15_000; const LISTENER_BASE_URL = API_URL.replace('/api/events', ''); const INDEXING_HEALTH_URL = @@ -72,6 +74,7 @@ export function EventExplorerPage() { const events = useEventStore((state) => state.events); const setEvents = useEventStore((state) => state.setEvents); + const appendEvents = useEventStore((state) => state.appendEvents); const setLoading = useEventStore((state) => state.setLoading); const setError = useEventStore((state) => state.setError); const markSyncSuccess = useEventStore((state) => state.markSyncSuccess); @@ -141,27 +144,56 @@ export function EventExplorerPage() { loadEvents(); loadStatus(); - // Poll for status updates so delivered/failed notifications are reflected - // without requiring a manual page refresh. - const intervalId = setInterval(async () => { - try { - const remoteEvents = await fetchEvents(API_URL); - if (!cancelled) { - setEvents(remoteEvents); - markSyncSuccess(); - } - } catch { - if (!cancelled) { - markSyncFailure('Background refresh failed'); + return () => { + cancelled = true; + }; + }, [lastFetchedAt, setEvents, setError, setLoading]); + + // Live updates stay outside the fetch effect. That effect stores events and + // changes lastFetchedAt, and its cleanup would otherwise close the stream + // (and the poll timer) as soon as the first response arrives. + useEffect(() => { + let cancelled = false; + let intervalId: ReturnType | undefined; + + const startPolling = () => { + if (intervalId || cancelled) return; + intervalId = setInterval(async () => { + try { + const remoteEvents = await fetchEvents(API_URL); + if (!cancelled) { + setEvents(remoteEvents); + markSyncSuccess(); + } + } catch { + if (!cancelled) { + markSyncFailure('Background refresh failed'); + } } - } - }, POLL_INTERVAL_MS); + }, POLL_INTERVAL_MS); + }; + + const sseDisabled = import.meta.env.VITE_EVENTS_SSE === 'false'; + const closeStream = + !sseDisabled && STREAM_URL + ? openEventStream(STREAM_URL, { + onEvent: (event) => { + if (cancelled) return; + appendEvents([event]); + markSyncSuccess(); + }, + onFallback: startPolling, + }) + : undefined; + + if (!closeStream) startPolling(); return () => { cancelled = true; - clearInterval(intervalId); + if (intervalId) clearInterval(intervalId); + closeStream?.(); }; - }, [lastFetchedAt, setEvents, setError, setLoading]); + }, [appendEvents, setEvents, markSyncSuccess, markSyncFailure]); // Clear stale events and re-fetch whenever the connected wallet address // changes (switch or disconnect). This is the fix for issue #175. diff --git a/dashboard/src/services/eventStream.test.ts b/dashboard/src/services/eventStream.test.ts new file mode 100644 index 00000000..647f81b6 --- /dev/null +++ b/dashboard/src/services/eventStream.test.ts @@ -0,0 +1,90 @@ +import { openEventStream, toEventStreamUrl } from './eventStream'; +import type { BlockchainEvent } from '../types/event'; + +class FakeEventSource { + static latest: FakeEventSource | null = null; + url: string; + onopen: ((ev: Event) => void) | null = null; + onerror: ((ev: Event) => void) | null = null; + closed = false; + private listeners = new Map void>(); + + constructor(url: string) { + this.url = url; + FakeEventSource.latest = this; + } + + addEventListener(type: string, listener: (event: MessageEvent) => void) { + this.listeners.set(type, listener); + } + + emit(type: string, data: string) { + this.listeners.get(type)?.({ data } as MessageEvent); + } + + close() { + this.closed = true; + } +} + +describe('event stream client', () => { + it('maps the events URL onto the stream endpoint', () => { + expect(toEventStreamUrl('http://localhost:8787/api/events')).toBe( + 'http://localhost:8787/api/events/stream', + ); + expect(toEventStreamUrl('http://localhost:8787/api/events/history')).toBeNull(); + }); + + it('delivers notification records and closes on cleanup', () => { + const received: BlockchainEvent[] = []; + const stop = openEventStream( + 'http://localhost:8787/api/events/stream', + { onEvent: (event) => received.push(event), onFallback: () => undefined }, + FakeEventSource as unknown as new (url: string) => FakeEventSource, + ); + + FakeEventSource.latest?.onopen?.(new Event('open')); + FakeEventSource.latest?.emit( + 'notification', + JSON.stringify({ + eventId: 'evt-1', + contractAddress: 'CABC', + eventName: 'notification_scheduled', + ledger: 1, + type: 'contract', + topic: [], + value: '{}', + receivedAt: 1, + }), + ); + + expect(received).toHaveLength(1); + expect(received[0]?.notificationStatus).toBe('active'); + + stop(); + expect(FakeEventSource.latest?.closed).toBe(true); + }); + + it('falls back when the first connection fails and does not fall back after a successful open', () => { + const fallback = jest.fn(); + openEventStream( + 'http://localhost:8787/api/events/stream', + { onEvent: () => undefined, onFallback: fallback }, + FakeEventSource as unknown as new (url: string) => FakeEventSource, + ); + + FakeEventSource.latest?.onerror?.(new Event('error')); + expect(fallback).toHaveBeenCalledTimes(1); + expect(FakeEventSource.latest?.closed).toBe(true); + + fallback.mockClear(); + openEventStream( + 'http://localhost:8787/api/events/stream', + { onEvent: () => undefined, onFallback: fallback }, + FakeEventSource as unknown as new (url: string) => FakeEventSource, + ); + FakeEventSource.latest?.onopen?.(new Event('open')); + FakeEventSource.latest?.onerror?.(new Event('error')); + expect(fallback).not.toHaveBeenCalled(); + }); +}); diff --git a/dashboard/src/services/eventStream.ts b/dashboard/src/services/eventStream.ts new file mode 100644 index 00000000..978fef36 --- /dev/null +++ b/dashboard/src/services/eventStream.ts @@ -0,0 +1,73 @@ +import type { BlockchainEvent } from '../types/event'; +import { hydrateNotificationStatus } from './eventsApi'; + +export interface EventStreamHandlers { + onEvent: (event: BlockchainEvent) => void; + /** Called once when the stream never opens, so the caller can poll instead. */ + onFallback: () => void; +} + +interface EventSourceLike { + addEventListener(type: string, listener: (event: MessageEvent) => void): void; + close(): void; + onopen: ((this: EventSourceLike, ev: Event) => void) | null; + onerror: ((this: EventSourceLike, ev: Event) => void) | null; +} + +type EventSourceConstructor = new (url: string) => EventSourceLike; + +/** + * Subscribe to `GET /api/events/stream`. + * + * The browser EventSource reconnects on its own and resends `Last-Event-ID`. + * If the first attempt fails (stream disabled or listener down), `onFallback` + * runs once and the socket is closed so polling can take over. + */ +export function openEventStream( + streamUrl: string, + handlers: EventStreamHandlers, + EventSourceImpl?: EventSourceConstructor, +): () => void { + const Ctor = EventSourceImpl ?? (globalThis.EventSource as EventSourceConstructor | undefined); + if (!Ctor) { + handlers.onFallback(); + return () => undefined; + } + + let opened = false; + let closed = false; + const source = new Ctor(streamUrl); + + source.onopen = () => { + opened = true; + }; + + source.onerror = () => { + if (opened || closed) return; + closed = true; + source.close(); + handlers.onFallback(); + }; + + source.addEventListener('notification', (message) => { + if (closed) return; + try { + const parsed = JSON.parse(message.data) as BlockchainEvent; + const [event] = hydrateNotificationStatus([parsed]); + if (event) handlers.onEvent(event); + } catch { + // Ignore a malformed record and keep the stream open. + } + }); + + return () => { + closed = true; + source.close(); + }; +} + +/** Map a `GET /api/events` URL onto the SSE endpoint. */ +export function toEventStreamUrl(eventsApiUrl: string): string | null { + if (!/\/api\/events\/?$/.test(eventsApiUrl.split('?')[0] ?? '')) return null; + return eventsApiUrl.replace(/\/api\/events\/?(\?.*)?$/, '/api/events/stream$1'); +} diff --git a/dashboard/src/services/eventsApi.ts b/dashboard/src/services/eventsApi.ts index f7d6776f..306d0b33 100644 --- a/dashboard/src/services/eventsApi.ts +++ b/dashboard/src/services/eventsApi.ts @@ -10,7 +10,7 @@ import { NOTIFICATION_STATUS_EVENTS } from '../types/event'; * set from the first render — not left `undefined` until a mutation action * patches them in. */ -function hydrateNotificationStatus(events: BlockchainEvent[]): BlockchainEvent[] { +export function hydrateNotificationStatus(events: BlockchainEvent[]): BlockchainEvent[] { return events.map((event) => { if (!event.eventName) return event; const status = NOTIFICATION_STATUS_EVENTS[event.eventName]; diff --git a/dashboard/src/theme/themeContrast.test.ts b/dashboard/src/theme/themeContrast.test.ts new file mode 100644 index 00000000..6d54c5d6 --- /dev/null +++ b/dashboard/src/theme/themeContrast.test.ts @@ -0,0 +1,69 @@ +import * as fs from 'fs'; +import * as path from 'path'; + +/** WCAG relative luminance contrast for two #rrggbb colors. */ +export function contrastRatio(foreground: string, background: string): number { + const luminance = (hex: string) => { + const channels = [0, 2, 4].map((index) => { + const value = parseInt(hex.slice(index + 1, index + 3), 16) / 255; + return value <= 0.03928 ? value / 12.92 : ((value + 0.055) / 1.055) ** 2.4; + }); + return 0.2126 * channels[0] + 0.7152 * channels[1] + 0.0722 * channels[2]; + }; + const lighter = Math.max(luminance(foreground), luminance(background)); + const darker = Math.min(luminance(foreground), luminance(background)); + return (lighter + 0.05) / (darker + 0.05); +} + +function block(css: string, selector: string): string { + const escaped = selector.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); + const match = css.match(new RegExp(`${escaped}\\s*\\{([^}]*)\\}`)); + return match?.[1] ?? ''; +} + +function decl(rules: string, name: string): string { + const match = rules.match(new RegExp(`${name}\\s*:\\s*(#[0-9a-fA-F]{6})`)); + if (!match) throw new Error(`missing ${name}`); + return match[1]; +} + +const css = fs.readFileSync(path.join(__dirname, '../index.css'), 'utf8'); + +const KEY_SCREENS = [ + '.event-explorer__lead', + '.activity-feed__subtitle', + '.timeline__entry-time', + '.webhook-dashboard__lead', + '.webhook-dashboard__eyebrow', + '.timeline__entry-error', +]; + +describe('dashboard theme contrast', () => { + const themes = [ + { name: 'dark', selector: ':root' }, + { name: 'light', selector: '[data-theme="light"]' }, + ] as const; + + it.each(themes)('$name text tokens meet WCAG AA against the page background', ({ selector }) => { + const rules = block(css, selector); + const background = decl(rules, 'background'); + for (const token of ['color', '--text-muted', '--text-accent', '--text-danger', '--hm-text-tertiary']) { + const foreground = decl(rules, token); + expect(contrastRatio(foreground, background)).toBeGreaterThanOrEqual(4.5); + } + }); + + it('binds key screens to theme tokens so both modes stay readable', () => { + const bindings: Record = { + '.event-explorer__lead': 'var(--text-muted)', + '.activity-feed__subtitle': 'var(--text-muted)', + '.timeline__entry-time': 'var(--text-muted)', + '.webhook-dashboard__lead': 'var(--text-muted)', + '.webhook-dashboard__eyebrow': 'var(--text-accent)', + '.timeline__entry-error': 'var(--text-danger)', + }; + for (const selector of KEY_SCREENS) { + expect(block(css, selector)).toContain(bindings[selector]); + } + }); +}); diff --git a/listener/.env.example b/listener/.env.example index e3eb452e..1184747f 100644 --- a/listener/.env.example +++ b/listener/.env.example @@ -58,6 +58,11 @@ CONTRACT_ADDRESSES=[{"address":"CXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX # Port the listener HTTP server listens on. EVENTS_API_PORT=8787 +# Optional Server-Sent Events feed at GET /api/events/stream. +# Leave false to keep dashboards on GET /api/events polling. +# EVENT_STREAM_ENABLED=false +# EVENT_STREAM_HEARTBEAT_MS=15000 + # Allowed CORS origin for the events API. Set to your dashboard URL in production. # Avoid using '*' in production environments. EVENTS_API_CORS_ORIGIN=http://localhost:5173 @@ -260,6 +265,13 @@ SCHEDULER_BATCH_SIZE=10 # Timing buffer to avoid dispatching notifications too early or too late (ms). SCHEDULER_TIMING_BUFFER_MS=60000 +# What to do with a notification whose executeAt passed while NotifyChain was +# unavailable, once it is later than the timing buffer: +# execute — deliver it late (default) +# expire — mark it EXPIRED and do not deliver +# intervene — hold it until POST /api/schedule/:id/resume +# SCHEDULER_MISSED_POLICY=execute + # ----------------------------------------------------------------------------- # Event Processing Queue # ----------------------------------------------------------------------------- diff --git a/listener/API_USAGE_COOKBOOK.md b/listener/API_USAGE_COOKBOOK.md index b8dbdaf3..97584ebf 100644 --- a/listener/API_USAGE_COOKBOOK.md +++ b/listener/API_USAGE_COOKBOOK.md @@ -87,6 +87,30 @@ notifier, or a downstream analytics pipeline. curl -sS 'http://localhost:8787/api/events?limit=5' | jq '.events[] | {eventId, contractAddress, topic, value, ledger}' ``` +### Live feed (optional Server-Sent Events) + +When the listener is started with `EVENT_STREAM_ENABLED=true`, dashboards can +subscribe instead of polling: + +```bash +curl -N http://localhost:8787/api/events/stream +``` + +Each record uses this format: + +``` +id: : +event: notification +data: {"eventId":"...","contractAddress":"...","eventName":"...","ledger":1,"type":"contract","topic":[],"value":"...","receivedAt":0} + +``` + +Reconnect by sending the last id in the `Last-Event-ID` header. The server +replays anything newer, then resumes live delivery. A `: keepalive` comment +is sent about every 15 seconds. When the client disconnects, the listener +drops that subscriber. If the flag is off, this path returns 404 and the poll +loop below remains the way to read events. + ### Poll on an interval (cron / watch loop) ```bash diff --git a/listener/DST_SCHEDULE_POLICY.md b/listener/DST_SCHEDULE_POLICY.md new file mode 100644 index 00000000..eac36957 --- /dev/null +++ b/listener/DST_SCHEDULE_POLICY.md @@ -0,0 +1,17 @@ +# Daylight-saving schedule policy + +Timezone-based schedules use a local wall time (`localDateTime`) plus an IANA +timezone (`timeZone`). `POST /api/schedule` accepts those fields in place of +an absolute `executeAt`. Optional `recurrence` is `daily` or `weekly`. + +| Local situation | What NotifyChain does | +| --- | --- | +| The wall time exists once | Schedule that UTC instant (`unique`). | +| The wall time is skipped because clocks jump forward | Move the wall time forward by the length of that zone's gap, then schedule that valid instant (`skipped`). `America/New_York` `2026-03-08T02:30:00` runs at `03:30` EDT (`2026-03-08T07:30:00.000Z`). A 30-minute gap is not rounded up to an hour: `Australia/Lord_Howe` `2026-10-04T02:15:00` runs at `02:45` (`2026-10-03T15:45:00.000Z`). | +| The wall time happens twice because clocks fall back | Schedule the earlier instant (`repeated`). `America/New_York` `2026-11-01T01:30:00` runs at `2026-11-01T05:30:00.000Z` (EDT), not the later EST reading. | + +Daily and weekly recurrence advances the requested local calendar date, then +applies the same rules. A 02:30 schedule that was shifted to 03:30 on the +spring-forward day returns to 02:30 on the next local date. + +The resolver lives in `listener/src/services/local-schedule.ts`. diff --git a/listener/src/__tests__/missed-schedule-recovery.integration.test.ts b/listener/src/__tests__/missed-schedule-recovery.integration.test.ts new file mode 100644 index 00000000..fea706da --- /dev/null +++ b/listener/src/__tests__/missed-schedule-recovery.integration.test.ts @@ -0,0 +1,129 @@ +/** + * Missed-schedule recovery: overdue notifications execute, expire, or wait + * for an operator depending on SCHEDULER missedSchedulePolicy. + */ + +import * as fs from 'fs'; +import * as path from 'path'; +import { Database } from '../database/database'; +import { ScheduledNotificationRepository } from '../services/scheduled-notification-repository'; +import { NotificationScheduler } from '../services/notification-scheduler'; +import { NotificationAPI } from '../services/notification-api'; +import { DiscordNotificationService } from '../services/discord-notification'; +import { resetWorkerManager } from '../services/worker-manager'; +import { NotificationStatus, NotificationType, MissedSchedulePolicy } from '../types/scheduled-notification'; + +jest.mock('../utils/logger', () => ({ + __esModule: true, + default: { info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn() }, +})); + +describe('missed-schedule recovery', () => { + const testDbPath = './data/test-missed-schedule-recovery.db'; + let db: Database; + let repository: ScheduledNotificationRepository; + let api: NotificationAPI; + let scheduler: NotificationScheduler; + let sendEventMock: jest.Mock; + + beforeAll(async () => { + const dbDir = path.dirname(testDbPath); + if (!fs.existsSync(dbDir)) fs.mkdirSync(dbDir, { recursive: true }); + if (fs.existsSync(testDbPath)) fs.unlinkSync(testDbPath); + db = new Database(testDbPath); + await db.initialize(); + repository = new ScheduledNotificationRepository(db); + api = new NotificationAPI(repository); + }); + + afterAll(async () => { + await scheduler?.stop(); + await db.close(); + if (fs.existsSync(testDbPath)) fs.unlinkSync(testDbPath); + }); + + beforeEach(async () => { + resetWorkerManager(); + jest.clearAllMocks(); + await db.run('DELETE FROM notification_execution_log'); + await db.run('DELETE FROM scheduled_notifications'); + sendEventMock = jest.fn().mockResolvedValue(true); + }); + + afterEach(async () => { + await scheduler?.stop(); + }); + + function startScheduler(policy: MissedSchedulePolicy) { + const discordService = { + sendEventNotification: sendEventMock, + } as unknown as DiscordNotificationService; + scheduler = new NotificationScheduler( + repository, + { + enabled: true, + pollIntervalMs: 50, + lockTimeoutMs: 30_000, + batchSize: 10, + timingBufferMs: 0, + processorId: `missed-${policy}`, + missedSchedulePolicy: policy, + }, + discordService, + ); + return scheduler.start(); + } + + async function createOverdue(): Promise { + return repository.create({ + payload: { message: 'overdue reminder', event: { id: 'missed-1' }, contractConfig: { address: 'CABC' } }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'discord_channel_alpha', + executeAt: new Date(Date.now() - 60_000), + maxRetries: 1, + }); + } + + async function waitFor(predicate: () => Promise, timeoutMs = 1500): Promise { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + if (await predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 40)); + } + throw new Error('timed out waiting for scheduler'); + } + + it('delivers an overdue notification when the policy is execute', async () => { + const id = await createOverdue(); + await startScheduler('execute'); + + await waitFor(async () => (await repository.getById(id))?.status === NotificationStatus.COMPLETED); + expect(sendEventMock).toHaveBeenCalledTimes(1); + }); + + it('expires an overdue notification without delivering it', async () => { + const id = await createOverdue(); + await startScheduler('expire'); + + await waitFor(async () => (await repository.getById(id))?.status === NotificationStatus.EXPIRED); + expect(sendEventMock).not.toHaveBeenCalled(); + const row = await repository.getById(id); + expect(row?.lastError).toMatch(/expired by recovery policy/); + }); + + it('holds an overdue notification until an operator resumes it', async () => { + const id = await createOverdue(); + await startScheduler('intervene'); + + await waitFor(async () => { + const row = await repository.getById(id); + return row?.status === NotificationStatus.PENDING && row.nextRetryAt != null; + }); + expect(sendEventMock).not.toHaveBeenCalled(); + + await expect(api.releaseMissedSchedule(id)).resolves.toBe(true); + + await waitFor(async () => (await repository.getById(id))?.status === NotificationStatus.COMPLETED); + expect(sendEventMock).toHaveBeenCalledTimes(1); + }); +}); diff --git a/listener/src/api/event-stream.test.ts b/listener/src/api/event-stream.test.ts new file mode 100644 index 00000000..ae99bc2c --- /dev/null +++ b/listener/src/api/event-stream.test.ts @@ -0,0 +1,140 @@ +import http from 'http'; +import { DisplayEvent } from '../types/display-event'; +import { EventRegistry } from '../store/event-registry'; +import { handleEventStream, formatSseEvent } from './event-stream'; + +function event(eventId: string, ledger = 1): DisplayEvent { + return { + eventId, + contractAddress: 'CABC', + eventName: 'TaskCreated', + ledger, + type: 'contract', + topic: ['TaskCreated'], + value: '1', + txHash: `hash-${eventId}`, + receivedAt: ledger, + }; +} + +/** Minimal registry so this test does not load the Stellar SDK mock. */ +function fakeRegistry(initial: DisplayEvent[] = []) { + const events = [...initial]; + const listeners = new Set<(item: DisplayEvent) => void>(); + const registry = { + getEvents: () => events.slice(), + subscribe(listener: (item: DisplayEvent) => void) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + add(item: DisplayEvent) { + events.push(item); + for (const listener of listeners) listener(item); + }, + }; + return registry; +} + +function mockResponse() { + const chunks: string[] = []; + const listeners = new Map void>>(); + return { + statusCode: 0, + chunks, + setHeader: jest.fn(), + write: (chunk: string) => { + chunks.push(chunk); + return true; + }, + on: (name: string, cb: () => void) => { + const list = listeners.get(name) ?? []; + list.push(cb); + listeners.set(name, list); + }, + emit(name: string) { + for (const cb of listeners.get(name) ?? []) cb(); + }, + }; +} + +function mockRequest(headers: http.IncomingHttpHeaders = {}) { + const listeners = new Map void>>(); + return { + headers, + on: (name: string, cb: () => void) => { + const list = listeners.get(name) ?? []; + list.push(cb); + listeners.set(name, list); + }, + emit(name: string) { + for (const cb of listeners.get(name) ?? []) cb(); + }, + }; +} + +describe('event stream', () => { + beforeEach(() => { + jest.useFakeTimers(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('writes new events in the documented SSE format', () => { + const registry = fakeRegistry(); + const req = mockRequest(); + const res = mockResponse(); + + handleEventStream( + req as unknown as http.IncomingMessage, + res as unknown as http.ServerResponse, + registry as unknown as EventRegistry, + { heartbeatMs: 1000 }, + ); + + const stored = event('evt-1', 10); + registry.add(stored); + + expect(res.chunks.join('')).toBe(formatSseEvent(stored)); + expect(res.chunks.join('')).toContain('id: CABC:evt-1\n'); + expect(res.chunks.join('')).toContain('event: notification\n'); + expect(res.chunks.join('')).toContain('"eventId":"evt-1"'); + }); + + it('replays events after Last-Event-ID and skips the one already seen', () => { + const registry = fakeRegistry([event('evt-1', 1), event('evt-2', 2)]); + const req = mockRequest({ 'last-event-id': 'CABC:evt-1' }); + const res = mockResponse(); + + handleEventStream( + req as unknown as http.IncomingMessage, + res as unknown as http.ServerResponse, + registry as unknown as EventRegistry, + ); + + expect(res.chunks).toEqual([formatSseEvent(event('evt-2', 2))]); + }); + + it('sends keepalive comments and drops the subscriber when the client leaves', () => { + const registry = fakeRegistry(); + const req = mockRequest(); + const res = mockResponse(); + handleEventStream( + req as unknown as http.IncomingMessage, + res as unknown as http.ServerResponse, + registry as unknown as EventRegistry, + { heartbeatMs: 5000 }, + ); + + jest.advanceTimersByTime(5000); + expect(res.chunks).toContain(': keepalive\n\n'); + + req.emit('close'); + const before = res.chunks.length; + registry.add(event('evt-after-close')); + jest.advanceTimersByTime(5000); + + expect(res.chunks).toHaveLength(before); + }); +}); diff --git a/listener/src/api/event-stream.ts b/listener/src/api/event-stream.ts new file mode 100644 index 00000000..51d2e449 --- /dev/null +++ b/listener/src/api/event-stream.ts @@ -0,0 +1,115 @@ +import http from 'http'; +import { DisplayEvent } from '../types/display-event'; +import { EventRegistry } from '../store/event-registry'; + +/** + * Optional live event feed (Server-Sent Events). + * + * Enable with `EVENT_STREAM_ENABLED=true`. The endpoint is + * `GET /api/events/stream`. While it is disabled the path responds 404 and + * dashboards keep polling `GET /api/events`. + * + * Each newly stored registry event is one SSE record: + * + * ``` + * id: : + * event: notification + * data: + * + * ``` + * + * `id` is the registry key. Browsers resend it as the `Last-Event-ID` request + * header after a disconnect. The server replays registry events that arrived + * after that id, then continues with live events. A connection with no + * `Last-Event-ID` receives only events stored after it connects (clients load + * history from `GET /api/events`). An unknown id, for example after eviction, + * replays the current registry so the client can catch up. + * + * A `: keepalive` comment is written on a fixed interval so idle proxies do + * not close the socket. When the request or response closes, the subscriber + * is removed and the keepalive timer is cleared. + */ +export const EVENT_STREAM_PATH = '/api/events/stream'; + +export interface EventStreamOptions { + /** Milliseconds between `: keepalive` comments. Default 15000. */ + heartbeatMs?: number; +} + +const DEFAULT_HEARTBEAT_MS = 15_000; + +export function eventStreamId(event: DisplayEvent): string { + return `${event.contractAddress}:${event.eventId}`; +} + +export function formatSseEvent(event: DisplayEvent): string { + return `id: ${eventStreamId(event)}\nevent: notification\ndata: ${JSON.stringify(event)}\n\n`; +} + +/** + * Attach one SSE client to `registry`. Returns a function that drops the + * client; the HTTP close path calls it automatically. + */ +export function handleEventStream( + req: http.IncomingMessage, + res: http.ServerResponse, + registry: EventRegistry, + options: EventStreamOptions = {}, +): () => void { + const heartbeatMs = + options.heartbeatMs && options.heartbeatMs > 0 ? options.heartbeatMs : DEFAULT_HEARTBEAT_MS; + + res.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); + res.setHeader('Cache-Control', 'no-cache, no-transform'); + res.setHeader('Connection', 'keep-alive'); + res.setHeader('X-Accel-Buffering', 'no'); + res.statusCode = 200; + + const sent = new Set(); + let closed = false; + + const writeEvent = (event: DisplayEvent): void => { + if (closed) return; + const id = eventStreamId(event); + if (sent.has(id)) return; + sent.add(id); + res.write(formatSseEvent(event)); + }; + + const unsubscribe = registry.subscribe(writeEvent); + + const lastEventId = headerValue(req.headers['last-event-id']); + for (const event of eventsAfter(registry, lastEventId)) { + writeEvent(event); + } + + const heartbeat = setInterval(() => { + if (!closed) res.write(': keepalive\n\n'); + }, heartbeatMs); + if (typeof heartbeat.unref === 'function') heartbeat.unref(); + + const cleanup = (): void => { + if (closed) return; + closed = true; + clearInterval(heartbeat); + unsubscribe(); + }; + + req.on('close', cleanup); + res.on('close', cleanup); + + return cleanup; +} + +function headerValue(value: string | string[] | undefined): string | undefined { + if (Array.isArray(value)) return value[0]; + return value && value.length > 0 ? value : undefined; +} + +function eventsAfter(registry: EventRegistry, lastEventId: string | undefined): DisplayEvent[] { + if (!lastEventId) return []; + const events = registry.getEvents(); + const index = events.findIndex((event) => eventStreamId(event) === lastEventId); + if (index === -1) return events; + return events.slice(index + 1); +} diff --git a/listener/src/api/events-server.ts b/listener/src/api/events-server.ts index b9475d07..f06428e7 100644 --- a/listener/src/api/events-server.ts +++ b/listener/src/api/events-server.ts @@ -15,9 +15,6 @@ import { sendOk, sendErr, sendJson, ErrorCode } from '../utils/response'; import { normalizePaginationParams } from '../utils/pagination'; import { handleApiError, ApiError } from './error-handler'; import { applyRequestIdMiddleware } from '../middleware/request-id'; -import { TemplateService } from '../services/template-service'; -import { handleTemplateRoutes } from './template-routes'; -import { sendOk, sendErr, sendJson, ErrorCode } from '../utils/response'; import { validateContentType, getMimeType } from '../middleware/content-type'; import { NotificationHistoryService } from '../services/notification-history'; import { SearchSuggestionService } from '../services/search-suggestion'; @@ -28,7 +25,6 @@ import { IdempotencyKeyReuseError, } from '../services/idempotency-key-service'; import { WebhookReplayCache } from '../services/webhook-replay-cache'; -import { IdempotencyKeyService, IdempotencyKeyReuseError } from '../services/idempotency-key-service'; import { WebhookSecret, RateLimitConfig, ContractConfig } from '../types'; import { RateLimiter } from './rate-limiter'; import { @@ -71,6 +67,8 @@ import { sanitizeCredentials } from '../utils/credential-sanitizer'; import { DiscordNotificationProvider } from '../services/providers/discord-provider'; import { WebhookNotificationProvider } from '../services/providers/webhook-provider'; import { ProviderHealthResult } from '../types/provider-capabilities'; +import { EVENT_STREAM_PATH, handleEventStream } from './event-stream'; +import { resolveLocalDateTime } from '../services/local-schedule'; export interface EventsServerOptions { port: number; @@ -136,6 +134,11 @@ export interface EventsServerOptions { maxBodyBytes?: number; /** Optional DataExportService override for administrative exports (#850). */ dataExportService?: DataExportService | null; + /** + * Optional SSE feed at GET /api/events/stream. Omitted or disabled leaves + * that path as 404 so existing poll clients are unchanged. + */ + eventStream?: { enabled: boolean; heartbeatMs?: number } | null; } type ServiceStatus = 'ok' | 'error' | 'not_configured'; @@ -503,7 +506,7 @@ export function createEventsServer(options: EventsServerOptions): http.Server { res.setHeader('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS'); res.setHeader( 'Access-Control-Allow-Headers', - 'Content-Type, X-API-Key, Authorization, X-Correlation-Id, X-Request-Id', + 'Content-Type, X-API-Key, Authorization, X-Correlation-Id, X-Request-Id, Last-Event-ID', ); res.setHeader( 'Access-Control-Expose-Headers', @@ -1175,24 +1178,79 @@ export function createEventsServer(options: EventsServerOptions): http.Server { } try { - if (!data.executeAt || !data.payload || !data.targetRecipient) { + const hasLocalSchedule = data.timeZone != null || data.localDateTime != null; + if ((!data.executeAt && !hasLocalSchedule) || !data.payload || !data.targetRecipient) { res.writeHead(400, { 'Content-Type': 'application/json' }); res.end( JSON.stringify({ - error: 'Missing required fields: executeAt, payload, targetRecipient', + error: 'Missing required fields: executeAt (or timeZone + localDateTime), payload, targetRecipient', code: 'MISSING_FIELDS', }), ); return; } - const executeAt = new Date(data.executeAt); - if (isNaN(executeAt.getTime())) { - res.writeHead(400, { 'Content-Type': 'application/json' }); - res.end( - JSON.stringify({ error: 'executeAt is not a valid date', code: 'INVALID_DATE' }), - ); - return; + let executeAt: Date; + let metadata = data.metadata; + if (data.timeZone != null || data.localDateTime != null) { + if (typeof data.timeZone !== 'string' || typeof data.localDateTime !== 'string') { + res.writeHead(400, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ + error: 'timeZone and localDateTime are both required for a local schedule', + code: 'INVALID_LOCAL_SCHEDULE', + }), + ); + return; + } + if ( + data.recurrence != null && + data.recurrence !== 'daily' && + data.recurrence !== 'weekly' + ) { + res.writeHead(400, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ + error: 'recurrence must be daily or weekly', + code: 'INVALID_LOCAL_SCHEDULE', + }), + ); + return; + } + try { + const resolved = resolveLocalDateTime(data.localDateTime, data.timeZone); + executeAt = resolved.executeAt; + const base = + metadata && typeof metadata === 'object' && !Array.isArray(metadata) ? metadata : {}; + metadata = { + ...base, + localSchedule: { + timeZone: data.timeZone, + localDateTime: resolved.requestedLocal, + recurrence: data.recurrence, + kind: resolved.kind, + resolvedLocal: resolved.resolvedLocal, + }, + }; + } catch (scheduleError) { + res.writeHead(400, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ + error: (scheduleError as Error).message, + code: 'INVALID_LOCAL_SCHEDULE', + }), + ); + return; + } + } else { + executeAt = new Date(data.executeAt); + if (isNaN(executeAt.getTime())) { + res.writeHead(400, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ error: 'executeAt is not a valid date', code: 'INVALID_DATE' }), + ); + return; + } } const schedule = async (): Promise<{ id: number }> => { @@ -1206,7 +1264,7 @@ export function createEventsServer(options: EventsServerOptions): http.Server { priority: data.priority, eventId: data.eventId, contractAddress: data.contractAddress, - metadata: data.metadata, + metadata, }); logger.info('Notification scheduled via API', { requestId, @@ -1444,6 +1502,31 @@ export function createEventsServer(options: EventsServerOptions): http.Server { return; } + // POST /api/schedule/:id/resume — release a notification held by the + // intervene missed-schedule policy so the next poll delivers it. + const resumeMatch = url.pathname.match(/^\/api\/schedule\/(\d+)\/resume$/); + if (req.method === 'POST' && resumeMatch) { + if (!options.notificationAPI) { + sendErr(res, 503, 'Scheduler not enabled', ErrorCode.SERVICE_UNAVAILABLE); + return; + } + const id = parseInt(resumeMatch[1], 10); + options.notificationAPI + .releaseMissedSchedule(id) + .then((released) => { + if (!released) { + sendErr(res, 404, 'No notification is waiting for intervention', ErrorCode.NOT_FOUND); + return; + } + sendOk(res, 200, { id, released: true }); + }) + .catch((error) => { + logger.error('Failed to resume missed notification', { error, requestId, correlationId, id }); + sendErr(res, 500, (error as Error).message, ErrorCode.INTERNAL_ERROR); + }); + return; + } + // GET /api/schedule/:id if (req.method === 'GET' && url.pathname.startsWith('/api/schedule/')) { if (!options.notificationAPI) { diff --git a/listener/src/config-schema.ts b/listener/src/config-schema.ts index 4fe94759..14f6307c 100644 --- a/listener/src/config-schema.ts +++ b/listener/src/config-schema.ts @@ -225,6 +225,7 @@ export const APP_CONFIG_SCHEMA: ConfigSchema = { batchSize: { type: 'number', min: 1 }, timingBufferMs: { type: 'number', min: 0 }, concurrency: { type: 'number', min: 1 }, + missedSchedulePolicy: { type: 'string', enum: ['execute', 'expire', 'intervene'] }, }, retryScheduler: { enabled: { type: 'boolean' }, diff --git a/listener/src/config.ts b/listener/src/config.ts index ecd9c38a..a6a3e7ab 100644 --- a/listener/src/config.ts +++ b/listener/src/config.ts @@ -21,6 +21,14 @@ export class ConfigError extends Error { } } +function parseMissedSchedulePolicy( + value: string | undefined, +): 'execute' | 'expire' | 'intervene' { + if (value === undefined || value === '' || value === 'execute') return 'execute'; + if (value === 'expire' || value === 'intervene') return value; + return 'execute'; +} + function trimEnv(name: string): string | undefined { const value = process.env[name]; return value === undefined ? undefined : value.trim(); @@ -472,6 +480,7 @@ export function loadConfig(): Config { batchSize: parseIntegerEnv('SCHEDULER_BATCH_SIZE', '10'), concurrency: parseIntegerEnv('WORKER_CONCURRENCY', '1'), timingBufferMs: parseIntegerEnv('SCHEDULER_TIMING_BUFFER_MS', '60000'), + missedSchedulePolicy: parseMissedSchedulePolicy(trimEnv('SCHEDULER_MISSED_POLICY')), }, retryScheduler: loadRetrySchedulerConfig(retryPolicy), retryPolicy, @@ -746,6 +755,18 @@ export function validateConfig(config: Config): void { if (config.scheduler.batchSize < 1) { errors.push(`SCHEDULER_BATCH_SIZE must be >= 1 (received: ${config.scheduler.batchSize}).`); } + const missedPolicy = trimEnv('SCHEDULER_MISSED_POLICY'); + if ( + missedPolicy !== undefined && + missedPolicy !== '' && + missedPolicy !== 'execute' && + missedPolicy !== 'expire' && + missedPolicy !== 'intervene' + ) { + errors.push( + `SCHEDULER_MISSED_POLICY must be execute, expire, or intervene (received: ${missedPolicy}).`, + ); + } } // ── Retry scheduler ──────────────────────────────────────────────────────── diff --git a/listener/src/index.ts b/listener/src/index.ts index 69ec3f0c..8e34360b 100644 --- a/listener/src/index.ts +++ b/listener/src/index.ts @@ -186,6 +186,10 @@ async function main() { metricsStore, healthMonitor, deliveryReceiptRepository, + eventStream: { + enabled: process.env.EVENT_STREAM_ENABLED === 'true', + heartbeatMs: Number(process.env.EVENT_STREAM_HEARTBEAT_MS || 15_000), + }, }); if (healthMonitor) { diff --git a/listener/src/services/local-schedule.test.ts b/listener/src/services/local-schedule.test.ts new file mode 100644 index 00000000..04e261de --- /dev/null +++ b/listener/src/services/local-schedule.test.ts @@ -0,0 +1,74 @@ +import { nextLocalOccurrence, resolveLocalDateTime } from './local-schedule'; + +describe('daylight-saving schedule policy', () => { + it('shifts a skipped America/New_York spring-forward time forward by the gap', () => { + const resolved = resolveLocalDateTime('2026-03-08T02:30:00', 'America/New_York'); + + expect(resolved.kind).toBe('skipped'); + expect(resolved.resolvedLocal).toBe('2026-03-08T03:30:00'); + expect(resolved.executeAt.toISOString()).toBe('2026-03-08T07:30:00.000Z'); + }); + + it('keeps the earlier instant when America/New_York repeats a local time', () => { + const resolved = resolveLocalDateTime('2026-11-01T01:30:00', 'America/New_York'); + + expect(resolved.kind).toBe('repeated'); + expect(resolved.resolvedLocal).toBe('2026-11-01T01:30:00'); + expect(resolved.executeAt.toISOString()).toBe('2026-11-01T05:30:00.000Z'); + }); + + it('resolves unique local times on either side of the New York spring transition', () => { + expect(resolveLocalDateTime('2026-03-08T01:30:00', 'America/New_York').executeAt.toISOString()).toBe( + '2026-03-08T06:30:00.000Z', + ); + expect(resolveLocalDateTime('2026-03-08T03:30:00', 'America/New_York').executeAt.toISOString()).toBe( + '2026-03-08T07:30:00.000Z', + ); + }); + + it('shifts a 30-minute Lord Howe spring gap by 30 minutes and keeps the earlier fall-back instant', () => { + const skipped = resolveLocalDateTime('2026-10-04T02:15:00', 'Australia/Lord_Howe'); + expect(skipped.kind).toBe('skipped'); + expect(skipped.resolvedLocal).toBe('2026-10-04T02:45:00'); + expect(skipped.executeAt.toISOString()).toBe('2026-10-03T15:45:00.000Z'); + + const repeated = resolveLocalDateTime('2026-04-05T01:45:00', 'Australia/Lord_Howe'); + expect(repeated.kind).toBe('repeated'); + expect(repeated.resolvedLocal).toBe('2026-04-05T01:45:00'); + expect(repeated.executeAt.toISOString()).toBe('2026-04-04T14:45:00.000Z'); + }); + + it('applies the same policy outside the US', () => { + const skipped = resolveLocalDateTime('2026-03-29T01:30:00', 'Europe/London'); + expect(skipped.kind).toBe('skipped'); + expect(skipped.resolvedLocal).toBe('2026-03-29T02:30:00'); + expect(skipped.executeAt.toISOString()).toBe('2026-03-29T01:30:00.000Z'); + + const repeated = resolveLocalDateTime('2026-10-25T01:30:00', 'Europe/London'); + expect(repeated.kind).toBe('repeated'); + expect(repeated.executeAt.toISOString()).toBe('2026-10-25T00:30:00.000Z'); + }); + + it('advances a daily local schedule across the spring gap without staying shifted', () => { + const transition = nextLocalOccurrence({ + timeZone: 'America/New_York', + localDateTime: '2026-03-07T02:30:00', + recurrence: 'daily', + }); + expect(transition.kind).toBe('skipped'); + expect(transition.resolvedLocal).toBe('2026-03-08T03:30:00'); + + const dayAfter = nextLocalOccurrence({ + timeZone: 'America/New_York', + localDateTime: '2026-03-08T02:30:00', + recurrence: 'daily', + }); + expect(dayAfter.kind).toBe('unique'); + expect(dayAfter.requestedLocal).toBe('2026-03-09T02:30:00'); + expect(dayAfter.executeAt.toISOString()).toBe('2026-03-09T06:30:00.000Z'); + }); + + it('rejects an unknown timezone', () => { + expect(() => resolveLocalDateTime('2026-03-08T02:30:00', 'Not/AZone')).toThrow(/IANA timezone/); + }); +}); diff --git a/listener/src/services/local-schedule.ts b/listener/src/services/local-schedule.ts new file mode 100644 index 00000000..e7d20040 --- /dev/null +++ b/listener/src/services/local-schedule.ts @@ -0,0 +1,260 @@ +/** + * Daylight-saving policy for timezone-based notification schedules. + * + * A local wall-clock time is not always one UTC instant: + * + * - **Repeated local time** (fall back). The clock is set back and the same + * reading happens twice. NotifyChain schedules the **earlier** instant, the + * first time that reading occurs. Example: `America/New_York` + * `2026-11-01T01:30:00` runs at `2026-11-01T05:30:00.000Z` (EDT), not an + * hour later when the clock shows 01:30 again (EST). + * - **Skipped local time** (spring forward). The clock jumps ahead and the + * reading never happens. NotifyChain moves the wall time forward by the + * length of that zone's gap, then schedules that valid instant. A one-hour + * gap and a 30-minute gap are not treated the same. Examples: + * `America/New_York` `2026-03-08T02:30:00` does not exist; it runs at + * `03:30` EDT (`2026-03-08T07:30:00.000Z`). `Australia/Lord_Howe` + * `2026-10-04T02:15:00` does not exist; the gap is 30 minutes, so it runs + * at `02:45` (`2026-10-03T15:45:00.000Z`). The occurrence is not dropped + * and is not placed inside the gap. + * - **Unique local time.** The reading maps to exactly one UTC instant. + * + * Recurrence (`daily` or `weekly`) advances the **requested** local calendar + * date and applies this same policy. A daily 02:30 that was shifted to 03:30 + * on the transition day is 02:30 again on the following local date. + */ + +export type LocalResolutionKind = 'unique' | 'skipped' | 'repeated'; +export type ScheduleRecurrence = 'daily' | 'weekly'; + +export interface LocalScheduleIntent { + timeZone: string; + /** Requested wall time, `YYYY-MM-DDTHH:mm:ss`, with no offset. */ + localDateTime: string; + recurrence?: ScheduleRecurrence; +} + +export interface LocalResolution { + kind: LocalResolutionKind; + executeAt: Date; + /** Wall time that will actually occur. */ + resolvedLocal: string; + /** Wall time that was requested, including a skipped reading. */ + requestedLocal: string; + timeZone: string; +} + +interface LocalParts { + year: number; + month: number; + day: number; + hour: number; + minute: number; + second: number; +} + +const LOCAL_DATE_TIME = /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2})(?::(\d{2}))?$/; + +export function resolveLocalDateTime(localDateTime: string, timeZone: string): LocalResolution { + assertTimeZone(timeZone); + const parts = parseLocal(localDateTime); + const requested = formatParts(parts); + const matches = findMatches(parts, timeZone); + + if (matches.length >= 2) { + return resolution('repeated', matches[0], requested, requested, timeZone); + } + if (matches.length === 1) { + return resolution('unique', matches[0], requested, requested, timeZone); + } + + const shifted = addMinutes(parts, transitionGapMinutes(parts, timeZone)); + const shiftedMatches = findMatches(shifted, timeZone); + if (shiftedMatches.length === 0) { + throw new Error(`Unable to resolve skipped local time ${requested} in ${timeZone}`); + } + return resolution('skipped', shiftedMatches[0], formatParts(shifted), requested, timeZone); +} + +/** Next daily or weekly occurrence of a requested local schedule. */ +export function nextLocalOccurrence(intent: LocalScheduleIntent): LocalResolution { + if (intent.recurrence !== 'daily' && intent.recurrence !== 'weekly') { + throw new Error('recurrence must be daily or weekly'); + } + assertTimeZone(intent.timeZone); + const parts = parseLocal(intent.localDateTime); + const advanced = addDays(parts, intent.recurrence === 'weekly' ? 7 : 1); + return resolveLocalDateTime(formatParts(advanced), intent.timeZone); +} + +export function readLocalSchedule(metadata: string | null | undefined): LocalScheduleIntent | null { + if (!metadata) return null; + try { + const parsed = JSON.parse(metadata) as { localSchedule?: Partial }; + const schedule = parsed.localSchedule; + if (!schedule?.timeZone || !schedule.localDateTime) return null; + if ( + schedule.recurrence !== undefined && + schedule.recurrence !== 'daily' && + schedule.recurrence !== 'weekly' + ) { + return null; + } + return { + timeZone: schedule.timeZone, + localDateTime: schedule.localDateTime, + recurrence: schedule.recurrence, + }; + } catch { + return null; + } +} + +function resolution( + kind: LocalResolutionKind, + executeAt: Date, + resolvedLocal: string, + requestedLocal: string, + timeZone: string, +): LocalResolution { + return { kind, executeAt, resolvedLocal, requestedLocal, timeZone }; +} + +function assertTimeZone(timeZone: string): void { + if (!timeZone || typeof timeZone !== 'string') { + throw new Error('timeZone is required'); + } + try { + Intl.DateTimeFormat('en-US', { timeZone }).format(0); + } catch { + throw new Error(`timeZone is not a valid IANA timezone: ${timeZone}`); + } +} + +function parseLocal(value: string): LocalParts { + const match = LOCAL_DATE_TIME.exec(value.trim()); + if (!match) { + throw new Error('localDateTime must be YYYY-MM-DDTHH:mm:ss without a timezone offset'); + } + const parts: LocalParts = { + year: Number(match[1]), + month: Number(match[2]), + day: Number(match[3]), + hour: Number(match[4]), + minute: Number(match[5]), + second: match[6] ? Number(match[6]) : 0, + }; + if (parts.hour > 23 || parts.minute > 59 || parts.second > 59) { + throw new Error('localDateTime is not a real calendar time'); + } + const probe = new Date(Date.UTC(parts.year, parts.month - 1, parts.day)); + if ( + probe.getUTCFullYear() !== parts.year || + probe.getUTCMonth() !== parts.month - 1 || + probe.getUTCDate() !== parts.day + ) { + throw new Error('localDateTime is not a real calendar date'); + } + return parts; +} + +function formatParts(parts: LocalParts): string { + const pad = (value: number, width = 2) => String(value).padStart(width, '0'); + return `${pad(parts.year, 4)}-${pad(parts.month)}-${pad(parts.day)}T${pad(parts.hour)}:${pad(parts.minute)}:${pad(parts.second)}`; +} + +function offsetMinutes(instant: Date, timeZone: string): number { + const parts = zonedParts(instant, timeZone); + const asUtc = Date.UTC(parts.year, parts.month - 1, parts.day, parts.hour, parts.minute, parts.second); + return Math.round((asUtc - instant.getTime()) / 60_000); +} + +function formatLocal(instant: Date, timeZone: string): string { + return formatParts(zonedParts(instant, timeZone)); +} + +function zonedParts(instant: Date, timeZone: string): LocalParts { + const formatter = new Intl.DateTimeFormat('en-US', { + timeZone, + hourCycle: 'h23', + year: 'numeric', + month: '2-digit', + day: '2-digit', + hour: '2-digit', + minute: '2-digit', + second: '2-digit', + }); + const mapped: Record = {}; + for (const part of formatter.formatToParts(instant)) { + if (part.type !== 'literal') mapped[part.type] = part.value; + } + let hour = Number(mapped.hour); + if (hour === 24) hour = 0; + return { + year: Number(mapped.year), + month: Number(mapped.month), + day: Number(mapped.day), + hour, + minute: Number(mapped.minute), + second: Number(mapped.second), + }; +} + +function findMatches(parts: LocalParts, timeZone: string): Date[] { + const naive = Date.UTC(parts.year, parts.month - 1, parts.day, parts.hour, parts.minute, parts.second); + const requested = formatParts(parts); + const offsets = new Set(); + // A local reading can sit many hours away from its UTC instant. Sample the + // whole practical offset range so both sides of a short overlap are seen. + for (let minutes = -16 * 60; minutes <= 16 * 60; minutes += 30) { + offsets.add(offsetMinutes(new Date(naive + minutes * 60_000), timeZone)); + } + const seen = new Set(); + const matches: Date[] = []; + for (const offset of offsets) { + const candidate = new Date(naive - offset * 60_000); + if (seen.has(candidate.getTime())) continue; + if (formatLocal(candidate, timeZone) === requested) { + seen.add(candidate.getTime()); + matches.push(candidate); + } + } + matches.sort((left, right) => left.getTime() - right.getTime()); + return matches; +} + +function transitionGapMinutes(parts: LocalParts, timeZone: string): number { + const naive = Date.UTC(parts.year, parts.month - 1, parts.day, parts.hour, parts.minute, parts.second); + // The transition instant is the UTC time of the wall clock, not the naive + // reading of that wall clock. Searching only a few hours around the naive + // UTC timestamp misses zones far from UTC and then assumes a 60-minute gap. + const start = naive - 20 * 3_600_000; + let previous = offsetMinutes(new Date(start), timeZone); + for (let minutes = 15; minutes <= 40 * 60; minutes += 15) { + const offset = offsetMinutes(new Date(start + minutes * 60_000), timeZone); + if (offset !== previous) return Math.abs(offset - previous); + } + return 60; +} + +function addMinutes(parts: LocalParts, minutes: number): LocalParts { + const utc = Date.UTC(parts.year, parts.month - 1, parts.day, parts.hour, parts.minute, parts.second); + return partsFromUtc(utc + minutes * 60_000); +} + +function addDays(parts: LocalParts, days: number): LocalParts { + const utc = Date.UTC(parts.year, parts.month - 1, parts.day + days, parts.hour, parts.minute, parts.second); + return partsFromUtc(utc); +} + +function partsFromUtc(utc: number): LocalParts { + const date = new Date(utc); + return { + year: date.getUTCFullYear(), + month: date.getUTCMonth() + 1, + day: date.getUTCDate(), + hour: date.getUTCHours(), + minute: date.getUTCMinutes(), + second: date.getUTCSeconds(), + }; +} diff --git a/listener/src/services/notification-api.ts b/listener/src/services/notification-api.ts index 8f69bc9e..7b5a31dc 100644 --- a/listener/src/services/notification-api.ts +++ b/listener/src/services/notification-api.ts @@ -258,6 +258,14 @@ export class NotificationAPI { return await this.repository.getById(id); } + /** + * Release a notification held because it missed its schedule under the + * `intervene` recovery policy. The next scheduler poll delivers it. + */ + async releaseMissedSchedule(id: number): Promise { + return await this.repository.releaseIntervention(id); + } + /** * Get scheduler statistics. */ diff --git a/listener/src/services/notification-scheduler.ts b/listener/src/services/notification-scheduler.ts index 1a224e65..09306a27 100644 --- a/listener/src/services/notification-scheduler.ts +++ b/listener/src/services/notification-scheduler.ts @@ -13,6 +13,7 @@ import { verifyPayloadIntegrity } from '../utils/payload-integrity'; import { DeliveryReceiptRepository } from './delivery-receipt-repository'; import { DeliveryResult } from '../types/provider-capabilities'; import { NotificationClaimLease, startClaimLease } from './notification-claim-lease'; +import { nextLocalOccurrence, readLocalSchedule } from './local-schedule'; /** * Background scheduler that processes scheduled notifications @@ -346,7 +347,50 @@ export class NotificationScheduler { return; } - if (timeDiff > this.config.timingBufferMs) { + if (timeDiff > this.config.timingBufferMs && !this.isInterventionReleased(notification)) { + const policy = this.config.missedSchedulePolicy ?? 'execute'; + if (policy === 'expire') { + const errorMessage = 'Missed schedule expired by recovery policy'; + await this.repository.markAsExpired(notification.id!, errorMessage); + await this.repository.logExecution({ + scheduledNotificationId: notification.id!, + executionAttempt, + executionTime: new Date(), + status: 'FAILED', + errorMessage, + durationMs: Date.now() - startTime, + }); + if (jobId) { + jobMonitor.failJob(jobId, errorMessage, { notificationId: notification.id }); + } + logger.info('Overdue notification expired by missed-schedule policy', { + requestId, + id: notification.id, + missedByMs: timeDiff, + }); + return; + } + if (policy === 'intervene') { + const errorMessage = 'Missed schedule requires operator intervention'; + await this.repository.holdForIntervention(notification.id!); + await this.repository.logExecution({ + scheduledNotificationId: notification.id!, + executionAttempt, + executionTime: new Date(), + status: 'FAILED', + errorMessage, + durationMs: Date.now() - startTime, + }); + if (jobId) { + jobMonitor.failJob(jobId, errorMessage, { notificationId: notification.id }); + } + logger.warn('Overdue notification held for operator intervention', { + requestId, + id: notification.id, + missedByMs: timeDiff, + }); + return; + } logger.warn('Missed scheduled notification detected; dispatching catch-up delivery', { requestId, id: notification.id, @@ -395,6 +439,7 @@ export class NotificationScheduler { if (success) { await this.repository.markAsCompleted(notification.id!, requestId); + await this.enqueueNextLocalOccurrence(notification, requestId); await this.repository.logExecution({ scheduledNotificationId: notification.id!, executionAttempt, @@ -477,6 +522,61 @@ export class NotificationScheduler { } } + private isInterventionReleased(notification: ScheduledNotification): boolean { + if (!notification.errorDetails) return false; + try { + const parsed = JSON.parse(notification.errorDetails) as { code?: string; releasedAt?: string }; + return parsed.code === 'MISSED_SCHEDULE_INTERVENTION' && typeof parsed.releasedAt === 'string'; + } catch { + return false; + } + } + + /** + * When a timezone schedule repeats daily or weekly, queue the next local + * occurrence with the daylight-saving policy in local-schedule.ts. + */ + private async enqueueNextLocalOccurrence( + notification: ScheduledNotification, + requestId: string, + ): Promise { + const intent = readLocalSchedule(notification.metadata); + if (!intent?.recurrence) return; + + try { + const next = nextLocalOccurrence(intent); + const metadata = notification.metadata ? JSON.parse(notification.metadata) : {}; + metadata.localSchedule = { + ...(metadata.localSchedule ?? {}), + timeZone: intent.timeZone, + localDateTime: next.requestedLocal, + recurrence: intent.recurrence, + kind: next.kind, + resolvedLocal: next.resolvedLocal, + }; + await this.repository.create( + { + payload: JSON.parse(notification.payload), + notificationType: notification.notificationType, + targetRecipient: notification.targetRecipient, + executeAt: next.executeAt, + maxRetries: notification.maxRetries, + priority: notification.priority, + eventId: notification.eventId ?? undefined, + contractAddress: notification.contractAddress ?? undefined, + metadata, + }, + requestId, + ); + } catch (error) { + logger.error('Failed to enqueue the next local occurrence', { + requestId, + id: notification.id, + error, + }); + } + } + private async expireIfPastDeadline( notification: ScheduledNotification, requestId: string, diff --git a/listener/src/services/scheduled-notification-repository.ts b/listener/src/services/scheduled-notification-repository.ts index dbf586d5..1fd5f388 100644 --- a/listener/src/services/scheduled-notification-repository.ts +++ b/listener/src/services/scheduled-notification-repository.ts @@ -361,9 +361,9 @@ export class ScheduledNotificationRepository { } /** Mark a notification expired and release its processing lock. */ - async markAsExpired(id: number): Promise { + async markAsExpired(id: number, reason = 'Notification expired before delivery'): Promise { const now = new Date().toISOString(); - const errorMessage = 'Notification expired before delivery'; + const errorMessage = reason; await this.db.run( `UPDATE scheduled_notifications SET status = ?, last_error = ?, error_details = ?, processing_completed_at = ?, @@ -382,6 +382,60 @@ export class ScheduledNotificationRepository { this.statsCache.invalidate(); } + /** + * Hold an overdue notification until an operator resumes it. + * `next_retry_at` is parked far in the future so the poll loop will not + * claim the row again. Status stays PENDING so no schema migration is required. + */ + async holdForIntervention(id: number): Promise { + const now = new Date().toISOString(); + const errorMessage = 'Missed schedule requires operator intervention'; + await this.db.run( + `UPDATE scheduled_notifications + SET status = ?, last_error = ?, error_details = ?, next_retry_at = ?, + updated_at = ?, processor_id = NULL, lock_expires_at = NULL + WHERE id = ? AND status = ?`, + [ + NotificationStatus.PENDING, + errorMessage, + JSON.stringify({ code: 'MISSED_SCHEDULE_INTERVENTION', heldAt: now }), + '9999-12-31T00:00:00.000Z', + now, + id, + NotificationStatus.PROCESSING, + ], + ); + this.statsCache.invalidate(); + } + + /** + * Let an operator release a notification held by the intervene policy. + * The next poll delivers it. Returns false when the row is not held. + */ + async releaseIntervention(id: number): Promise { + const existing = await this.getById(id); + if (!existing || existing.status !== NotificationStatus.PENDING) return false; + + let details: { code?: string; releasedAt?: string; heldAt?: string }; + try { + details = JSON.parse(existing.errorDetails ?? ''); + } catch { + return false; + } + if (details.code !== 'MISSED_SCHEDULE_INTERVENTION' || details.releasedAt) return false; + + const now = new Date().toISOString(); + details.releasedAt = now; + const result = await this.db.run( + `UPDATE scheduled_notifications + SET next_retry_at = NULL, error_details = ?, updated_at = ? + WHERE id = ? AND status = ?`, + [JSON.stringify(details), now, id, NotificationStatus.PENDING], + ); + this.statsCache.invalidate(); + return result.changes > 0; + } + /** * Mark notification as failed or retry (sets next_retry_at for backoff scheduling) */ diff --git a/listener/src/store/event-registry.ts b/listener/src/store/event-registry.ts index 44b66161..a9fdd14e 100644 --- a/listener/src/store/event-registry.ts +++ b/listener/src/store/event-registry.ts @@ -6,8 +6,11 @@ import logger from '../utils/logger'; const DEFAULT_MAX_EVENTS = 10000; const DEFAULT_TTL_MS = 24 * 60 * 60 * 1000; // 24 hours +export type RegistryListener = (event: DisplayEvent) => void; + export class EventRegistry { private readonly events = new Map(); + private readonly listeners = new Set(); private readonly maxEvents: number; private ttlMs: number; private cleanupTimer: ReturnType | null = null; @@ -81,6 +84,7 @@ export class EventRegistry { }; this.events.set(eventKey, displayEvent); + this.notify(displayEvent); this.lastIngestedLedger = displayEvent.ledger; this.lastIngestedAt = displayEvent.receivedAt; this.maxLedgerSeen = @@ -119,6 +123,17 @@ export class EventRegistry { ); } + /** + * Subscribe to events stored after this call. Duplicate inputs that return + * the existing record do not notify listeners. Returns an unsubscribe function. + */ + subscribe(listener: RegistryListener): () => void { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + count(): number { return this.events.size; } @@ -153,6 +168,12 @@ export class EventRegistry { private getEventKey(eventId: string, contractAddress: string): string { return `${contractAddress}:${eventId}`; } + + private notify(event: DisplayEvent): void { + for (const listener of this.listeners) { + listener(event); + } + } } export const eventRegistry = new EventRegistry(); diff --git a/listener/src/types/index.ts b/listener/src/types/index.ts index 24d41547..38890585 100644 --- a/listener/src/types/index.ts +++ b/listener/src/types/index.ts @@ -125,6 +125,8 @@ export interface SchedulerConfig { batchSize: number; concurrency: number; timingBufferMs: number; + /** `execute` (default), `expire`, or `intervene`. See SCHEDULER_MISSED_POLICY. */ + missedSchedulePolicy?: 'execute' | 'expire' | 'intervene'; } export interface EventQueueConfig { diff --git a/listener/src/types/scheduled-notification.ts b/listener/src/types/scheduled-notification.ts index 64b8bc54..ee6dfd7a 100644 --- a/listener/src/types/scheduled-notification.ts +++ b/listener/src/types/scheduled-notification.ts @@ -102,6 +102,16 @@ export interface NotificationExecutionLog { durationMs?: number | null; } +/** + * What to do when a notification is still pending after its executeAt and the + * scheduler was not available to run it inside the timing buffer. + * + * - `execute` — deliver it late (default, previous catch-up behavior) + * - `expire` — mark it EXPIRED and do not deliver + * - `intervene` — hold it until an operator resumes it + */ +export type MissedSchedulePolicy = 'execute' | 'expire' | 'intervene'; + export interface SchedulerConfig { enabled: boolean; pollIntervalMs: number; @@ -112,6 +122,8 @@ export interface SchedulerConfig { retryDelayMs?: number; /** Max notifications processed concurrently per poll cycle (WORKER_CONCURRENCY). Default 1 = serial. */ concurrency?: number; + /** Overdue notifications outside the timing buffer. Default `execute`. */ + missedSchedulePolicy?: MissedSchedulePolicy; } export interface DeadLetterQueueEntry {