Validate Stellar RPC responses before event-processing pipeline (#630)

listener/src/services/event-subscriber.ts

@@ -8,6 +8,7 @@ import {
   getEventName,
   matchesEventFilter,
   validateEventPayload,
+  validateRpcResponse,
 } from '../utils/event-utils';
 import { DiscordNotificationService } from './discord-notification';
 import { NotificationRetryQueue } from './notification-retry-queue';
@@ -99,6 +100,17 @@ export class EventSubscriber {
     for (const contractConfig of this.config.contractAddresses) {
       try {
         const response = await this.getContractEvents(contractConfig);
+
+        const responseValidation = validateRpcResponse(response);
+        if (!responseValidation.valid) {
+          logger.error('Rejecting invalid RPC response, skipping contract', {
+            requestId,
+            contractAddress: contractConfig.address,
+            reason: responseValidation.reason,
+          });
+          continue;
+        }
+
         const events = response.events || [];
 
         // Detect potential reorg if events exist and we have previous state

listener/src/utils/event-utils.test.ts

@@ -3,6 +3,7 @@ import {
   getEventName,
   matchesEventFilter,
   validateEventPayload,
+  validateRpcResponse,
 } from './event-utils';
 
 function createValidEvent(overrides: Record<string, unknown> = {}) {
@@ -21,6 +22,24 @@ function createValidEvent(overrides: Record<string, unknown> = {}) {
   };
 }
 
+function createValidRpcResponse(overrides: Record<string, unknown> = {}) {
+  return {
+    latestLedger: 123,
+    events: [
+      {
+        id: 'event-1',
+        type: 'contract',
+        ledger: 100,
+        txHash: 'hash',
+        topic: [],
+        value: 1,
+      },
+    ],
+    cursor: 'cursor-1',
+    ...overrides,
+  };
+}
+
 describe('event-utils', () => {
   describe('validateEventPayload', () => {
     it('accepts a complete event payload', () => {
@@ -93,4 +112,53 @@ describe('event-utils', () => {
       expect(matchesEventFilter(null, ['TaskCreated'])).toBe(false);
     });
   });
+
+  describe('validateRpcResponse', () => {
+    it('accepts a complete RPC response', () => {
+      expect(validateRpcResponse(createValidRpcResponse() as any)).toEqual({
+        valid: true,
+      });
+    });
+
+    it('rejects a null response', () => {
+      const result = validateRpcResponse(null);
+      expect(result.valid).toBe(false);
+      expect(result.reason).toMatch(/missing/i);
+    });
+
+    it('rejects a non-object response', () => {
+      const result = validateRpcResponse('not-an-object' as any);
+      expect(result.valid).toBe(false);
+      expect(result.reason).toMatch(/not an object/i);
+    });
+
+    it('rejects a response missing the events field', () => {
+      const result = validateRpcResponse(
+        createValidRpcResponse({ events: undefined }) as any
+      );
+      expect(result.valid).toBe(false);
+      expect(result.reason).toMatch(/events/i);
+    });
+
+    it('rejects a response whose events field is not an array', () => {
+      const result = validateRpcResponse(
+        createValidRpcResponse({ events: 'not-an-array' }) as any
+      );
+      expect(result.valid).toBe(false);
+      expect(result.reason).toMatch(/events field is not an array/i);
+    });
+
+    it('rejects a response whose cursor is not a string', () => {
+      const result = validateRpcResponse(
+        createValidRpcResponse({ cursor: 123 }) as any
+      );
+      expect(result.valid).toBe(false);
+      expect(result.reason).toMatch(/cursor/i);
+    });
+
+    it('accepts a response without a cursor', () => {
+      const { cursor, ...rest } = createValidRpcResponse();
+      expect(validateRpcResponse(rest as any)).toEqual({ valid: true });
+    });
+  });
 });

listener/src/utils/event-utils.ts

@@ -5,6 +5,33 @@ export interface EventValidationResult {
   reason?: string;
 }
 
+export interface RpcResponseValidationResult {
+  valid: boolean;
+  reason?: string;
+}
+
+export function validateRpcResponse(
+  response: StellarSDK.rpc.Api.GetEventsResponse | null | undefined
+): RpcResponseValidationResult {
+  if (!response || typeof response !== 'object') {
+    return { valid: false, reason: 'RPC response is missing or not an object' };
+  }
+
+  if (response.events === undefined || response.events === null) {
+    return { valid: false, reason: 'RPC response is missing the events field' };
+  }
+
+  if (!Array.isArray(response.events)) {
+    return { valid: false, reason: 'RPC response events field is not an array' };
+  }
+
+  if (response.cursor !== undefined && typeof response.cursor !== 'string') {
+    return { valid: false, reason: 'RPC response cursor field is not a string' };
+  }
+
+  return { valid: true };
+}
+
 export function validateEventPayload(
   event: StellarSDK.rpc.Api.EventResponse
 ): EventValidationResult {