feat(listener): add configurable event retention policy (#626)

ENVIRONMENT_VARIABLES_AND_SECRETS.md

@@ -128,6 +128,7 @@ Both variables must be provided together or neither.
 | `POLL_INTERVAL_MS` | `30000` | No | How often the listener polls Stellar for new contract events (ms). |
 | `MAX_RECONNECT_ATTEMPTS` | `5` | No | Maximum number of reconnect attempts when the RPC endpoint fails. |
 | `RECONNECT_DELAY_MS` | `5000` | No | Delay between reconnect attempts (ms). |
+| `PROCESSED_EVENT_RETENTION_MS` | `2592000000` (30 days) | No | How long processed event metadata is retained for persistent deduplication and operations. Expired records are removed during database cleanup; minimum `60000` ms. |
 
 ### 2.10 Retry queue (in-memory)
 

listener/.env.example

@@ -120,6 +120,10 @@ MAX_RECONNECT_ATTEMPTS=5
 # Delay between reconnect attempts (ms).
 RECONNECT_DELAY_MS=5000
 
+# How long to retain processed event metadata for persistent deduplication (ms).
+# Expired records are removed during database cleanup. Default: 30 days.
+PROCESSED_EVENT_RETENTION_MS=2592000000
+
 # -----------------------------------------------------------------------------
 # Retry Queue (in-memory, fast retries)
 # -----------------------------------------------------------------------------

listener/src/config.test.ts

@@ -1,4 +1,4 @@
-import { ConfigError, loadConfig } from './config';
+import { ConfigError, loadConfig, validateConfig } from './config';
 
 describe('Config validation', () => {
   const originalEnv = process.env;
@@ -98,11 +98,36 @@ describe('Config validation', () => {
         notificationRetentionMs: 604800000,
         rateLimitEventRetentionMs: 86400000,
         eventRetentionMs: 86400000,
+        processedEventRetentionMs: 2592000000,
         executionLogRetentionMs: 7776000000,
       },
     });
   });
 
+  it('loads a configured processed event retention duration', () => {
+    process.env.PROCESSED_EVENT_RETENTION_MS = '3600000';
+
+    expect(loadConfig().cleanup?.processedEventRetentionMs).toBe(3600000);
+  });
+
+  it('rejects invalid processed event retention configuration', () => {
+    process.env.PROCESSED_EVENT_RETENTION_MS = 'not-a-duration';
+
+    expect(() => loadConfig()).toThrow(
+      'PROCESSED_EVENT_RETENTION_MS must be a valid integer, got "not-a-duration"'
+    );
+  });
+
+  it('rejects processed event retention shorter than one minute', () => {
+    process.env.PROCESSED_EVENT_RETENTION_MS = '59999';
+
+    const config = loadConfig();
+
+    expect(() => validateConfig(config)).toThrow(
+      'PROCESSED_EVENT_RETENTION_MS must be >= 60000 ms (received: 59999).'
+    );
+  });
+
   it('loads notification deduplication settings when Discord is configured', () => {
     process.env.DISCORD_WEBHOOK_URL = 'https://discord.com/api/webhooks/123/abc';
     process.env.DISCORD_WEBHOOK_ID = '123';

listener/src/config.ts

@@ -159,6 +159,10 @@ function loadCleanupConfig(): AppCleanupConfig {
     notificationRetentionMs: parseIntegerEnv('NOTIFICATION_RETENTION_MS', String(7 * 24 * 60 * 60 * 1000)),
     rateLimitEventRetentionMs: parseIntegerEnv('RATE_LIMIT_EVENT_RETENTION_MS', String(24 * 60 * 60 * 1000)),
     eventRetentionMs: parseIntegerEnv('EVENT_RETENTION_MS', String(24 * 60 * 60 * 1000)),
+    processedEventRetentionMs: parseIntegerEnv(
+      'PROCESSED_EVENT_RETENTION_MS',
+      String(30 * 24 * 60 * 60 * 1000),
+    ),
     executionLogRetentionMs: parseIntegerEnv(
       'EXECUTION_LOG_RETENTION_MS',
       String(90 * 24 * 60 * 60 * 1000),
@@ -465,6 +469,12 @@ export function validateConfig(config: Config): void {
           `(received: ${config.cleanup.notificationRetentionMs}).`,
       );
     }
+    if (config.cleanup.processedEventRetentionMs < 60_000) {
+      errors.push(
+        `PROCESSED_EVENT_RETENTION_MS must be >= 60000 ms ` +
+          `(received: ${config.cleanup.processedEventRetentionMs}).`,
+      );
+    }
   }
 
   if (errors.length > 0) {

listener/src/services/cleanup-service.test.ts

@@ -98,8 +98,8 @@ describe('CleanupService', () => {
 
     const result = await service.runDbCleanup();
 
-    // Three DELETE calls: scheduled_notifications + rate_limit_events + execution_log
-    expect(db.run).toHaveBeenCalledTimes(3);
+    // Four DELETE calls, including processed event metadata.
+    expect(db.run).toHaveBeenCalledTimes(4);
     expect(db.run).toHaveBeenCalledWith(
       expect.stringContaining('DELETE FROM scheduled_notifications'),
       expect.any(Array),
@@ -112,9 +112,29 @@ describe('CleanupService', () => {
       expect.stringContaining('DELETE FROM notification_execution_log'),
       expect.any(Array),
     );
+    expect(db.run).toHaveBeenCalledWith(
+      expect.stringContaining('DELETE FROM processed_events'),
+      expect.any(Array),
+    );
     expect(result.notifications).toBe(5);
     expect(result.rateLimitEvents).toBe(5);
     expect(result.executionLogs).toBe(5);
+    expect(result.processedEvents).toBe(5);
+  });
+
+  it('uses the configured retention duration for processed event metadata', async () => {
+    const db = makeDb();
+    const registry = new EventRegistry();
+    const service = new CleanupService(db, registry, {
+      processedEventRetentionMs: 60_000,
+    });
+
+    await service.runDbCleanup();
+
+    expect(db.run).toHaveBeenCalledWith(
+      expect.stringContaining('DELETE FROM processed_events'),
+      [60],
+    );
   });
 
   it('stop clears the interval and stops registry cleanup', async () => {

listener/src/services/cleanup-service.ts

@@ -11,13 +11,16 @@ export interface CleanupConfig {
   rateLimitEventRetentionMs: number;
   /** Retain notification execution log rows for this long (ms). Default: 90 days. */
   executionLogRetentionMs: number;
+  /** Retain processed event metadata for this long (ms). Default: 30 days. */
+  processedEventRetentionMs: number;
 }
 
 const DEFAULTS: CleanupConfig = {
   intervalMs: 60 * 60 * 1000,
   notificationRetentionMs: 7 * 24 * 60 * 60 * 1000,
   rateLimitEventRetentionMs: 24 * 60 * 60 * 1000,
   executionLogRetentionMs: 90 * 24 * 60 * 60 * 1000,
+  processedEventRetentionMs: 30 * 24 * 60 * 60 * 1000,
 };
 
 export class CleanupService {
@@ -48,12 +51,18 @@ export class CleanupService {
     logger.info('CleanupService stopped');
   }
 
-  async runDbCleanup(): Promise<{ notifications: number; executionLogs: number; rateLimitEvents: number }> {
+  async runDbCleanup(): Promise<{
+    notifications: number;
+    executionLogs: number;
+    rateLimitEvents: number;
+    processedEvents: number;
+  }> {
     const notificationCutoff = new Date(Date.now() - this.config.notificationRetentionMs).toISOString();
     const rateLimitCutoff = new Date(Date.now() - this.config.rateLimitEventRetentionMs).toISOString();
     const executionLogCutoff = new Date(Date.now() - this.config.executionLogRetentionMs).toISOString();
+    const processedEventRetentionSeconds = this.config.processedEventRetentionMs / 1000;
 
-    const [notifResult, rateLimitResult, executionLogResult] = await Promise.all([
+    const [notifResult, rateLimitResult, executionLogResult, processedEventResult] = await Promise.all([
       this.db.run(
         `DELETE FROM scheduled_notifications
          WHERE status IN ('COMPLETED','FAILED','CANCELLED')
@@ -68,12 +77,18 @@ export class CleanupService {
         `DELETE FROM notification_execution_log WHERE execution_time < ?`,
         [executionLogCutoff],
       ),
+      this.db.run(
+        `DELETE FROM processed_events
+         WHERE processed_at < datetime('now', '-' || ? || ' seconds')`,
+        [processedEventRetentionSeconds],
+      ),
     ]);
 
     const result = {
       notifications: notifResult.changes,
       executionLogs: executionLogResult.changes,
       rateLimitEvents: rateLimitResult.changes,
+      processedEvents: processedEventResult.changes,
     };
 
     logger.info('DB cleanup completed', result);

listener/src/types/index.ts

@@ -93,6 +93,8 @@ export interface AppCleanupConfig {
   rateLimitEventRetentionMs: number;
   /** Retain in-memory events for this long (ms). */
   eventRetentionMs: number;
+  /** Retain processed event metadata for this long (ms). Default: 30 days. */
+  processedEventRetentionMs: number;
   /** Retain notification execution log rows for this long (ms). */
   executionLogRetentionMs: number;
 }