diff --git a/dashboard/package.json b/dashboard/package.json index dff18de..3796ccd 100644 --- a/dashboard/package.json +++ b/dashboard/package.json @@ -3,6 +3,9 @@ "private": true, "version": "1.0.0", "type": "module", + "engines": { + "node": ">=22" + }, "scripts": { "build": "node ./node_modules/typescript/bin/tsc --noEmit && node ./node_modules/vite/bin/vite.js build", "preview": "node ./node_modules/vite/bin/vite.js preview", diff --git a/dashboard/src/pages/NotificationSearchPage.test.tsx b/dashboard/src/pages/NotificationSearchPage.test.tsx index d61fc62..5f642f4 100644 --- a/dashboard/src/pages/NotificationSearchPage.test.tsx +++ b/dashboard/src/pages/NotificationSearchPage.test.tsx @@ -4,6 +4,15 @@ import { NotificationSearchPage } from './NotificationSearchPage'; import { searchNotifications } from '../services/eventsApi'; import type { NotificationSearchResponse } from '../services/eventsApi'; +jest.mock('../services/eventsApi', () => ({ + searchNotifications: jest.fn(), +})); + +const mockedSearch = searchNotifications as jest.MockedFunction; + +jest.mock('../services/eventsApi', () => ({ + searchNotifications: jest.fn(), +})); jest.mock('../services/eventsApi', () => { const actual = jest.requireActual('../services/eventsApi') as typeof import('../services/eventsApi'); return { diff --git a/listener/src/services/discord-notification.ts b/listener/src/services/discord-notification.ts index 976ef5e..47b7924 100644 --- a/listener/src/services/discord-notification.ts +++ b/listener/src/services/discord-notification.ts @@ -6,6 +6,7 @@ import { NotificationDeduplicator, generateFingerprint } from './notification-de import { getNotificationAnalyticsAggregator, NotificationAnalyticsAggregator } from './notification-analytics-aggregator'; import { sendWebhook } from './webhook-sender'; import { NotificationType } from '../types/scheduled-notification'; +import { generateCorrelationId } from '../utils/request-id'; export const MAX_DISCORD_EMBED_LENGTH = 6000; export const MAX_DISCORD_FIELD_VALUE_LENGTH = 1024; @@ -31,7 +32,6 @@ export function createDiscordService(config: DiscordConfig): DiscordNotification } // --------------------------------------------------------------------------- -<<<<<<< HEAD // Discord content safety // --------------------------------------------------------------------------- @@ -58,7 +58,6 @@ export function sanitizeForDiscord(text: string): string { return text .replace(MENTION_PATTERN, '[mention removed]') .replace(MARKDOWN_CHARS, '\\$1'); -======= // Internal helpers // --------------------------------------------------------------------------- @@ -86,7 +85,6 @@ async function safeReadResponseBody(response: Response, maxLength = 300): Promis } catch { return null; } ->>>>>>> 5bc550e (fix(listener): improve Discord delivery failure logging) } export class DiscordNotificationService { @@ -111,6 +109,7 @@ export class DiscordNotificationService { contractConfig: ContractConfig, requestId?: string ): Promise { + const correlationId = requestId ?? generateCorrelationId(); const fingerprint = generateFingerprint(event.id, contractConfig.address); if (this.deduplicator.isDuplicate(fingerprint)) { @@ -124,13 +123,16 @@ export class DiscordNotificationService { logger.info('Skipping duplicate notification', { eventId: event.id, contractAddress: contractConfig.address, + requestId: correlationId, + correlationId, fingerprint, deduplication: this.deduplicator.getMetrics(), }); return true; } const logContext = { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, contractAddress: contractConfig.address, webhookId: this.config.webhookId, @@ -151,6 +153,12 @@ export class DiscordNotificationService { if (response.ok) { this.deduplicator.markSent(fingerprint); + logger.info('Discord notification sent successfully', { + eventId: event.id, + contractAddress: contractConfig.address, + requestId: correlationId, + correlationId, + }); logger.info('Discord notification delivered', { ...logContext, durationMs, @@ -195,9 +203,8 @@ export class DiscordNotificationService { const delayMs = Math.pow(2, attempt) * backoffBaseSeconds * 1000; logger.warn('Retrying Discord webhook', { ...logContext, - attempt: attempt + 1, - nextDelayMs: delayMs, - maxRetries, + delayMs, + attempt, }); await this.delay(delayMs); diff --git a/listener/src/services/event-processing-queue.ts b/listener/src/services/event-processing-queue.ts index e83cb16..6b0d4a1 100644 --- a/listener/src/services/event-processing-queue.ts +++ b/listener/src/services/event-processing-queue.ts @@ -1,6 +1,7 @@ import * as StellarSDK from '@stellar/stellar-sdk'; import { ContractConfig } from '../types'; import logger from '../utils/logger'; +import { generateCorrelationId } from '../utils/request-id'; export enum Priority { Low = 0, @@ -78,11 +79,13 @@ export class EventProcessingQueue { requestId?: string, priority: Priority = Priority.Medium ): boolean { + const correlationId = requestId ?? generateCorrelationId(); const fingerprint = buildEventFingerprint(event, contractConfig.address); if (this.queuedFingerprints.has(fingerprint)) { logger.info('Skipping duplicate event queue entry', { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, contractAddress: contractConfig.address, fingerprint, @@ -94,7 +97,8 @@ export class EventProcessingQueue { const nextRetryAt = Date.now() + delayMs; logger.info('Event queued for processing', { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, contractAddress: contractConfig.address, delayMs, @@ -107,7 +111,7 @@ export class EventProcessingQueue { this.queue.push({ event, contractConfig, - requestId: requestId ?? '', + requestId: correlationId, retryCount: 0, nextRetryAt, fingerprint, @@ -216,6 +220,7 @@ export class EventProcessingQueue { this.metrics.processingTimes.push(duration); logger.info('Event processing succeeded', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, }); @@ -232,6 +237,7 @@ export class EventProcessingQueue { this.metrics.processingTimes.push(duration); logger.error('Event processing permanently failed after max retries', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, totalAttempts: attempt, @@ -244,6 +250,7 @@ export class EventProcessingQueue { logger.warn('Event processing failed, scheduling retry', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, attempt, @@ -266,6 +273,7 @@ export class EventProcessingQueue { this.metrics.processingTimes.push(duration); logger.error('Event processing crashed after max retries', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, totalAttempts: attempt, @@ -279,6 +287,7 @@ export class EventProcessingQueue { logger.error('Event processing crashed, scheduling retry', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, attempt, diff --git a/listener/src/services/event-subscriber.test.ts b/listener/src/services/event-subscriber.test.ts index 2fd5bf2..3dc8215 100644 --- a/listener/src/services/event-subscriber.test.ts +++ b/listener/src/services/event-subscriber.test.ts @@ -156,6 +156,14 @@ describe('EventSubscriber', () => { type: 'contract', }) ); + expect(mockLogger.info).toHaveBeenCalledWith( + 'Event processing complete', + expect.objectContaining({ + eventId: 'event-abc', + outcome: 'success', + durationMs: expect.any(Number), + }) + ); }); it('processes each valid event in a batch', async () => { @@ -174,6 +182,67 @@ describe('EventSubscriber', () => { expect(countLogCalls('info', 'Processing event')).toBe(3); }); + it('continues processing valid events after a malformed event', async () => { + const malformedEvent = createMockEvent({ + id: 'event-malformed', + topic: [undefined as unknown as xdr.ScVal], + }); + mockGetEvents.mockResolvedValue({ + events: [malformedEvent, createMockEvent({ id: 'event-valid' })], + cursor: 'cursor-mixed', + }); + + const subscriber = new EventSubscriber(testConfig); + await expect((subscriber as any).checkForEvents()).resolves.toBeUndefined(); + + expect(countLogCalls('info', 'Processing event')).toBe(1); + expect(mockLogger.warn).toHaveBeenCalledWith( + 'Skipping malformed event', + expect.objectContaining({ + contractAddress: contractConfig.address, + eventId: 'event-malformed', + eventIndex: 0, + }) + ); + expect(mockLogger.info).toHaveBeenCalledWith( + 'Processing event', + expect.objectContaining({ eventId: 'event-valid' }) + ); + }); + + it('continues processing after an individual event-processing failure', async () => { + mockGetEvents.mockResolvedValue({ + events: [ + createMockEvent({ id: 'event-fails' }), + createMockEvent({ id: 'event-after-failure' }), + ], + cursor: 'cursor-processing-failure', + }); + + const subscriber = new EventSubscriber(testConfig); + const processEvent = jest + .spyOn(subscriber as any, 'processEvent') + .mockRejectedValueOnce(new Error('unexpected event shape')) + .mockResolvedValueOnce(true); + + await expect((subscriber as any).checkForEvents()).resolves.toBeUndefined(); + + expect(processEvent).toHaveBeenCalledTimes(2); + expect(mockLogger.warn).toHaveBeenCalledWith( + 'Event processing failed; continuing batch', + expect.objectContaining({ + eventId: 'event-fails', + error: 'unexpected event shape', + }) + ); + expect(processEvent).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ id: 'event-after-failure' }), + contractConfig, + expect.any(String) + ); + }); + it('does not log received events when RPC returns an empty list', async () => { mockGetEvents.mockResolvedValue({ events: [], cursor: 'cursor-empty' }); @@ -533,6 +602,11 @@ describe('EventSubscriber', () => { expect.any(Object), expect.any(String) ); + const correlationId = mockDiscordService.sendEventNotification.mock.calls[0][2]; + expect(mockLogger.info).toHaveBeenCalledWith( + 'Processing event', + expect.objectContaining({ correlationId }) + ); }); it('logs warning when Discord notification fails', async () => { @@ -563,6 +637,10 @@ describe('EventSubscriber', () => { 'Discord notification failed, adding to retry queue', expect.objectContaining({ eventId: 'event-1' }) ); + const failureLog = (mockLogger.warn as jest.Mock).mock.calls.find( + (call: unknown[]) => call[0] === 'Discord notification failed, adding to retry queue' + ); + expect(failureLog?.[1]).toEqual(expect.objectContaining({ correlationId: expect.any(String) })); }); }); diff --git a/listener/src/services/event-subscriber.ts b/listener/src/services/event-subscriber.ts index 54a902a..9134b25 100644 --- a/listener/src/services/event-subscriber.ts +++ b/listener/src/services/event-subscriber.ts @@ -3,7 +3,7 @@ import { Config, ContractConfig } from '../types'; import { eventRegistry } from '../store/event-registry'; import { preferenceStore } from '../store/preference-store'; import logger from '../utils/logger'; -import { generateRequestId } from '../utils/request-id'; +import { generateCorrelationId, generateRequestId } from '../utils/request-id'; import { getEventName, matchesEventFilter, @@ -155,6 +155,28 @@ export class EventSubscriber { } } + const processableEvents: Array<{ + event: StellarSDK.rpc.Api.EventResponse; + correlationId: string; + }> = []; + for (const [eventIndex, event] of events.entries()) { + const correlationId = generateCorrelationId(); + try { + if (this.shouldProcessEvent(event, contractConfig, requestId, correlationId)) { + processableEvents.push({ event, correlationId }); + } + } catch (error) { + logger.warn('Skipping malformed event', { + requestId, + correlationId, + contractAddress: contractConfig.address, + eventIndex, + eventId: event?.id, + eventType: event?.type, + error: error instanceof Error ? error.message : String(error), + }); + } + } const processableEvents = events.filter((event: StellarSDK.rpc.Api.EventResponse) => this.shouldProcessEvent(event, contractConfig, requestId) ); @@ -168,11 +190,24 @@ export class EventSubscriber { }); } - for (const event of processableEvents) { - if (this.eventQueue) { - this.eventQueue.enqueue(event, contractConfig, requestId); - } else { - await this.processEvent(event, contractConfig, requestId); + for (const [eventIndex, processableEvent] of processableEvents.entries()) { + const { event, correlationId } = processableEvent; + try { + if (this.eventQueue) { + this.eventQueue.enqueue(event, contractConfig, correlationId); + } else { + await this.processEvent(event, contractConfig, correlationId); + } + } catch (error) { + logger.warn('Event processing failed; continuing batch', { + requestId, + correlationId, + contractAddress: contractConfig.address, + eventIndex, + eventId: event?.id, + eventType: event?.type, + error: error instanceof Error ? error.message : String(error), + }); } } @@ -209,7 +244,8 @@ export class EventSubscriber { private shouldProcessEvent( event: StellarSDK.rpc.Api.EventResponse, contractConfig: ContractConfig, - requestId: string = '' + requestId: string = '', + correlationId: string = requestId ): boolean { // Check if event has expired if (this.expirationService && !this.expirationService.shouldProcess(event)) { @@ -230,6 +266,7 @@ export class EventSubscriber { if (!validation.valid) { logger.warn('Skipping invalid event payload', { requestId, + correlationId, contractAddress: contractConfig.address, eventId: event.id, reason: validation.reason, @@ -335,7 +372,8 @@ export class EventSubscriber { private async processEvent( event: StellarSDK.rpc.Api.EventResponse, contractConfig: ContractConfig, - requestId: string = '' + requestId: string = '', + correlationId: string = '' ): Promise { const eventStart = Date.now(); const eventName = getEventName(event.topic); @@ -345,7 +383,8 @@ export class EventSubscriber { const duplicate = await this.deduplicationService.isDuplicate(event.id, contractConfig.address); if (duplicate.isDuplicate) { logger.warn('Skipping event: already processed (persistent deduplication)', { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, contractAddress: contractConfig.address, isReorgDuplicate: duplicate.isReorgDuplicate, @@ -378,7 +417,8 @@ export class EventSubscriber { }); logger.info('Processing event', { - requestId, + requestId: correlationId, + correlationId, contractAddress: displayEvent.contractAddress, eventId: displayEvent.eventId, eventName: displayEvent.eventName, @@ -397,6 +437,7 @@ export class EventSubscriber { logger.info('Skipping Discord notification: category disabled by user preferences', { eventId: event.id, userId, + correlationId, }); } else { try { @@ -409,7 +450,8 @@ export class EventSubscriber { if (!success && this.retryQueue) { logger.warn('Discord notification failed, adding to retry queue', { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, }); this.retryQueue.enqueue(event, contractConfig, requestId); @@ -418,7 +460,8 @@ export class EventSubscriber { } catch (error) { processingError = error instanceof Error ? error.message : String(error); logger.error('Error sending Discord notification', { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, error: processingError, }); @@ -441,9 +484,11 @@ export class EventSubscriber { } logger.info('Event processing complete', { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, notificationSent, + outcome: !this.discordService || notificationSent ? 'success' : 'failure', durationMs: Date.now() - eventStart, }); diff --git a/listener/src/services/notification-retry-queue.ts b/listener/src/services/notification-retry-queue.ts index 3633fcd..765ca62 100644 --- a/listener/src/services/notification-retry-queue.ts +++ b/listener/src/services/notification-retry-queue.ts @@ -1,6 +1,7 @@ import * as StellarSDK from '@stellar/stellar-sdk'; import { ContractConfig } from '../types'; import logger from '../utils/logger'; +import { generateCorrelationId } from '../utils/request-id'; import { getEventName } from '../utils/event-utils'; import { getNotificationAnalyticsAggregator, NotificationAnalyticsAggregator } from './notification-analytics-aggregator'; import { NotificationType } from '../types/scheduled-notification'; @@ -85,11 +86,13 @@ export class NotificationRetryQueue { requestId?: string, priority: Priority = Priority.Medium ): void { + const correlationId = requestId ?? generateCorrelationId(); const fingerprint = buildRetryFingerprint(event, contractConfig.address); if (this.queuedFingerprints.has(fingerprint)) { logger.info('Skipping duplicate retry queue entry', { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, contractAddress: contractConfig.address, fingerprint, @@ -101,7 +104,8 @@ export class NotificationRetryQueue { const nextRetryAt = Date.now() + delayMs; logger.info('Notification queued for retry', { - requestId, + requestId: correlationId, + correlationId, eventId: event.id, contractAddress: contractConfig.address, delayMs, @@ -179,6 +183,7 @@ export class NotificationRetryQueue { logger.info('Retrying failed notification', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, attempt, @@ -210,6 +215,7 @@ export class NotificationRetryQueue { }); logger.info('Retry succeeded', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, attempt, @@ -232,6 +238,7 @@ export class NotificationRetryQueue { }); logger.error('Notification permanently failed after max retries', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, totalAttempts: attempt, @@ -244,6 +251,7 @@ export class NotificationRetryQueue { logger.warn('Retry failed, scheduling next attempt', { requestId: item.requestId, + correlationId: item.requestId, eventId: item.event.id, contractAddress: item.contractConfig.address, attempt, diff --git a/listener/src/utils/request-id.ts b/listener/src/utils/request-id.ts index 9450508..0a9c61f 100644 --- a/listener/src/utils/request-id.ts +++ b/listener/src/utils/request-id.ts @@ -10,6 +10,10 @@ export function generateRequestId(): string { } /** + * Generates a non-sensitive identifier for tracing one notification workflow. + */ +export function generateCorrelationId(): string { + return randomUUID(); * Client-supplied request IDs must be printable ASCII tokens of bounded length. * Rejects empty values, control characters, whitespace, and oversized strings * so untrusted header content is never reused as a log/trace key (#686).