From 0e99dd39e81f63f628165635edf06d8a6a9d3ad1 Mon Sep 17 00:00:00 2001 From: Demilade10 Date: Fri, 31 Jul 2026 12:04:32 +0100 Subject: [PATCH] Add pending jobs queue endpoint and verify queue visibility (#490) --- listener/src/api/events-server.ts | 33 ++ listener/src/api/queue-visibility.test.ts | 386 ++++++++++++++++++ listener/src/services/notification-api.ts | 10 + .../scheduled-notification-repository.ts | 44 ++ 4 files changed, 473 insertions(+) create mode 100644 listener/src/api/queue-visibility.test.ts diff --git a/listener/src/api/events-server.ts b/listener/src/api/events-server.ts index d7b13c2..7569d7e 100644 --- a/listener/src/api/events-server.ts +++ b/listener/src/api/events-server.ts @@ -901,6 +901,39 @@ export function createEventsServer(options: EventsServerOptions): http.Server { return; } + // GET /api/schedule/queue — queue visibility: list pending jobs + // Returns jobs currently waiting in the queue with id, type, enqueued + // time (createdAt), scheduled delivery time (executeAt), priority, and + // retryCount. Accepts an optional ?limit= query param (default 100). + if (req.method === 'GET' && url.pathname === '/api/schedule/queue') { + if (!options.notificationAPI) { + res.writeHead(503, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ error: 'Scheduler not enabled' })); + return; + } + + const limitParam = url.searchParams.get('limit'); + const limit = limitParam ? Math.max(1, Math.min(500, parseInt(limitParam, 10) || 100)) : undefined; + + options.notificationAPI.getPendingJobs(limit) + .then((jobs) => { + logger.info('Handling GET /api/schedule/queue', { + requestId, + correlationId, + count: jobs.length, + durationMs: Date.now() - startTime, + }); + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ count: jobs.length, jobs })); + }) + .catch((error) => { + logger.error('Failed to get pending jobs', { error, requestId, correlationId }); + res.writeHead(500, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ error: (error as Error).message })); + }); + return; + } + // GET /api/schedule/:id if (req.method === 'GET' && url.pathname.startsWith('/api/schedule/')) { if (!options.notificationAPI) { diff --git a/listener/src/api/queue-visibility.test.ts b/listener/src/api/queue-visibility.test.ts new file mode 100644 index 0000000..75a9853 --- /dev/null +++ b/listener/src/api/queue-visibility.test.ts @@ -0,0 +1,386 @@ +/** + * Integration tests for queue visibility API endpoints (GitHub issue #490). + * + * Covers three acceptance criteria: + * 1. Queue statistics — GET /api/schedule/stats (already existed; verified here) + * 2. Pending jobs list — GET /api/schedule/queue (new) + * 3. Queue health — GET /api/notifications/health (already existed; verified here) + */ + +import { describe, it, expect, beforeEach, afterEach } from '@jest/globals'; +import http from 'http'; +import { Database } from '../database/database'; +import { ScheduledNotificationRepository } from '../services/scheduled-notification-repository'; +import { NotificationAPI } from '../services/notification-api'; +import { NotificationHealthMonitor } from '../services/notification-health-monitor'; +import { createEventsServer, EventsServerOptions } from './events-server'; +import { NotificationType, NotificationStatus } from '../types/scheduled-notification'; +import path from 'path'; +import fs from 'fs/promises'; + +// --------------------------------------------------------------------------- +// Test helpers +// --------------------------------------------------------------------------- + +const TEST_PORT = 38090; +const TEST_DB_PATH = path.join(__dirname, '../../test-data/test-queue-visibility.db'); + +async function makeRequest( + server: http.Server, + method: string, + pathname: string, +): Promise<{ status: number; body: unknown }> { + return new Promise((resolve, reject) => { + const port = (server.address() as { port: number }).port; + const req = http.request( + { + hostname: '127.0.0.1', + port, + path: pathname, + method, + headers: { 'Content-Type': 'application/json' }, + }, + (res) => { + let data = ''; + res.on('data', (chunk) => { data += chunk; }); + res.on('end', () => { + try { + resolve({ status: res.statusCode!, body: JSON.parse(data) }); + } catch { + resolve({ status: res.statusCode!, body: data }); + } + }); + }, + ); + req.on('error', reject); + req.end(); + }); +} + +// --------------------------------------------------------------------------- +// Suite +// --------------------------------------------------------------------------- + +describe('Queue Visibility API (issue #490)', () => { + let db: Database; + let repository: ScheduledNotificationRepository; + let notificationAPI: NotificationAPI; + let server: http.Server; + + beforeEach(async () => { + // Clean up any leftover test database from a previous run. + try { await fs.unlink(TEST_DB_PATH); } catch { /* ignore */ } + + db = new Database(TEST_DB_PATH); + await db.initialize(); + repository = new ScheduledNotificationRepository(db); + notificationAPI = new NotificationAPI(repository); + + const options: EventsServerOptions = { + port: TEST_PORT, + stellarRpcUrl: 'https://soroban-testnet.stellar.org', + notificationAPI, + }; + + server = createEventsServer(options); + await new Promise((resolve) => server.listen(TEST_PORT, resolve)); + }); + + afterEach(async () => { + await new Promise((resolve, reject) => + server.close((err) => (err ? reject(err) : resolve())), + ); + await db.close(); + try { await fs.unlink(TEST_DB_PATH); } catch { /* ignore */ } + }); + + // ── Criterion 1: Queue statistics ──────────────────────────────────────── + + describe('GET /api/schedule/stats — queue statistics', () => { + it('returns counts for pending, processing, completed, failed, and overdue', async () => { + // Seed: one pending, one completed + await repository.create({ + payload: { test: 'pending' }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'https://example.com/hook', + executeAt: new Date(Date.now() + 60_000), + maxRetries: 3, + }); + + const completedId = await repository.create({ + payload: { test: 'completed' }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'https://example.com/hook', + executeAt: new Date(Date.now() - 1000), + maxRetries: 3, + }); + await repository.markAsCompleted(completedId); + + const { status, body } = await makeRequest(server, 'GET', '/api/schedule/stats'); + + expect(status).toBe(200); + const stats = body as Record; + // Must include all required count fields. + expect(stats).toHaveProperty('pending'); + expect(stats).toHaveProperty('processing'); + expect(stats).toHaveProperty('completed'); + expect(stats).toHaveProperty('failed'); + expect(stats).toHaveProperty('overdue'); + // Spot-check the values match what we seeded. + expect(stats.pending).toBe(1); + expect(stats.completed).toBe(1); + }); + + it('returns 503 when the scheduler is not enabled', async () => { + // Rebuild the server without a notificationAPI. + await new Promise((resolve, reject) => + server.close((err) => (err ? reject(err) : resolve())), + ); + + server = createEventsServer({ + port: TEST_PORT, + stellarRpcUrl: 'https://soroban-testnet.stellar.org', + notificationAPI: null, + }); + await new Promise((resolve) => server.listen(TEST_PORT, resolve)); + + const { status } = await makeRequest(server, 'GET', '/api/schedule/stats'); + expect(status).toBe(503); + }); + }); + + // ── Criterion 2: Pending jobs list ─────────────────────────────────────── + + describe('GET /api/schedule/queue — pending jobs', () => { + it('returns an empty list when no notifications are pending', async () => { + const { status, body } = await makeRequest(server, 'GET', '/api/schedule/queue'); + + expect(status).toBe(200); + const payload = body as { count: number; jobs: unknown[] }; + expect(payload.count).toBe(0); + expect(payload.jobs).toEqual([]); + }); + + it('returns pending jobs with required fields: id, notificationType, createdAt, executeAt, priority, retryCount', async () => { + await repository.create({ + payload: { msg: 'hello' }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'https://example.com/hook', + executeAt: new Date(Date.now() + 120_000), + maxRetries: 3, + priority: 5, + }); + + const { status, body } = await makeRequest(server, 'GET', '/api/schedule/queue'); + + expect(status).toBe(200); + const payload = body as { count: number; jobs: Record[] }; + expect(payload.count).toBe(1); + expect(payload.jobs).toHaveLength(1); + + const job = payload.jobs[0]; + // All minimum required fields must be present. + expect(job).toHaveProperty('id'); + expect(typeof job.id).toBe('number'); + expect(job).toHaveProperty('notificationType', NotificationType.DISCORD); + expect(job).toHaveProperty('createdAt'); + expect(typeof job.createdAt).toBe('string'); + expect(job).toHaveProperty('executeAt'); + expect(typeof job.executeAt).toBe('string'); + expect(job).toHaveProperty('priority'); + expect(typeof job.priority).toBe('number'); + expect(job).toHaveProperty('retryCount'); + expect(typeof job.retryCount).toBe('number'); + }); + + it('does not return completed or failed notifications', async () => { + // Create and complete one notification. + const completedId = await repository.create({ + payload: { msg: 'done' }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'https://example.com/hook', + executeAt: new Date(Date.now() - 1000), + maxRetries: 1, + }); + await repository.markAsCompleted(completedId); + + // Create and exhaust a failed one. + const failedId = await repository.create({ + payload: { msg: 'fail' }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'https://example.com/hook', + executeAt: new Date(Date.now() - 1000), + maxRetries: 1, + }); + await repository.markAsFailedOrRetry(failedId, new Error('oops'), 0, 1); + + // Add one pending notification. + await repository.create({ + payload: { msg: 'waiting' }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'https://example.com/hook', + executeAt: new Date(Date.now() + 60_000), + maxRetries: 3, + }); + + const { status, body } = await makeRequest(server, 'GET', '/api/schedule/queue'); + + expect(status).toBe(200); + const payload = body as { count: number; jobs: Record[] }; + // Only the PENDING notification should appear. + expect(payload.count).toBe(1); + expect(payload.jobs).toHaveLength(1); + }); + + it('respects the ?limit= query parameter', async () => { + // Create 5 pending notifications. + for (let i = 0; i < 5; i++) { + await repository.create({ + payload: { seq: i }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'https://example.com/hook', + executeAt: new Date(Date.now() + 60_000 + i * 1000), + maxRetries: 3, + }); + } + + const { status, body } = await makeRequest( + server, + 'GET', + '/api/schedule/queue?limit=3', + ); + + expect(status).toBe(200); + const payload = body as { count: number; jobs: unknown[] }; + expect(payload.count).toBe(3); + expect(payload.jobs).toHaveLength(3); + }); + + it('returns 503 when the scheduler is not enabled', async () => { + await new Promise((resolve, reject) => + server.close((err) => (err ? reject(err) : resolve())), + ); + + server = createEventsServer({ + port: TEST_PORT, + stellarRpcUrl: 'https://soroban-testnet.stellar.org', + notificationAPI: null, + }); + await new Promise((resolve) => server.listen(TEST_PORT, resolve)); + + const { status } = await makeRequest(server, 'GET', '/api/schedule/queue'); + expect(status).toBe(503); + }); + + it('is accessible via the /api/v1/ version alias', async () => { + await repository.create({ + payload: { msg: 'v1-test' }, + notificationType: NotificationType.DISCORD, + targetRecipient: 'https://example.com/hook', + executeAt: new Date(Date.now() + 60_000), + maxRetries: 3, + }); + + const { status, body } = await makeRequest( + server, + 'GET', + '/api/v1/schedule/queue', + ); + + expect(status).toBe(200); + const payload = body as { count: number; jobs: unknown[] }; + expect(payload.count).toBe(1); + }); + }); + + // ── Criterion 3: Queue health ───────────────────────────────────────────── + + describe('GET /api/notifications/health — queue health signal', () => { + it('returns 503 with an error message when the health monitor is not configured', async () => { + // The server was created without a healthMonitor option — the default. + const { status, body } = await makeRequest( + server, + 'GET', + '/api/notifications/health', + ); + + expect(status).toBe(503); + const payload = body as { error: string }; + expect(payload.error).toBeTruthy(); + }); + + it('returns 200 with a health report when the monitor is configured and healthy', async () => { + // Rebuild the server with a health monitor that has already run its first check. + await new Promise((resolve, reject) => + server.close((err) => (err ? reject(err) : resolve())), + ); + + const monitor = new NotificationHealthMonitor(null, null, { + intervalMs: 60_000, // don't auto-run after the initial check + repository, + }); + monitor.start(); + + server = createEventsServer({ + port: TEST_PORT, + stellarRpcUrl: 'https://soroban-testnet.stellar.org', + notificationAPI, + healthMonitor: monitor, + }); + await new Promise((resolve) => server.listen(TEST_PORT, resolve)); + + const { status, body } = await makeRequest( + server, + 'GET', + '/api/notifications/health', + ); + + monitor.stop(); + + expect(status).toBe(200); + const report = body as Record; + // Must expose overall status and a queue sub-section for health signal. + expect(report).toHaveProperty('status'); + expect(['healthy', 'degraded', 'unhealthy']).toContain(report.status); + expect(report).toHaveProperty('queue'); + const queue = report.queue as Record; + expect(queue).toHaveProperty('status'); + expect(queue).toHaveProperty('pendingJobs'); + expect(typeof queue.pendingJobs).toBe('number'); + }); + + it('reports deadLetterQueueDepth in the queue sub-section', async () => { + await new Promise((resolve, reject) => + server.close((err) => (err ? reject(err) : resolve())), + ); + + const monitor = new NotificationHealthMonitor(null, null, { + intervalMs: 60_000, + repository, + }); + monitor.start(); + + server = createEventsServer({ + port: TEST_PORT, + stellarRpcUrl: 'https://soroban-testnet.stellar.org', + notificationAPI, + healthMonitor: monitor, + }); + await new Promise((resolve) => server.listen(TEST_PORT, resolve)); + + const { status, body } = await makeRequest( + server, + 'GET', + '/api/notifications/health', + ); + + monitor.stop(); + + expect(status).toBe(200); + const report = body as Record; + const queue = report.queue as Record; + expect(queue).toHaveProperty('deadLetterQueueDepth'); + expect(typeof queue.deadLetterQueueDepth).toBe('number'); + }); + }); +}); diff --git a/listener/src/services/notification-api.ts b/listener/src/services/notification-api.ts index 107700b..ce5c27b 100644 --- a/listener/src/services/notification-api.ts +++ b/listener/src/services/notification-api.ts @@ -127,6 +127,16 @@ export class NotificationAPI { return await this.repository.getStats(); } + /** + * List pending jobs for queue visibility. + * Returns jobs currently waiting in the queue with their id, type, + * enqueue time (createdAt), scheduled delivery time (executeAt), + * priority, and retry count. + */ + async getPendingJobs(limit?: number) { + return await this.repository.getPendingJobs(limit); + } + /** * Get execution metrics with deduplication * Use this for dashboard metrics to prevent double-counting retried notifications diff --git a/listener/src/services/scheduled-notification-repository.ts b/listener/src/services/scheduled-notification-repository.ts index 765a9a8..2be5b75 100644 --- a/listener/src/services/scheduled-notification-repository.ts +++ b/listener/src/services/scheduled-notification-repository.ts @@ -565,6 +565,50 @@ export class ScheduledNotificationRepository { return result.changes; } + /** + * List pending jobs for queue visibility. + * Returns notifications in PENDING status ordered by priority then enqueue + * time (execute_at), with each entry carrying the minimum fields needed for + * queue inspection: id, notificationType, createdAt (enqueue time), + * executeAt (scheduled delivery time), priority, and retryCount. + */ + async getPendingJobs(limit: number = 100): Promise< + Array<{ + id: number; + notificationType: string; + createdAt: string; + executeAt: string; + priority: number; + retryCount: number; + }> + > { + const sql = ` + SELECT id, notification_type, created_at, execute_at, priority, retry_count + FROM scheduled_notifications + WHERE status = ? + ORDER BY priority ASC, execute_at ASC + LIMIT ? + `; + + const rows = await this.db.all<{ + id: number; + notification_type: string; + created_at: string; + execute_at: string; + priority: number; + retry_count: number; + }>(sql, [NotificationStatus.PENDING, limit]); + + return rows.map((row) => ({ + id: row.id, + notificationType: row.notification_type, + createdAt: row.created_at, + executeAt: row.execute_at, + priority: row.priority, + retryCount: row.retry_count, + })); + } + /** * Get statistics about scheduled notifications */