feat(verification): rebuild payment verification as a typed multi-step pipeline

backend/src/services/payment-monitor.service.ts

@@ -3,12 +3,9 @@ import stellarService, { PaymentPageRecord, PaymentRecord } from './stellar.serv
 import invoiceService, { InvoiceService, Queryable } from './invoice.service';
 import { SELLER_PUBLIC_KEY, STELLAR_NETWORK } from '../config/stellar';
 import { pool } from '../config/database';
-import { checkInvoiceIsPayable, verifyHorizonPayment } from './payment-verification';
+import { checkInvoiceIsPayable, executePaymentVerificationPipeline } from './payment-verification';
 import { canonicalAmount } from '../utils/safe-amount-compare';
-import { PaymentClaimError } from '../domain/payment-attribution';
-import { IllegalStateTransitionError } from '../domain/invoice-lifecycle';
 import {
-  parseSettlementTime,
   SettlementTimeUnavailableError,
 } from '../domain/invoice-settlement';
 import { monitorBackoffMs } from '../utils/monitor-retry-backoff';
@@ -543,10 +540,12 @@ export class PaymentMonitorService {
       network: this.network,
     });
 
-    const verification = verifyHorizonPayment({
+    const pipeline = await executePaymentVerificationPipeline({
+      invoice,
       txHash: payment.txHash,
       network: this.network,
-      transaction: { memo: payment.memo, memo_type: payment.memoType },
+      expectedNetwork: this.network,
+      transaction: { memo: payment.memo, memo_type: payment.memoType, created_at: payment.createdAt },
       operations: [{
         type: 'payment',
         from: payment.from,
@@ -556,102 +555,77 @@ export class PaymentMonitorService {
         asset_code: isNative ? undefined : payment.assetCode,
         asset_issuer: payment.assetIssuer,
       }],
-      expected: {
-        memo: invoice.memo,
-        amount: invoice.amount,
-        destination: invoice.sellerPublicKey,
-        assetCode: invoice.assetCode,
-        assetIssuer: invoice.assetIssuer,
-        network: this.network,
+      storage: {
+        markAsPaid: async (id, txHash, payerPublicKey, payerInfo, options) => {
+          await this.saveTransaction(payment, id);
+          return this.invoices.markAsPaid(
+            id,
+            txHash,
+            payerPublicKey ?? payment.from,
+            payerInfo,
+            options
+          );
+        },
+        getInvoiceById: async (id) => {
+          const byMemo = await this.invoices.getInvoiceByMemo(invoice.memo);
+          return byMemo && byMemo.id === id ? byMemo : null;
+        },
       },
+      requireSettledAt: true,
     });
 
-    if (!verification.ok) {
+    if (!pipeline.ok) {
+      if (pipeline.code === 'TRANSACTION_CLOSE_TIME_UNAVAILABLE') {
+        throw new SettlementTimeUnavailableError();
+      }
       emitEvent('warn', 'payment.verify.rejected', context, {
         invoiceRef: logReference(invoice.id),
         txRef: logReference(payment.txHash),
-        errorCode: verification.code,
+        errorCode: pipeline.code,
         network: this.network,
         durationMs: Date.now() - startedAt,
       });
+      const transition = pipeline.details;
+      const lifecycleAudit =
+        pipeline.stage === 'persist_paid' &&
+        typeof transition?.lifecycleCode === 'string' &&
+        typeof transition?.fromStatus === 'string' &&
+        typeof transition?.toStatus === 'string'
+          ? {
+              code: transition.lifecycleCode,
+              fromStatus: transition.fromStatus,
+              toStatus: transition.toStatus,
+            }
+          : {};
       await this.invoices.logPaymentEvent(
         invoice.id,
-        verification.code === 'AMOUNT_TOO_LOW' || verification.code === 'AMOUNT_MISMATCH'
+        pipeline.code === 'AMOUNT_TOO_LOW' || pipeline.code === 'AMOUNT_MISMATCH'
           ? 'PARTIAL_PAYMENT'
           : 'PAYMENT_REJECTED',
         {
-          code: verification.code,
+          code: pipeline.code,
+          stage: pipeline.stage,
           txHash: payment.txHash,
           expectedAmount: canonicalAmount(invoice.amount) ?? String(invoice.amount),
           receivedAmount: payment.amount,
           payerPublicKey: payment.from,
+          ...lifecycleAudit,
         }
       );
       return;
     }
 
-    const settledAt = parseSettlementTime(payment.createdAt) ?? verification.value.settledAt;
-    if (!settledAt) {
-      throw new SettlementTimeUnavailableError();
-    }
-
-    try {
-      await this.saveTransaction(payment, invoice.id);
-      await this.invoices.markAsPaid(
-        invoice.id,
-        payment.txHash,
-        payment.from,
-        undefined,
-        { settledAt, destinationMuxedId: verification.value.toMuxedId }
-      );
-      this.processedTxHashes.add(payment.txHash);
-      this.unregisterWatch(invoice.id);
-      emitEvent('info', 'invoice.paid', context, {
-        invoiceRef: logReference(invoice.id),
-        sellerRef: logReference(invoice.sellerPublicKey),
-        txRef: logReference(payment.txHash),
-        assetCode: invoice.assetCode || 'XLM',
-        network: this.network,
-        storage: this.database ? 'postgres' : 'memory',
-        durationMs: Date.now() - startedAt,
-      });
-    } catch (error) {
-      if (error instanceof PaymentClaimError) {
-        // A transaction that already settled another invoice must not settle this one
-        emitEvent('warn', 'payment.verify.rejected', context, {
-          invoiceRef: logReference(invoice.id),
-          txRef: logReference(payment.txHash),
-          errorCode: error.code || 'TX_HASH_ALREADY_USED',
-          network: this.network,
-          durationMs: Date.now() - startedAt,
-        });
-        await this.invoices.logPaymentEvent(
-          invoice.id,
-          'PAYMENT_REJECTED',
-          {
-            code: 'TX_HASH_ALREADY_USED',
-            txHash: payment.txHash,
-            error: error.message,
-          }
-        );
-        return;
-      }
-      if (error instanceof IllegalStateTransitionError) {
-        // Status raced (e.g. paid elsewhere); keep the stream healthy.
-        console.warn(
-          'Payment monitor skipped settling invoice due to state conflict:',
-          error.message
-        );
-        await this.invoices.logPaymentEvent(invoice.id, 'PAYMENT_REJECTED', {
-          code: error.code,
-          txHash: payment.txHash,
-          fromStatus: error.fromStatus,
-          toStatus: error.toStatus,
-        });
-        return;
-      }
-      throw error;
-    }
+    this.processedTxHashes.add(payment.txHash);
+    this.unregisterWatch(invoice.id);
+    emitEvent('info', 'invoice.paid', context, {
+      invoiceRef: logReference(invoice.id),
+      sellerRef: logReference(invoice.sellerPublicKey),
+      txRef: logReference(payment.txHash),
+      assetCode: invoice.assetCode || 'XLM',
+      network: this.network,
+      storage: this.database ? 'postgres' : 'memory',
+      durationMs: Date.now() - startedAt,
+    });
   }
 
   private async saveTransaction(payment: PaymentRecord, invoiceId: string) {