feat: harden horizon sync parsing

src/services/eventParser.ts

@@ -7,6 +7,8 @@
   ValidationEventPayload 
 } from '../types/horizonSync.js'
 
+type DecodedPayload = Record<string, unknown>
+
 /**
  * Result of parsing a Horizon event
  */
@@ -39,6 +41,52 @@
   txHash: string
 }
 
+function decodePayloadRecord(xdrData: string): DecodedPayload | null {
+  const candidates = [xdrData]
+
+  try {
+    const decoded = Buffer.from(xdrData, 'base64').toString('utf8')
+    if (decoded && decoded !== xdrData) {
+      candidates.push(decoded)
+    }
+  } catch {
+    // Ignore invalid base64 and fall back to direct JSON parsing.
+  }
+
+  for (const candidate of candidates) {
+    try {
+      const parsed = JSON.parse(candidate) as unknown
+      if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
+        return parsed as DecodedPayload
+      }
+    } catch {
+      // Try the next candidate.
+    }
+  }
+
+  return null
+}
+
+function readStringField(record: DecodedPayload, key: string): string | undefined {
+  const value = record[key]
+  return typeof value === 'string' ? value : undefined
+}
+
+function readDateField(record: DecodedPayload, key: string): Date | undefined {
+  const value = record[key]
+
+  if (value instanceof Date) {
+    return Number.isNaN(value.getTime()) ? undefined : value
+  }
+
+  if (typeof value === 'string' || typeof value === 'number') {
+    const parsed = new Date(value)
+    return Number.isNaN(parsed.getTime()) ? undefined : parsed
+  }
+
+  return undefined
+}
+
 /**
  * Validates vault_created event payload
  * 
@@ -148,17 +196,20 @@
           console.error(`Vault created validation error: ${createdError}`)
           return null
         }
-        return payload
-
-      case 'vault_completed':
-      case 'vault_failed':
-      case 'vault_cancelled':
-        payload = {
-          vaultId,
-          status: eventType.replace('vault_', '') as 'completed' | 'failed' | 'cancelled'
-        }
-
-        // Validate vault status payload
+      }
+
+      return payload
+
+    case 'vault_completed':
+    case 'vault_failed':
+    case 'vault_cancelled':
+      payload = {
+        vaultId: readStringField(decoded, 'vaultId') ?? '',
+        status: ((readStringField(decoded, 'status') ??
+          eventType.replace('vault_', '')) as VaultEventPayload['status'])
+      }
+
+      {
         const statusError = validateVaultStatusPayload(payload)
         if (statusError) {
           console.error(`Vault status validation error: ${statusError}`)
@@ -166,7 +217,7 @@
         }
         return payload
 
       default:
         return null
     }
   } catch (error) {
@@ -248,6 +299,23 @@
     console.error('Error parsing milestone payload XDR:', error)
     return null
   }
+
+  const payload: MilestoneEventPayload = {
+    milestoneId: readStringField(decoded, 'milestoneId') ?? '',
+    vaultId: readStringField(decoded, 'vaultId') ?? '',
+    title: readStringField(decoded, 'title') ?? '',
+    description: readStringField(decoded, 'description') ?? '',
+    targetAmount: readStringField(decoded, 'targetAmount') ?? '',
+    deadline: readDateField(decoded, 'deadline') ?? new Date('invalid')
+  }
+
+  const error = validateMilestonePayload(payload)
+  if (error) {
+    console.error(`Milestone validation error: ${error}`)
+    return null
+  }
+
+  return payload
 }
 
 /**
@@ -323,6 +391,8 @@
     console.error('Error parsing validation payload XDR:', error)
     return null
   }
+
+  return payload
 }
 
 /**

src/services/eventProcessor.ts

@@ -1,6 +1,6 @@
 import { Knex } from 'knex'
 import { ParsedEvent, ProcessorConfig, VaultEventPayload, MilestoneEventPayload, ValidationEventPayload } from '../types/horizonSync.js'
-import { retryWithBackoff, DEFAULT_RETRY_CONFIG } from '../utils/retry.js'
+import { retryWithBackoff, isRetryable } from '../utils/retry.js'
 import { createAuditLog } from '../lib/audit-logs.js'
 import { IdempotencyService } from './idempotency.js'
 
@@ -74,8 +74,9 @@ export class EventProcessor {
         eventId: event.eventId
       }
     } catch (error) {
-      retryCount = this.config.maxRetries
       const errorMessage = error instanceof Error ? error.message : 'Unknown error'
+      const retryable = error instanceof Error ? isRetryable(error) : false
+      retryCount = retryable ? this.config.maxRetries : 0
       const processingDurationMs = Date.now() - startTime
 
       // Create audit log for failed processing
@@ -94,8 +95,10 @@ export class EventProcessor {
         }
       })
 
-      // Move to dead letter queue if retries exhausted
-      await this.moveToDeadLetterQueue(event, errorMessage, retryCount)
+      // Only retryable failures that exhaust retries should be dead-lettered.
+      if (retryable) {
+        await this.moveToDeadLetterQueue(event, errorMessage, retryCount)
+      }
 
       return {
         success: false,