feat(listener): add notification retry queue with exponential backoff

listener/src/index.ts

@@ -27,6 +27,10 @@ function loadConfig(): Config {
     eventsApiPort: parseInt(process.env.EVENTS_API_PORT || '8787'),
     eventsApiCorsOrigin: process.env.EVENTS_API_CORS_ORIGIN || 'http://localhost:5173',
     discord,
+    retryQueue: {
+      baseDelayMs: parseInt(process.env.RETRY_BASE_DELAY_MS || '5000'),
+      maxRetries: parseInt(process.env.RETRY_MAX_RETRIES || '5'),
+    },
   };
 }
 

listener/src/services/discord-notification.ts

@@ -75,9 +75,6 @@ export class DiscordNotificationService {
       }
 
       this.deduplicator.markSent(fingerprint);
-      logger.info('Discord notification sent successfully', {
-        eventId: event.id,
-        contractAddress: contractConfig.address,
       logger.info('Discord notification delivered', {
         ...logContext,
         durationMs,

listener/src/services/event-subscriber.test.ts

@@ -535,7 +535,7 @@ describe('EventSubscriber', () => {
       await (subscriber as any).checkForEvents();
 
       expect(mockLogger.warn).toHaveBeenCalledWith(
-        'Failed to send Discord notification, event will still be processed',
+        'Discord notification failed, adding to retry queue',
         expect.objectContaining({ eventId: 'event-1' })
       );
     });

listener/src/services/event-subscriber.ts

@@ -9,6 +9,7 @@ import {
   validateEventPayload,
 } from '../utils/event-utils';
 import { DiscordNotificationService } from './discord-notification';
+import { NotificationRetryQueue } from './notification-retry-queue';
 
 export class EventSubscriber {
   private config: Config;
@@ -17,23 +18,31 @@ export class EventSubscriber {
   private reconnectAttempts: number = 0;
   private lastCursors: Map<string, string> = new Map();
   private discordService: DiscordNotificationService | null = null;
+  private retryQueue: NotificationRetryQueue | null = null;
 
   constructor(config: Config) {
     this.config = config;
     this.server = new StellarSDK.rpc.Server(config.stellarRpcUrl);
     if (config.discord) {
       this.discordService = new DiscordNotificationService(config.discord);
+      this.retryQueue = new NotificationRetryQueue(
+        (event, contractConfig, requestId) =>
+          this.discordService!.sendEventNotification(event, contractConfig, requestId),
+        config.retryQueue
+      );
     }
   }
 
   async start(): Promise<void> {
     this.isRunning = true;
     logger.info('Starting event subscriber service');
+    this.retryQueue?.start();
     this.poll();
   }
 
   async stop(): Promise<void> {
     this.isRunning = false;
+    this.retryQueue?.stop();
     logger.info('Stopping event subscriber service');
   }
 
@@ -196,11 +205,12 @@ export class EventSubscriber {
         contractConfig,
         requestId
       );
-      if (!success) {
-        logger.warn('Failed to send Discord notification, event will still be processed', {
+      if (!success && this.retryQueue) {
+        logger.warn('Discord notification failed, adding to retry queue', {
           requestId,
           eventId: event.id,
         });
+        this.retryQueue.enqueue(event, contractConfig, requestId);
       }
     }
 

listener/src/services/notification-retry-queue.test.ts

@@ -0,0 +1,279 @@
+import { xdr } from '@stellar/stellar-sdk';
+import * as StellarSDK from '@stellar/stellar-sdk';
+import { NotificationRetryQueue, NotificationFn } from './notification-retry-queue';
+
+jest.mock('../utils/logger', () => ({
+  __esModule: true,
+  default: {
+    info: jest.fn(),
+    warn: jest.fn(),
+    error: jest.fn(),
+  },
+}));
+
+function createMockEvent(
+  overrides: Partial<StellarSDK.rpc.Api.EventResponse> = {}
+): StellarSDK.rpc.Api.EventResponse {
+  return {
+    id: 'event-123',
+    type: 'contract',
+    ledger: 1000,
+    ledgerClosedAt: '2026-01-01T00:00:00Z',
+    transactionIndex: 1,
+    operationIndex: 0,
+    inSuccessfulContractCall: true,
+    txHash: 'abc123',
+    topic: [xdr.ScVal.scvSymbol('test_event')],
+    value: xdr.ScVal.scvString('test value'),
+    ...overrides,
+  };
+}
+
+const mockContractConfig = { address: 'CA123', events: ['test_event'] };
+
+describe('NotificationRetryQueue', () => {
+  beforeEach(() => {
+    jest.useFakeTimers();
+    jest.clearAllMocks();
+  });
+
+  afterEach(() => {
+    jest.useRealTimers();
+  });
+
+  describe('enqueue', () => {
+    it('adds an item to the queue', () => {
+      const notificationFn: NotificationFn = jest.fn();
+      const queue = new NotificationRetryQueue(notificationFn, { baseDelayMs: 1000 });
+
+      queue.enqueue(createMockEvent(), mockContractConfig);
+
+      expect(queue.size()).toBe(1);
+    });
+
+    it('logs when an item is queued', () => {
+      const logger = jest.requireMock('../utils/logger').default;
+      const notificationFn: NotificationFn = jest.fn();
+      const queue = new NotificationRetryQueue(notificationFn, { baseDelayMs: 1000 });
+
+      queue.enqueue(createMockEvent({ id: 'evt-q' }), mockContractConfig, 'req-1');
+
+      expect(logger.info).toHaveBeenCalledWith(
+        'Notification queued for retry',
+        expect.objectContaining({ eventId: 'evt-q', requestId: 'req-1' })
+      );
+    });
+  });
+
+  describe('processQueue', () => {
+    it('retries a notification after the base delay', async () => {
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(true);
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 1000,
+        processIntervalMs: 100,
+      });
+      queue.start();
+
+      queue.enqueue(createMockEvent(), mockContractConfig);
+
+      // Before delay expires — should not have retried yet
+      jest.advanceTimersByTime(500);
+      await Promise.resolve();
+      expect(notificationFn).not.toHaveBeenCalled();
+
+      // After delay expires — should retry
+      jest.advanceTimersByTime(600);
+      await Promise.resolve();
+      await Promise.resolve();
+      expect(notificationFn).toHaveBeenCalledTimes(1);
+
+      queue.stop();
+    });
+
+    it('removes the item from the queue on success', async () => {
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(true);
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 100,
+        processIntervalMs: 50,
+      });
+      queue.start();
+
+      queue.enqueue(createMockEvent(), mockContractConfig);
+
+      jest.advanceTimersByTime(200);
+      await Promise.resolve();
+      await Promise.resolve();
+
+      expect(queue.size()).toBe(0);
+      queue.stop();
+    });
+
+    it('logs success on a successful retry', async () => {
+      const logger = jest.requireMock('../utils/logger').default;
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(true);
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 100,
+        processIntervalMs: 50,
+      });
+      queue.start();
+
+      queue.enqueue(createMockEvent({ id: 'evt-ok' }), mockContractConfig, 'req-ok');
+      jest.advanceTimersByTime(200);
+      await Promise.resolve();
+      await Promise.resolve();
+
+      expect(logger.info).toHaveBeenCalledWith(
+        'Retry succeeded',
+        expect.objectContaining({ eventId: 'evt-ok', attempt: 1 })
+      );
+      queue.stop();
+    });
+  });
+
+  describe('exponential backoff', () => {
+    it('doubles the delay on each successive failure', async () => {
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(false);
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 1000,
+        maxRetries: 5,
+        processIntervalMs: 100,
+      });
+      queue.start();
+
+      queue.enqueue(createMockEvent(), mockContractConfig);
+
+      // Trigger attempt 1 (after 1000 ms base delay)
+      jest.advanceTimersByTime(1100);
+      await Promise.resolve();
+      await Promise.resolve();
+      expect(notificationFn).toHaveBeenCalledTimes(1);
+
+      // Trigger attempt 2 (after 2000 ms from attempt 1)
+      jest.advanceTimersByTime(2100);
+      await Promise.resolve();
+      await Promise.resolve();
+      expect(notificationFn).toHaveBeenCalledTimes(2);
+
+      queue.stop();
+    });
+
+    it('logs a warning with the next retry delay on failure', async () => {
+      const logger = jest.requireMock('../utils/logger').default;
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(false);
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 1000,
+        maxRetries: 3,
+        processIntervalMs: 100,
+      });
+      queue.start();
+
+      queue.enqueue(createMockEvent({ id: 'evt-backoff' }), mockContractConfig);
+
+      jest.advanceTimersByTime(1100);
+      await Promise.resolve();
+      await Promise.resolve();
+
+      expect(logger.warn).toHaveBeenCalledWith(
+        'Retry failed, scheduling next attempt',
+        expect.objectContaining({ eventId: 'evt-backoff', attempt: 1, delayMs: 2000 })
+      );
+      queue.stop();
+    });
+  });
+
+  describe('max retries', () => {
+    it('stops retrying after maxRetries attempts', async () => {
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(false);
+      const maxRetries = 3;
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 100,
+        maxRetries,
+        processIntervalMs: 50,
+      });
+      queue.start();
+      queue.enqueue(createMockEvent(), mockContractConfig);
+
+      const flush = async () => {
+        for (let i = 0; i < 5; i++) await Promise.resolve();
+      };
+
+      // attempt 1 fires at t=100ms (base delay)
+      jest.advanceTimersByTime(100);
+      await flush();
+      expect(notificationFn).toHaveBeenCalledTimes(1);
+
+      // attempt 2 fires at t=300ms (100 + 100*2^1 = 300)
+      jest.advanceTimersByTime(200);
+      await flush();
+      expect(notificationFn).toHaveBeenCalledTimes(2);
+
+      // attempt 3 fires at t=700ms (300 + 100*2^2 = 700)
+      jest.advanceTimersByTime(400);
+      await flush();
+      expect(notificationFn).toHaveBeenCalledTimes(maxRetries);
+      expect(queue.size()).toBe(0);
+
+      queue.stop();
+    });
+
+    it('logs an error when the notification permanently fails', async () => {
+      const logger = jest.requireMock('../utils/logger').default;
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(false);
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 100,
+        maxRetries: 1,
+        processIntervalMs: 50,
+      });
+      queue.start();
+
+      queue.enqueue(createMockEvent({ id: 'evt-dead' }), mockContractConfig, 'req-dead');
+
+      jest.advanceTimersByTime(200);
+      await Promise.resolve();
+      await Promise.resolve();
+
+      expect(logger.error).toHaveBeenCalledWith(
+        'Notification permanently failed after max retries',
+        expect.objectContaining({ eventId: 'evt-dead', totalAttempts: 1 })
+      );
+      queue.stop();
+    });
+  });
+
+  describe('start / stop', () => {
+    it('does not process items when stopped', async () => {
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(true);
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 100,
+        processIntervalMs: 50,
+      });
+
+      queue.enqueue(createMockEvent(), mockContractConfig);
+      // Never call queue.start()
+
+      jest.advanceTimersByTime(1000);
+      await Promise.resolve();
+
+      expect(notificationFn).not.toHaveBeenCalled();
+    });
+
+    it('calling start twice does not double-process items', async () => {
+      const notificationFn: NotificationFn = jest.fn().mockResolvedValue(true);
+      const queue = new NotificationRetryQueue(notificationFn, {
+        baseDelayMs: 100,
+        processIntervalMs: 50,
+      });
+      queue.start();
+      queue.start(); // second call should be a no-op
+
+      queue.enqueue(createMockEvent(), mockContractConfig);
+
+      jest.advanceTimersByTime(200);
+      await Promise.resolve();
+      await Promise.resolve();
+
+      expect(notificationFn).toHaveBeenCalledTimes(1);
+      queue.stop();
+    });
+  });
+});