3-changes

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",

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<typeof searchNotifications>;
+
+jest.mock('../services/eventsApi', () => ({
+  searchNotifications: jest.fn(),
+}));
 jest.mock('../services/eventsApi', () => {
   const actual = jest.requireActual('../services/eventsApi') as typeof import('../services/eventsApi');
   return {

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<boolean> {
+    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);

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,

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) }));
     });
   });