diff --git a/contract/contracts/hello-world/src/base/events.rs b/contract/contracts/hello-world/src/base/events.rs index cd207d9..193fc4f 100644 --- a/contract/contracts/hello-world/src/base/events.rs +++ b/contract/contracts/hello-world/src/base/events.rs @@ -117,6 +117,28 @@ pub struct AutoshareUpdated { pub id: BytesN<32>, } +/// Emitted when a recipient updates a delivery channel preference. +/// +/// Off-chain consumers can filter on `recipient` and inspect `channel` / +/// `enabled` in the event data to react to channel configuration changes +/// without decoding full preference storage. +#[contractevent] +#[derive(Clone)] +pub struct ChannelPreferenceUpdated { + #[topic] + pub recipient: Address, + #[topic] + pub category: NotificationCategory, + #[topic] + pub priority: NotificationPriority, + /// Delivery channel that changed (0 = Wallet, 1 = Email, 2 = InApp). + pub channel: u32, + /// Whether the channel is now enabled. + pub enabled: bool, + /// Ledger timestamp when the preference was updated. + pub updated_at: u64, +} + /// Emitted when an AutoShare group is deactivated by its creator. #[contractevent(data_format = "single-value")] #[derive(Clone)] diff --git a/contract/contracts/hello-world/src/preferences_logic.rs b/contract/contracts/hello-world/src/preferences_logic.rs index ea7e84d..41b8bcc 100644 --- a/contract/contracts/hello-world/src/preferences_logic.rs +++ b/contract/contracts/hello-world/src/preferences_logic.rs @@ -3,6 +3,9 @@ /// Provides get/set/reset operations for per-user notification preferences. /// Preferences are stored in persistent storage keyed by recipient address. use crate::base::errors::Error; +use crate::base::events::{ + ChannelPreferenceUpdated, NotificationCategory as EventCategory, NotificationPriority, +}; use crate::base::preferences::{ default_categories, default_channels, load_preferences, save_preferences, CategoryPreference, ChannelPreference, DeliveryChannel, NotificationCategory, RecipientPreferences, @@ -43,6 +46,12 @@ pub fn set_preferences( }; save_preferences(&env, &prefs); + + // Emit one event per channel so consumers see the updated fields. + for ch in prefs.channels.iter() { + emit_channel_preference_updated(&env, &recipient, &ch.channel, ch.enabled, prefs.updated_at); + } + Ok(()) } @@ -76,13 +85,17 @@ pub fn set_channel_preference( } if !found { - prefs - .channels - .push_back(ChannelPreference { channel, enabled }); + prefs.channels.push_back(ChannelPreference { + channel: channel.clone(), + enabled, + }); } prefs.updated_at = env.ledger().timestamp(); save_preferences(&env, &prefs); + + emit_channel_preference_updated(&env, &recipient, &channel, enabled, prefs.updated_at); + Ok(()) } @@ -138,6 +151,11 @@ pub fn reset_preferences(env: Env, recipient: Address) -> Result<(), Error> { }; save_preferences(&env, &prefs); + + for ch in prefs.channels.iter() { + emit_channel_preference_updated(&env, &recipient, &ch.channel, ch.enabled, prefs.updated_at); + } + Ok(()) } @@ -221,3 +239,21 @@ fn category_discriminant(category: &NotificationCategory) -> u32 { NotificationCategory::General => 4, } } + +fn emit_channel_preference_updated( + env: &Env, + recipient: &Address, + channel: &DeliveryChannel, + enabled: bool, + updated_at: u64, +) { + ChannelPreferenceUpdated { + recipient: recipient.clone(), + category: EventCategory::Notification, + priority: NotificationPriority::Medium, + channel: channel_discriminant(channel), + enabled, + updated_at, + } + .publish(env); +} diff --git a/contract/contracts/hello-world/src/tests/preferences_test.rs b/contract/contracts/hello-world/src/tests/preferences_test.rs index 40f5b9c..4870e6a 100644 --- a/contract/contracts/hello-world/src/tests/preferences_test.rs +++ b/contract/contracts/hello-world/src/tests/preferences_test.rs @@ -372,4 +372,119 @@ mod preferences_tests { assert!(!client.is_channel_enabled(&user2, &DeliveryChannel::Wallet)); assert!(client.is_channel_enabled(&user2, &DeliveryChannel::Email)); } + + // ============================================================================ + // ChannelPreferenceUpdated event emission + // ============================================================================ + + fn count_channel_preference_events(env: &Env) -> u32 { + use soroban_sdk::testutils::Events; + use soroban_sdk::{Symbol, TryFromVal}; + let target = Symbol::new(env, "channel_preference_updated"); + let mut n = 0u32; + for (_addr, topics, _data) in env.events().all().iter() { + if topics.is_empty() { + continue; + } + let first = topics.get(0).unwrap(); + if let Ok(name) = Symbol::try_from_val(env, &first) { + if name == target { + n += 1; + } + } + } + n + } + + #[test] + fn test_set_channel_preference_emits_channel_preference_updated() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let recipient = test_env.users.get(0).unwrap(); + + let before = count_channel_preference_events(&test_env.env); + client.set_channel_preference(&recipient, &DeliveryChannel::Email, &false); + let after = count_channel_preference_events(&test_env.env); + + assert_eq!( + after - before, + 1, + "set_channel_preference should emit ChannelPreferenceUpdated" + ); + + // Inspect latest event topics/data for updated fields + use soroban_sdk::testutils::Events; + use soroban_sdk::{Symbol, TryFromVal, Val}; + let target = Symbol::new(&test_env.env, "channel_preference_updated"); + let mut found = false; + for (_addr, topics, data) in test_env.env.events().all().iter() { + if topics.is_empty() { + continue; + } + let first = topics.get(0).unwrap(); + if let Ok(name) = Symbol::try_from_val(&test_env.env, &first) { + if name == target { + found = true; + // recipient is an indexed topic + assert!(topics.len() >= 2, "event should include recipient topic"); + // data map/struct should be present with updated fields + let _data: Val = data; + } + } + } + assert!(found, "ChannelPreferenceUpdated event must be present"); + } + + #[test] + fn test_set_preferences_emits_channel_events_for_each_channel() { + let test_env = setup_test_env(); + let client = AutoShareContractClient::new(&test_env.env, &test_env.autoshare_contract); + let recipient = test_env.users.get(0).unwrap(); + + let mut channels = Vec::new(&test_env.env); + channels.push_back(ChannelPreference { + channel: DeliveryChannel::Wallet, + enabled: true, + }); + channels.push_back(ChannelPreference { + channel: DeliveryChannel::Email, + enabled: false, + }); + channels.push_back(ChannelPreference { + channel: DeliveryChannel::InApp, + enabled: true, + }); + + let mut categories = Vec::new(&test_env.env); + categories.push_back(CategoryPreference { + category: NotificationCategory::Payment, + enabled: true, + }); + categories.push_back(CategoryPreference { + category: NotificationCategory::GroupMembership, + enabled: true, + }); + categories.push_back(CategoryPreference { + category: NotificationCategory::GroupStatus, + enabled: true, + }); + categories.push_back(CategoryPreference { + category: NotificationCategory::SystemAlerts, + enabled: true, + }); + categories.push_back(CategoryPreference { + category: NotificationCategory::General, + enabled: true, + }); + + let before = count_channel_preference_events(&test_env.env); + client.set_preferences(&recipient, &channels, &categories); + let after = count_channel_preference_events(&test_env.env); + + assert_eq!( + after - before, + 3, + "set_preferences should emit one ChannelPreferenceUpdated per channel" + ); + } } diff --git a/dashboard/src/pages/NotificationSearchPage.tsx b/dashboard/src/pages/NotificationSearchPage.tsx index bf0655c..f1de656 100644 --- a/dashboard/src/pages/NotificationSearchPage.tsx +++ b/dashboard/src/pages/NotificationSearchPage.tsx @@ -2,14 +2,20 @@ import { useState, useCallback, useEffect, useRef } from 'react'; import { NotificationSearchSkeleton } from '../components/NotificationSearchSkeleton'; import { getEventsApiBaseUrl } from '../config/eventsApiUrl'; import { useDebounce } from '../hooks/useDebounce'; -import { getEventsApiBaseUrl } from '../config/eventsApiUrl'; import { searchNotifications, type NotificationSearchResult, type NotificationSearchResponse, + type NotificationSearchParams, } from '../services/eventsApi'; +import { + buildNotificationExportBlob, + downloadBlob, + type NotificationExportFormat, +} from '../utils/notificationExport'; const PAGE_SIZE = 20; +const EXPORT_PAGE_SIZE = 100; const API_BASE = getEventsApiBaseUrl().replace(/\/api\/events\/?$/, ''); /** Delivery / processing status values used by scheduled + processed notifications. */ @@ -31,9 +37,6 @@ export const NOTIFICATION_TYPE_OPTIONS = [ { value: 'webhook', label: 'Webhook' }, { value: 'sms', label: 'SMS' }, ]; -const API_BASE = getEventsApiBaseUrl(); - -const STATUS_OPTIONS = ['', 'PENDING', 'PROCESSING', 'COMPLETED', 'FAILED', 'CANCELLED', 'PROCESSED']; export function NotificationSearchPage() { const [query, setQuery] = useState(''); @@ -49,6 +52,9 @@ export function NotificationSearchPage() { const [response, setResponse] = useState(null); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); + const [exporting, setExporting] = useState(false); + const [exportFormat, setExportFormat] = useState('json'); + const [exportMessage, setExportMessage] = useState(null); const debouncedQuery = useDebounce(query, 300); const debouncedSender = useDebounce(sender, 300); @@ -68,6 +74,28 @@ export function NotificationSearchPage() { const abortRef = useRef(null); + const currentFilters = useCallback((): NotificationSearchParams => { + return { + q: debouncedQuery || undefined, + sender: debouncedSender || undefined, + txHash: debouncedTxHash || undefined, + eventId: debouncedEventId || undefined, + status: status || undefined, + type: type || undefined, + startDate: dateFrom || undefined, + endDate: dateTo || undefined, + }; + }, [ + debouncedQuery, + debouncedSender, + debouncedTxHash, + debouncedEventId, + status, + type, + dateFrom, + dateTo, + ]); + const runSearch = useCallback(async () => { if (!hasParams) { setResponse(null); @@ -83,14 +111,7 @@ export function NotificationSearchPage() { try { const result = await searchNotifications(API_BASE, { - q: debouncedQuery || undefined, - sender: debouncedSender || undefined, - txHash: debouncedTxHash || undefined, - eventId: debouncedEventId || undefined, - status: status || undefined, - type: type || undefined, - startDate: dateFrom || undefined, - endDate: dateTo || undefined, + ...currentFilters(), limit: PAGE_SIZE, offset: (page - 1) * PAGE_SIZE, }); @@ -101,18 +122,7 @@ export function NotificationSearchPage() { } finally { setLoading(false); } - }, [ - debouncedQuery, - debouncedSender, - debouncedTxHash, - debouncedEventId, - status, - type, - dateFrom, - dateTo, - page, - hasParams, - ]); + }, [currentFilters, page, hasParams]); // Re-run search whenever debounced params change; reset page when filters change const filtersKey = `${debouncedQuery}|${debouncedSender}|${debouncedTxHash}|${debouncedEventId}|${status}|${type}|${dateFrom}|${dateTo}`; @@ -140,6 +150,46 @@ export function NotificationSearchPage() { setPage(1); setResponse(null); setError(null); + setExportMessage(null); + } + + async function handleExport() { + if (!hasParams) { + setExportMessage('Apply at least one filter before exporting.'); + return; + } + + setExporting(true); + setExportMessage(null); + + try { + const filters = currentFilters(); + const all: NotificationSearchResult[] = []; + let offset = 0; + let total = Infinity; + + while (offset < total) { + const pageResult = await searchNotifications(API_BASE, { + ...filters, + limit: EXPORT_PAGE_SIZE, + offset, + }); + all.push(...pageResult.results); + total = pageResult.total; + offset += pageResult.results.length; + if (pageResult.results.length === 0) break; + } + + const { blob, filename } = buildNotificationExportBlob(all, exportFormat, filters); + downloadBlob(blob, filename); + setExportMessage( + `Exported ${all.length} notification${all.length === 1 ? '' : 's'} as ${exportFormat.toUpperCase()}.` + ); + } catch (err: unknown) { + setExportMessage(err instanceof Error ? err.message : 'Export failed'); + } finally { + setExporting(false); + } } const totalPages = response ? response.totalPages : 0; @@ -151,6 +201,7 @@ export function NotificationSearchPage() {

Notification Search

Filter scheduled and processed notifications by type, delivery status, date range, sender, or free-text. + Export respects the filters you apply (JSON or CSV).

@@ -263,10 +314,52 @@ export function NotificationSearchPage() { - {hasParams && ( - + )} + + + + + + + + {exportMessage && ( +

+ {exportMessage} +

)} @@ -359,19 +452,13 @@ function NotificationResultCard({ result }: { result: NotificationSearchResult } {result.eventId && ( <>
Event ID
-
{result.eventId}
+
{result.eventId}
)} {result.txHash && ( <>
Tx Hash
-
{result.txHash}
- - )} - {result.contractAddress && ( - <> -
Contract
-
{result.contractAddress}
+
{result.txHash}
)} {result.targetRecipient && ( diff --git a/dashboard/src/utils/notificationExport.test.ts b/dashboard/src/utils/notificationExport.test.ts new file mode 100644 index 0000000..68152c6 --- /dev/null +++ b/dashboard/src/utils/notificationExport.test.ts @@ -0,0 +1,62 @@ +import { + buildNotificationExportCsv, + buildNotificationExportJson, + buildNotificationExportBlob, +} from './notificationExport'; +import type { NotificationSearchResult } from '../services/eventsApi'; + +const sample: NotificationSearchResult[] = [ + { + id: 1, + source: 'scheduled', + eventId: 'evt-1', + txHash: 'abc', + contractAddress: 'CC…', + notificationType: 'discord', + targetRecipient: 'https://hooks.example/1', + status: 'COMPLETED', + createdAt: '2026-07-26T12:00:00.000Z', + payload: '{"message":"hi"}', + }, + { + id: 2, + source: 'processed', + eventId: null, + txHash: null, + contractAddress: null, + notificationType: 'email', + targetRecipient: 'a@b.com', + status: 'FAILED', + createdAt: '2026-07-26T13:00:00.000Z', + payload: null, + }, +]; + +describe('notificationExport', () => { + it('builds JSON that includes filters and all rows', () => { + const json = buildNotificationExportJson(sample, { status: 'COMPLETED', q: 'hi' }); + const parsed = JSON.parse(json); + expect(parsed.format).toBe('json'); + expect(parsed.filters.status).toBe('COMPLETED'); + expect(parsed.total).toBe(2); + expect(parsed.notifications).toHaveLength(2); + }); + + it('builds CSV with header and escaped fields', () => { + const csv = buildNotificationExportCsv(sample); + const lines = csv.split('\n'); + expect(lines[0]).toContain('id,source,eventId'); + expect(lines).toHaveLength(3); + expect(lines[1]).toContain('COMPLETED'); + }); + + it('creates downloadable blobs for json and csv', () => { + const jsonBlob = buildNotificationExportBlob(sample, 'json', { type: 'discord' }); + expect(jsonBlob.filename).toMatch(/\.json$/); + expect(jsonBlob.blob.type).toContain('json'); + + const csvBlob = buildNotificationExportBlob(sample, 'csv'); + expect(csvBlob.filename).toMatch(/\.csv$/); + expect(csvBlob.blob.type).toContain('csv'); + }); +}); diff --git a/dashboard/src/utils/notificationExport.ts b/dashboard/src/utils/notificationExport.ts new file mode 100644 index 0000000..6c9386b --- /dev/null +++ b/dashboard/src/utils/notificationExport.ts @@ -0,0 +1,145 @@ +/** + * Notification export helpers for filtered search results. + * + * ## Export format + * + * ### JSON (`application/json`) + * ```json + * { + * "exportedAt": "2026-07-26T15:00:00.000Z", + * "format": "json", + * "filters": { "q": "...", "status": "COMPLETED", ... }, + * "total": 2, + * "notifications": [ + * { + * "id": 1, + * "source": "scheduled", + * "eventId": null, + * "txHash": null, + * "contractAddress": null, + * "notificationType": "discord", + * "targetRecipient": "https://...", + * "status": "COMPLETED", + * "createdAt": "2026-07-26T14:00:00.000Z", + * "payload": "{...}" + * } + * ] + * } + * ``` + * + * ### CSV (`text/csv`) + * Header row: + * `id,source,eventId,txHash,contractAddress,notificationType,targetRecipient,status,createdAt,payload` + * + * - Values are comma-separated. + * - Fields containing commas/quotes/newlines are double-quoted; quotes are escaped as `""`. + * - `payload` is exported as a single CSV cell (JSON string escaped). + * - Applied search filters are *not* embedded in the CSV body; they are reflected by + * which rows are included (only matching notifications are exported). + */ + +import type { + NotificationSearchParams, + NotificationSearchResult, +} from '../services/eventsApi'; + +export type NotificationExportFormat = 'json' | 'csv'; + +export interface NotificationExportDocument { + exportedAt: string; + format: 'json'; + filters: Partial; + total: number; + notifications: NotificationSearchResult[]; +} + +export function buildNotificationExportJson( + notifications: NotificationSearchResult[], + filters: Partial = {} +): string { + const doc: NotificationExportDocument = { + exportedAt: new Date().toISOString(), + format: 'json', + filters, + total: notifications.length, + notifications, + }; + return JSON.stringify(doc, null, 2); +} + +export function buildNotificationExportCsv( + notifications: NotificationSearchResult[] +): string { + const headers = [ + 'id', + 'source', + 'eventId', + 'txHash', + 'contractAddress', + 'notificationType', + 'targetRecipient', + 'status', + 'createdAt', + 'payload', + ]; + + const rows = notifications.map((n) => + [ + String(n.id), + n.source, + n.eventId ?? '', + n.txHash ?? '', + n.contractAddress ?? '', + n.notificationType ?? '', + n.targetRecipient ?? '', + n.status, + n.createdAt, + n.payload ?? '', + ] + .map(escapeCsvField) + .join(',') + ); + + return [headers.join(','), ...rows].join('\n'); +} + +export function buildNotificationExportBlob( + notifications: NotificationSearchResult[], + format: NotificationExportFormat, + filters: Partial = {} +): { blob: Blob; filename: string } { + const stamp = new Date().toISOString().replace(/[:.]/g, '-'); + if (format === 'csv') { + return { + blob: new Blob([buildNotificationExportCsv(notifications)], { + type: 'text/csv;charset=utf-8', + }), + filename: `notifications_export_${stamp}.csv`, + }; + } + + return { + blob: new Blob([buildNotificationExportJson(notifications, filters)], { + type: 'application/json;charset=utf-8', + }), + filename: `notifications_export_${stamp}.json`, + }; +} + +export function downloadBlob(blob: Blob, filename: string): void { + const url = URL.createObjectURL(blob); + const link = document.createElement('a'); + link.href = url; + link.download = filename; + document.body.appendChild(link); + link.click(); + document.body.removeChild(link); + URL.revokeObjectURL(url); +} + +function escapeCsvField(value: string): string { + if (/[",\n\r]/.test(value)) { + return `"${value.replace(/"/g, '""')}"`; + } + return value; +} diff --git a/docs/BULK_NOTIFICATION_IMPORT.md b/docs/BULK_NOTIFICATION_IMPORT.md new file mode 100644 index 0000000..45635ae --- /dev/null +++ b/docs/BULK_NOTIFICATION_IMPORT.md @@ -0,0 +1,68 @@ +# Bulk Notification Import + +Administrators can import multiple notifications from structured JSON or CSV +files via `POST /api/notifications/import`. + +## Endpoint + +``` +POST /api/notifications/import +Content-Type: application/json | text/csv +X-API-Key: # required when apiKeys are configured +``` + +## Accepted formats + +### JSON + +Array of records, or `{ "notifications": [ ... ] }`: + +```json +{ + "notifications": [ + { + "id": "n1", + "recipient": "https://discord.com/api/webhooks/...", + "channel": "discord", + "message": "Hello", + "executeAt": "2026-07-27T12:00:00.000Z" + } + ] +} +``` + +### CSV + +Header row required. Recognized columns: + +`id`, `recipient` / `target_recipient`, `channel` / `type`, `message`, +`execute_at` / `executeAt` + +```csv +id,recipient,channel,message,execute_at +n1,https://hooks.example/1,webhook,Ping,2026-07-27T12:00:00.000Z +``` + +## Behaviour + +- Valid records are scheduled via the normal scheduler API. +- Invalid records are **skipped** (not fatal); processing continues. +- Response always includes an import **summary**. + +```json +{ + "total": 3, + "imported": 2, + "skipped": 1, + "importedIds": [101, 102], + "skippedRecords": [ + { "index": 1, "reason": "Invalid channel: carrier-pigeon" } + ], + "format": "json", + "durationMs": 45 +} +``` + +## Channels + +`discord`, `webhook`, `email`, `sms` diff --git a/docs/JOB_MONITORING.md b/docs/JOB_MONITORING.md new file mode 100644 index 0000000..dbd00ab --- /dev/null +++ b/docs/JOB_MONITORING.md @@ -0,0 +1,78 @@ +# Background Job Monitoring + +Monitor scheduled notification jobs and inspect execution status for maintenance. + +## Overview + +The listener records every scheduled-notification job through `JobMonitor` +(`listener/src/services/job-monitor.ts`). The scheduler starts a monitored job +when it begins processing a notification and marks it **completed** or +**failed** when delivery finishes. + +Failed jobs are logged at `error` level and retained in an in-memory failure +log for quick inspection. + +## Endpoints + +### `GET /api/schedule/jobs` + +Returns a snapshot of recent job activity. + +```bash +curl http://localhost:3000/api/schedule/jobs?limit=25 +``` + +**Response fields** + +| Field | Description | +|-------|-------------| +| `activeCount` | Jobs currently running | +| `completedCount` | Completed jobs in history | +| `failedCount` | Failed jobs retained | +| `recentJobs` | Recent job records (running + finished) | +| `recentFailures` | Recent failed job records | + +**Job record shape** + +```json +{ + "jobId": "notification-42", + "jobName": "scheduled-notification", + "status": "failed", + "startedAt": "2026-07-26T15:00:00.000Z", + "finishedAt": "2026-07-26T15:00:01.200Z", + "durationMs": 1200, + "error": "webhook timeout", + "metadata": { "notificationId": 42, "type": "discord" } +} +``` + +### `GET /api/schedule/jobs/failures` + +Returns only failed jobs (most recent first). + +```bash +curl http://localhost:3000/api/schedule/jobs/failures?limit=50 +``` + +## Status values + +| Status | Meaning | +|--------|---------| +| `running` | Job accepted and currently executing | +| `completed` | Delivery succeeded | +| `failed` | Delivery failed or rejected (integrity, timing, etc.) | + +## Operational notes + +- History is in-memory (capped). Restart clears the monitor; durable attempt + history remains in `notification_execution_log` / history APIs. +- Combine with `GET /api/schedule/stats` and `GET /api/notifications/health` + for queue depth and worker health. +- Failed jobs always emit a structured log line: `Job monitor: job failed`. + +## Related docs + +- [MONITORING_INTEGRATION.md](./MONITORING_INTEGRATION.md) +- [METRICS_API_DOCUMENTATION.md](../METRICS_API_DOCUMENTATION.md) +- [listener/README-SCHEDULER.md](../listener/README-SCHEDULER.md) diff --git a/docs/MONITORING_INTEGRATION.md b/docs/MONITORING_INTEGRATION.md index 57f57fa..c8cf425 100644 --- a/docs/MONITORING_INTEGRATION.md +++ b/docs/MONITORING_INTEGRATION.md @@ -22,6 +22,16 @@ Notification ID: 100 --- +## Background Job Monitoring + +Scheduled jobs are also tracked by `JobMonitor`. See **[JOB_MONITORING.md](./JOB_MONITORING.md)** for: + +- `GET /api/schedule/jobs` — job status snapshot +- `GET /api/schedule/jobs/failures` — failed job log +- Job record schema and operational notes + +--- + ## Best Practices ### ✅ DO: Use the Metrics API diff --git a/docs/NOTIFICATION_EXPORT_FORMAT.md b/docs/NOTIFICATION_EXPORT_FORMAT.md new file mode 100644 index 0000000..b8186d2 --- /dev/null +++ b/docs/NOTIFICATION_EXPORT_FORMAT.md @@ -0,0 +1,56 @@ +# Notification Export Format + +Users can export filtered notification search results from the dashboard +**Notification Search** page via the **Export** button. + +## Behaviour + +- Export uses the **currently applied filters** (query, sender, tx hash, event + ID, status, type, date range). +- All matching rows are fetched (paginated under the hood) and downloaded. +- Supported formats: **JSON** and **CSV**. + +## JSON + +```json +{ + "exportedAt": "2026-07-26T15:00:00.000Z", + "format": "json", + "filters": { + "status": "COMPLETED", + "type": "discord", + "q": "billing" + }, + "total": 1, + "notifications": [ + { + "id": 42, + "source": "scheduled", + "eventId": null, + "txHash": null, + "contractAddress": null, + "notificationType": "discord", + "targetRecipient": "https://discord.com/api/webhooks/...", + "status": "COMPLETED", + "createdAt": "2026-07-26T14:00:00.000Z", + "payload": "{\"message\":\"...\"}" + } + ] +} +``` + +## CSV + +Header: + +``` +id,source,eventId,txHash,contractAddress,notificationType,targetRecipient,status,createdAt,payload +``` + +- Commas / quotes / newlines inside fields are escaped per RFC 4180-style rules. +- Filters are applied by row selection (only matching notifications appear). + +## Implementation + +- UI: `dashboard/src/pages/NotificationSearchPage.tsx` +- Helpers: `dashboard/src/utils/notificationExport.ts` diff --git a/listener/src/api/events-server.ts b/listener/src/api/events-server.ts index 6468f90..f8eac12 100644 --- a/listener/src/api/events-server.ts +++ b/listener/src/api/events-server.ts @@ -49,6 +49,8 @@ import { ArchiveStore } from '../services/archive-store'; import { ArchiveService } from '../services/archive-service'; import { NotificationMetricsStore } from '../services/notification-metrics-store'; import { NotificationHealthMonitor } from '../services/notification-health-monitor'; +import { getJobMonitor } from '../services/job-monitor'; +import { NotificationImportService } from '../services/notification-import-service'; export interface EventsServerOptions { port: number; @@ -739,6 +741,49 @@ export function createEventsServer(options: EventsServerOptions): http.Server { return; } + // POST /api/notifications/import — bulk import from JSON or CSV + if (req.method === 'POST' && url.pathname === '/api/notifications/import') { + if (!options.notificationAPI) { + res.writeHead(503, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ error: 'Scheduler not enabled' })); + return; + } + + const apiKeyHeader = req.headers['x-api-key']; + if (options.apiKeys && options.apiKeys.length > 0) { + const provided = Array.isArray(apiKeyHeader) ? apiKeyHeader[0] : apiKeyHeader; + const allowed = options.apiKeys.some((k) => k.key === provided); + if (!allowed) { + res.writeHead(401, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ error: 'Unauthorized' })); + return; + } + } + + let body = ''; + req.on('data', (chunk) => { body += chunk.toString(); }); + req.on('end', async () => { + try { + const contentType = req.headers['content-type'] || ''; + const importer = new NotificationImportService(options.notificationAPI!); + const summary = await importer.importFromBody(body, contentType, { requestId }); + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify(summary)); + logger.info('Bulk notification import finished', { + requestId, + correlationId, + imported: summary.imported, + skipped: summary.skipped, + }); + } catch (error) { + logger.error('Failed to import notifications', { error, requestId, correlationId }); + res.writeHead(400, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ error: (error as Error).message })); + } + }); + return; + } + // POST /api/schedule if (req.method === 'POST' && url.pathname === '/api/schedule') { if (!options.notificationAPI) { @@ -824,6 +869,32 @@ export function createEventsServer(options: EventsServerOptions): http.Server { return; } + // GET /api/schedule/jobs — background job monitoring snapshot + if (req.method === 'GET' && url.pathname === '/api/schedule/jobs') { + const monitor = getJobMonitor(); + const limitParam = url.searchParams.get('limit'); + const limit = limitParam ? Math.min(Math.max(parseInt(limitParam, 10) || 25, 1), 200) : 25; + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ + ...monitor.getSnapshot(), + recentJobs: monitor.listRecentJobs(limit), + recentFailures: monitor.listFailures(limit), + }) + ); + return; + } + + // GET /api/schedule/jobs/failures — failed job log + if (req.method === 'GET' && url.pathname === '/api/schedule/jobs/failures') { + const monitor = getJobMonitor(); + const limitParam = url.searchParams.get('limit'); + const limit = limitParam ? Math.min(Math.max(parseInt(limitParam, 10) || 50, 1), 200) : 50; + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ failures: monitor.listFailures(limit), count: monitor.listFailures(limit).length })); + return; + } + // GET /api/schedule/:id if (req.method === 'GET' && url.pathname.startsWith('/api/schedule/')) { if (!options.notificationAPI) { diff --git a/listener/src/services/job-monitor.test.ts b/listener/src/services/job-monitor.test.ts new file mode 100644 index 0000000..1dcdc6e --- /dev/null +++ b/listener/src/services/job-monitor.test.ts @@ -0,0 +1,47 @@ +import { JobMonitor, getJobMonitor, resetJobMonitor } from './job-monitor'; + +describe('JobMonitor', () => { + let monitor: JobMonitor; + + beforeEach(() => { + resetJobMonitor(); + monitor = new JobMonitor(); + }); + + afterEach(() => { + resetJobMonitor(); + }); + + it('records job status through start and complete', () => { + monitor.startJob('job-1', 'notification-poll', { batchSize: 10 }); + expect(monitor.getJob('job-1')?.status).toBe('running'); + + monitor.completeJob('job-1', { processed: 3 }); + const job = monitor.getJob('job-1'); + expect(job?.status).toBe('completed'); + expect(job?.finishedAt).toBeDefined(); + expect(job?.durationMs).toBeGreaterThanOrEqual(0); + expect(job?.metadata?.processed).toBe(3); + }); + + it('logs failed jobs with error details', () => { + monitor.startJob('job-2', 'discord-delivery'); + monitor.failJob('job-2', 'webhook timeout'); + + const failures = monitor.listFailures(); + expect(failures).toHaveLength(1); + expect(failures[0].status).toBe('failed'); + expect(failures[0].error).toBe('webhook timeout'); + + const snapshot = monitor.getSnapshot(); + expect(snapshot.failedCount).toBe(1); + expect(snapshot.activeCount).toBe(0); + }); + + it('exposes a process-wide singleton', () => { + const a = getJobMonitor(); + a.startJob('shared', 'scheduler-tick'); + const b = getJobMonitor(); + expect(b.getJob('shared')?.status).toBe('running'); + }); +}); diff --git a/listener/src/services/job-monitor.ts b/listener/src/services/job-monitor.ts new file mode 100644 index 0000000..b5075df --- /dev/null +++ b/listener/src/services/job-monitor.ts @@ -0,0 +1,164 @@ +import logger from '../utils/logger'; + +export type JobStatus = 'running' | 'completed' | 'failed'; + +export interface JobRecord { + jobId: string; + jobName: string; + status: JobStatus; + startedAt: string; + finishedAt?: string; + durationMs?: number; + error?: string; + metadata?: Record; +} + +export interface JobMonitorSnapshot { + activeCount: number; + completedCount: number; + failedCount: number; + recentJobs: JobRecord[]; + recentFailures: JobRecord[]; +} + +const MAX_HISTORY = 200; +const MAX_FAILURES = 100; + +/** + * In-memory monitor for scheduled / background job execution status. + * Records start/complete/fail so operators can inspect recent runs + * and failed jobs without querying raw execution logs. + */ +export class JobMonitor { + private active = new Map(); + private history: JobRecord[] = []; + private failures: JobRecord[] = []; + + startJob(jobId: string, jobName: string, metadata?: Record): void { + const record: JobRecord = { + jobId, + jobName, + status: 'running', + startedAt: new Date().toISOString(), + metadata, + }; + this.active.set(jobId, record); + logger.debug('Job monitor: job started', { jobId, jobName }); + } + + completeJob(jobId: string, metadata?: Record): void { + const existing = this.active.get(jobId); + if (!existing) { + logger.warn('Job monitor: complete called for unknown job', { jobId }); + return; + } + + const finishedAt = new Date().toISOString(); + const durationMs = Date.now() - Date.parse(existing.startedAt); + const record: JobRecord = { + ...existing, + status: 'completed', + finishedAt, + durationMs, + metadata: { ...existing.metadata, ...metadata }, + }; + + this.active.delete(jobId); + this.pushHistory(record); + logger.debug('Job monitor: job completed', { jobId, durationMs }); + } + + failJob(jobId: string, error: string, metadata?: Record): void { + const existing = this.active.get(jobId) ?? { + jobId, + jobName: 'unknown', + status: 'running' as JobStatus, + startedAt: new Date().toISOString(), + }; + + const finishedAt = new Date().toISOString(); + const durationMs = Date.now() - Date.parse(existing.startedAt); + const record: JobRecord = { + ...existing, + status: 'failed', + finishedAt, + durationMs, + error, + metadata: { ...existing.metadata, ...metadata }, + }; + + this.active.delete(jobId); + this.pushHistory(record); + this.pushFailure(record); + + logger.error('Job monitor: job failed', { + jobId, + jobName: record.jobName, + error, + durationMs, + }); + } + + getJob(jobId: string): JobRecord | undefined { + return ( + this.active.get(jobId) ?? + this.history.find((j) => j.jobId === jobId) ?? + this.failures.find((j) => j.jobId === jobId) + ); + } + + listRecentJobs(limit = 50): JobRecord[] { + const active = Array.from(this.active.values()); + return [...active, ...this.history].slice(0, limit); + } + + listFailures(limit = 50): JobRecord[] { + return this.failures.slice(0, limit); + } + + getSnapshot(): JobMonitorSnapshot { + return { + activeCount: this.active.size, + completedCount: this.history.filter((j) => j.status === 'completed').length, + failedCount: this.failures.length, + recentJobs: this.listRecentJobs(25), + recentFailures: this.listFailures(25), + }; + } + + reset(): void { + this.active.clear(); + this.history = []; + this.failures = []; + } + + private pushHistory(record: JobRecord): void { + this.history.unshift(record); + if (this.history.length > MAX_HISTORY) { + this.history.length = MAX_HISTORY; + } + } + + private pushFailure(record: JobRecord): void { + this.failures.unshift(record); + if (this.failures.length > MAX_FAILURES) { + this.failures.length = MAX_FAILURES; + } + } +} + +let instance: JobMonitor | null = null; + +export function getJobMonitor(): JobMonitor { + if (!instance) { + instance = new JobMonitor(); + } + return instance; +} + +export function resetJobMonitor(): void { + if (instance) { + instance.reset(); + } + instance = null; +} diff --git a/listener/src/services/notification-import-service.test.ts b/listener/src/services/notification-import-service.test.ts new file mode 100644 index 0000000..e7bef6d --- /dev/null +++ b/listener/src/services/notification-import-service.test.ts @@ -0,0 +1,88 @@ +import { NotificationImportService } from './notification-import-service'; +import { NotificationAPI } from './notification-api'; +import { NotificationType } from '../types/scheduled-notification'; + +describe('NotificationImportService', () => { + let scheduleNotification: jest.Mock; + let service: NotificationImportService; + + beforeEach(() => { + scheduleNotification = jest.fn().mockResolvedValue(42); + const api = { scheduleNotification } as unknown as NotificationAPI; + service = new NotificationImportService(api); + }); + + it('imports valid JSON records and returns a summary', async () => { + const executeAt = new Date(Date.now() + 60 * 60 * 1000).toISOString(); + const summary = await service.importFromBody([ + { + id: 'n1', + recipient: 'https://discord.com/api/webhooks/1', + channel: 'discord', + message: 'Hello', + executeAt, + }, + { + id: 'n2', + recipient: 'user@example.com', + channel: 'email', + message: 'Hi', + executeAt, + }, + ]); + + expect(summary.imported).toBe(2); + expect(summary.skipped).toBe(0); + expect(summary.importedIds).toEqual([42, 42]); + expect(summary.format).toBe('json'); + expect(scheduleNotification).toHaveBeenCalledTimes(2); + expect(scheduleNotification.mock.calls[0][0].notificationType).toBe(NotificationType.DISCORD); + }); + + it('skips invalid records safely without failing the batch', async () => { + const executeAt = new Date(Date.now() + 60 * 60 * 1000).toISOString(); + const summary = await service.importFromBody({ + notifications: [ + { + id: 'ok', + recipient: 'https://hooks.example/1', + channel: 'webhook', + message: 'Valid', + executeAt, + }, + { + id: 'bad', + channel: 'carrier-pigeon', + message: 'Nope', + executeAt, + }, + { + id: 'missing', + recipient: 'x@y.com', + channel: 'email', + executeAt, + }, + ], + }); + + expect(summary.total).toBe(3); + expect(summary.imported).toBe(1); + expect(summary.skipped).toBe(2); + expect(summary.skippedRecords.map((s) => s.index).sort()).toEqual([1, 2]); + }); + + it('parses CSV and imports valid rows', async () => { + const executeAt = new Date(Date.now() + 60 * 60 * 1000).toISOString(); + const csv = + `id,recipient,channel,message,execute_at\n` + + `c1,https://discord.com/api/webhooks/9,discord,Ping,${executeAt}\n` + + `c2,,sms,Missing recipient,${executeAt}\n`; + + const summary = await service.importFromBody(csv, 'text/csv'); + + expect(summary.format).toBe('csv'); + expect(summary.imported).toBe(1); + expect(summary.skipped).toBe(1); + expect(summary.skippedRecords[0].reason).toMatch(/recipient/i); + }); +}); diff --git a/listener/src/services/notification-import-service.ts b/listener/src/services/notification-import-service.ts new file mode 100644 index 0000000..95c05f1 --- /dev/null +++ b/listener/src/services/notification-import-service.ts @@ -0,0 +1,281 @@ +import logger from '../utils/logger'; +import { NotificationAPI } from './notification-api'; +import { VALID_CHANNELS } from '../utils/batch-validator'; +import { NotificationType } from '../types/scheduled-notification'; + +export interface ImportRecord { + id?: string; + recipient?: string; + channel?: string; + message?: string; + executeAt?: string; + notificationType?: string; + payload?: Record; + targetRecipient?: string; + maxRetries?: number; + priority?: number; + metadata?: Record; +} + +export interface ImportSkipDetail { + index: number; + reason: string; + record?: unknown; +} + +export interface ImportSummary { + total: number; + imported: number; + skipped: number; + importedIds: number[]; + skippedRecords: ImportSkipDetail[]; + format: 'json' | 'csv'; + durationMs: number; +} + +export interface ImportOptions { + /** Default execute-at offset in ms from now when record omits executeAt (default 5 min). */ + defaultExecuteAtOffsetMs?: number; + requestId?: string; +} + +const CHANNEL_TO_TYPE: Record = { + discord: NotificationType.DISCORD, + email: NotificationType.EMAIL, + webhook: NotificationType.WEBHOOK, + sms: NotificationType.SMS, +}; + +/** + * Parses structured JSON or CSV notification files and imports valid rows. + * Invalid records are skipped safely; a summary is always returned. + */ +export class NotificationImportService { + constructor(private notificationAPI: NotificationAPI) {} + + async importFromBody( + body: string | unknown, + contentType?: string, + options: ImportOptions = {} + ): Promise { + const start = Date.now(); + const format = this.detectFormat(body, contentType); + const records = + format === 'csv' + ? this.parseCsv(typeof body === 'string' ? body : String(body)) + : this.parseJson(body); + + return this.importRecords(records, format, options, start); + } + + async importRecords( + records: ImportRecord[], + format: 'json' | 'csv' = 'json', + options: ImportOptions = {}, + startedAt = Date.now() + ): Promise { + const skippedRecords: ImportSkipDetail[] = []; + const importedIds: number[] = []; + const offsetMs = options.defaultExecuteAtOffsetMs ?? 5 * 60 * 1000; + + for (let index = 0; index < records.length; index++) { + const record = records[index]; + const validationError = this.validateImportRecord(record); + if (validationError) { + skippedRecords.push({ index, reason: validationError, record }); + continue; + } + + try { + const executeAt = record.executeAt + ? new Date(record.executeAt) + : new Date(Date.now() + offsetMs); + + if (isNaN(executeAt.getTime()) || executeAt <= new Date()) { + skippedRecords.push({ + index, + reason: 'executeAt must be a valid future timestamp', + record, + }); + continue; + } + + const targetRecipient = record.targetRecipient || record.recipient!; + const channel = (record.channel || record.notificationType || 'discord').toLowerCase(); + const notificationType = + CHANNEL_TO_TYPE[channel] ?? NotificationType.DISCORD; + + const payload = + record.payload ?? + ({ + id: record.id, + message: record.message, + channel, + recipient: targetRecipient, + } as Record); + + const id = await this.notificationAPI.scheduleNotification( + { + payload, + notificationType, + targetRecipient, + executeAt, + maxRetries: record.maxRetries, + priority: record.priority, + metadata: { + ...record.metadata, + importId: record.id, + importedAt: new Date().toISOString(), + }, + }, + options.requestId + ); + + importedIds.push(id); + } catch (err) { + const reason = err instanceof Error ? err.message : 'Unknown import error'; + skippedRecords.push({ index, reason, record }); + logger.warn('Skipped notification during bulk import', { + index, + reason, + requestId: options.requestId, + }); + } + } + + const summary: ImportSummary = { + total: records.length, + imported: importedIds.length, + skipped: skippedRecords.length, + importedIds, + skippedRecords, + format, + durationMs: Date.now() - startedAt, + }; + + logger.info('Bulk notification import complete', { + requestId: options.requestId, + imported: summary.imported, + skipped: summary.skipped, + total: summary.total, + format: summary.format, + }); + + return summary; + } + + private detectFormat(body: string | unknown, contentType?: string): 'json' | 'csv' { + if (contentType?.includes('text/csv') || contentType?.includes('application/csv')) { + return 'csv'; + } + if (typeof body === 'string') { + const trimmed = body.trim(); + if (trimmed.startsWith('[') || trimmed.startsWith('{')) { + return 'json'; + } + if (trimmed.includes(',') && /recipient/i.test(trimmed)) { + return 'csv'; + } + } + return 'json'; + } + + private parseJson(body: string | unknown): ImportRecord[] { + const data = typeof body === 'string' ? JSON.parse(body || 'null') : body; + if (Array.isArray(data)) { + return data as ImportRecord[]; + } + if ( + data && + typeof data === 'object' && + Array.isArray((data as { notifications?: unknown }).notifications) + ) { + return (data as { notifications: ImportRecord[] }).notifications; + } + if ( + data && + typeof data === 'object' && + Array.isArray((data as { records?: unknown }).records) + ) { + return (data as { records: ImportRecord[] }).records; + } + throw new Error('JSON import body must be an array or { notifications: [] }'); + } + + private parseCsv(csv: string): ImportRecord[] { + const lines = csv + .split(/\r?\n/) + .map((l) => l.trim()) + .filter((l) => l.length > 0); + + if (lines.length < 2) { + return []; + } + + const headers = this.splitCsvLine(lines[0]).map((h) => h.trim().toLowerCase()); + const records: ImportRecord[] = []; + + for (let i = 1; i < lines.length; i++) { + const cols = this.splitCsvLine(lines[i]); + const row: Record = {}; + headers.forEach((h, idx) => { + row[h] = (cols[idx] ?? '').trim(); + }); + + records.push({ + id: row.id || row.notification_id, + recipient: row.recipient || row.targetrecipient || row.target_recipient, + channel: row.channel || row.type || row.notificationtype, + message: row.message || row.body || row.content, + executeAt: row.executeat || row.execute_at || row.scheduled_at, + notificationType: row.notificationtype || row.notification_type || row.type, + targetRecipient: row.targetrecipient || row.target_recipient || row.recipient, + }); + } + + return records; + } + + private splitCsvLine(line: string): string[] { + const result: string[] = []; + let current = ''; + let inQuotes = false; + + for (let i = 0; i < line.length; i++) { + const ch = line[i]; + if (ch === '"') { + if (inQuotes && line[i + 1] === '"') { + current += '"'; + i++; + } else { + inQuotes = !inQuotes; + } + } else if (ch === ',' && !inQuotes) { + result.push(current); + current = ''; + } else { + current += ch; + } + } + result.push(current); + return result; + } + + private validateImportRecord(record: ImportRecord): string | null { + const recipient = record.recipient || record.targetRecipient; + if (!recipient) { + return 'Missing required field: recipient'; + } + const channel = (record.channel || record.notificationType || '').toLowerCase(); + if (!channel) { + return 'Missing required field: channel'; + } + if (!(VALID_CHANNELS as readonly string[]).includes(channel)) { + return `Invalid channel: ${channel}`; + } + if (!record.message && !record.payload) { + return 'Missing required field: message or payload'; + } + return null; + } +} diff --git a/listener/src/services/notification-scheduler.ts b/listener/src/services/notification-scheduler.ts index 0fbb026..7286ef1 100644 --- a/listener/src/services/notification-scheduler.ts +++ b/listener/src/services/notification-scheduler.ts @@ -7,6 +7,7 @@ import { DiscordNotificationService } from './discord-notification'; import { BatchValidationService } from './batch-validation-service'; import { NotificationChannel } from '../utils/batch-validator'; import { getWorkerManager } from './worker-manager'; +import { getJobMonitor } from './job-monitor'; /** * Background scheduler that processes scheduled notifications @@ -176,7 +177,8 @@ export class NotificationScheduler { return; } - // Process each notification with job tracking + // Process each notification with job tracking + monitoring + const jobMonitor = getJobMonitor(); for (const notification of notifications) { const jobId = `notification-${notification.id}`; if (!workerManager.startJob(jobId)) { @@ -191,8 +193,14 @@ export class NotificationScheduler { continue; } + jobMonitor.startJob(jobId, 'scheduled-notification', { + notificationId: notification.id, + type: notification.notificationType, + requestId, + }); + try { - await this.processNotification(notification, requestId); + await this.processNotification(notification, requestId, jobId); } finally { workerManager.completeJob(jobId); } @@ -219,10 +227,12 @@ export class NotificationScheduler { */ private async processNotification( notification: ScheduledNotification, - requestId: string + requestId: string, + jobId?: string ): Promise { const startTime = Date.now(); const executionAttempt = notification.retryCount + 1; + const jobMonitor = getJobMonitor(); try { logger.info('Processing scheduled notification', { @@ -251,6 +261,11 @@ export class NotificationScheduler { notification.retryCount, notification.maxRetries ); + if (jobId) { + jobMonitor.failJob(jobId, 'Not yet due for execution', { + notificationId: notification.id, + }); + } return; } @@ -262,6 +277,7 @@ export class NotificationScheduler { now, missedByMs: timeDiff, }); + } // Verify payload integrity before executing const secret = process.env.PAYLOAD_INTEGRITY_SECRET; if (secret) { @@ -282,6 +298,11 @@ export class NotificationScheduler { notification.maxRetries, // exhaust retries — don't retry a tampered payload notification.maxRetries ); + if (jobId) { + jobMonitor.failJob(jobId, 'Payload integrity check failed: hash mismatch', { + notificationId: notification.id, + }); + } return; } } @@ -301,6 +322,13 @@ export class NotificationScheduler { durationMs, }); + if (jobId) { + jobMonitor.completeJob(jobId, { + notificationId: notification.id, + durationMs, + }); + } + logger.info('Notification delivered successfully', { requestId, id: notification.id, @@ -320,6 +348,13 @@ export class NotificationScheduler { durationMs, }); + if (jobId) { + jobMonitor.failJob(jobId, (error as Error).message, { + notificationId: notification.id, + attempt: executionAttempt, + }); + } + const willRetry = notification.retryCount + 1 < notification.maxRetries; const nextRetryAt = willRetry ? new Date(Date.now() + (this.config.retryDelayMs ?? 5_000))