Mastering Distributed Transactions in Payment Systems: A Deep Dive

class TwoPhaseCommitCoordinator:
    def __init__(self, participants):
        self.participants = participants
        self.transaction_id = self._generate_transaction_id()

    def _generate_transaction_id(self):
        # In a real system, this would be a robust unique ID generator
        import uuid
        return str(uuid.uuid4())

    def execute_transaction(self, operation_data):
        print(f"Coordinator: Starting transaction {self.transaction_id}")
        prepared_participants = []

        # Phase 1: Prepare
        try:
            for participant in self.participants:
                print(f"Coordinator: Sending PREPARE to {participant.name}")
                if not participant.prepare(self.transaction_id, operation_data):
                    raise Exception(f"Participant {participant.name} failed to prepare")
                prepared_participants.append(participant)
        except Exception as e:
            print(f"Coordinator: PREPARE FAILED. Initiating ABORT. Reason: {e}")
            # Rollback all prepared participants
            for p in prepared_participants:
                p.rollback(self.transaction_id)
            return False

        # Phase 2: Commit
        try:
            print(f"Coordinator: All participants prepared. Sending COMMIT.")
            for participant in self.participants:
                participant.commit(self.transaction_id)
            print(f"Coordinator: Transaction {self.transaction_id} committed successfully.")
            return True
        except Exception as e:
            print(f"Coordinator: COMMIT FAILED. Initiating ABORT. Reason: {e}")
            # This is tricky in 2PC. If commit fails for one, others might have committed.
            # This highlights 2PC's limitations.
            for p in prepared_participants: # Try to rollback, but might be too late
                p.rollback(self.transaction_id)
            return False

# Simplified Participant class
class Participant:
    def __init__(self, name):
        self.name = name
        self.state = {}

    def prepare(self, tx_id, data):
        # Simulate doing work and preparing for commit
        print(f"  {self.name}: Preparing transaction {tx_id} with data {data}...")
        # In a real system, would write to a local transaction log
        self.state[tx_id] = {'prepared_data': data, 'status': 'prepared'}
        return True # Or False if prepare fails

    def commit(self, tx_id):
        if self.state.get(tx_id, {}).get('status') == 'prepared':
            print(f"  {self.name}: Committing transaction {tx_id}.")
            self.state[tx_id]['status'] = 'committed'
            return True
        return False

    def rollback(self, tx_id):
        if tx_id in self.state:
            print(f"  {self.name}: Rolling back transaction {tx_id}.")
            del self.state[tx_id]
            return True
        return False

# Example Usage:
# p1 = Participant("AccountService")
# p2 = Participant("LedgerService")
# coordinator = TwoPhaseCommitCoordinator([p1, p2])
# coordinator.execute_transaction({"amount": 100, "from": "A", "to": "B"})

// payment.service.ts
import { kafkaProducer } from './kafka';
import { Database } from './database';

interface TransferRequest {
  transactionId: string;
  senderAccountId: string;
  recipientAccountId: string;
  amount: number;
}

interface FundsDebitedEvent {
  transactionId: string;
  senderAccountId: string;
  amount: number;
}

interface FundsCreditFailedEvent {
  transactionId: string;
  senderAccountId: string;
  reason: string;
}

export class PaymentService {
  constructor(private db: Database) {}

  async initiateTransfer(request: TransferRequest): Promise<void> {
    // Start local transaction
    await this.db.beginTransaction();
    try {
      // Debit sender's account
      await this.db.execute(`UPDATE accounts SET balance = balance - $1 WHERE id = $2`, [request.amount, request.senderAccountId]);
      // Record local transaction outcome
      await this.db.execute(`INSERT INTO transactions (id, type, status) VALUES ($1, 'debit', 'pending')`, [request.transactionId]);

      // Publish FundsDebitedEvent
      const event: FundsDebitedEvent = {
        transactionId: request.transactionId,
        senderAccountId: request.senderAccountId,
        amount: request.amount,
      };
      await kafkaProducer.send({
        topic: 'payment-events',
        messages: [{ value: JSON.stringify(event), key: event.transactionId }],
      });

      await this.db.commitTransaction();
      console.log(`Funds debited for ${request.transactionId}. Event published.`);
    } catch (error) {
      await this.db.rollbackTransaction();
      console.error(`Failed to debit funds for ${request.transactionId}:`, error);
      // Optionally publish a 'DebitFailedEvent' if needed for other services
      throw error;
    }
  }

  async compensateDebit(event: FundsCreditFailedEvent): Promise<void> {
    await this.db.beginTransaction();
    try {
      // Re-credit sender's account
      await this.db.execute(`UPDATE accounts SET balance = balance + $1 WHERE id = $2`, [event.amount, event.senderAccountId]);
      // Update transaction status
      await this.db.execute(`UPDATE transactions SET status = 'compensated' WHERE id = $1`, [event.transactionId]);
      await this.db.commitTransaction();
      console.log(`Compensation: Funds re-credited for ${event.transactionId}.`);
    } catch (error) {
      await this.db.rollbackTransaction();
      console.error(`Compensation failed for ${event.transactionId}:`, error);
      // Manual intervention might be needed
    }
  }
}

// ledger.service.ts
import { kafkaConsumer } from './kafka';

interface FundsCreditedEvent {
  transactionId: string;
  recipientAccountId: string;
  amount: number;
}

export class LedgerService {
  constructor(private db: Database) {
    kafkaConsumer.subscribe({ topic: 'payment-events', fromBeginning: false });
    kafkaConsumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        const event = JSON.parse(message.value.toString());
        if (event.type === 'FundsDebitedEvent') {
          await this.creditRecipient(event);
        } else if (event.type === 'FundsCreditFailedEvent') {
          // Optionally handle if ledger needs to react to compensation
        }
      },
    });
  }

  async creditRecipient(event: FundsDebitedEvent): Promise<void> {
    await this.db.beginTransaction();
    try {
      // Credit recipient's account
      await this.db.execute(`UPDATE accounts SET balance = balance + $1 WHERE id = $2`, [event.amount, event.recipientAccountId]);
      // Record local transaction outcome
      await this.db.execute(`UPDATE transactions SET status = 'completed' WHERE id = $1`, [event.transactionId]);

      // Publish FundsCreditedEvent
      const creditEvent: FundsCreditedEvent = {
        transactionId: event.transactionId,
        recipientAccountId: event.recipientAccountId,
        amount: event.amount,
      };
      await kafkaProducer.send({
        topic: 'payment-events',
        messages: [{ value: JSON.stringify(creditEvent), key: creditEvent.transactionId }],
      });

      await this.db.commitTransaction();
      console.log(`Funds credited for ${event.transactionId}. Event published.`);
    } catch (error) {
      await this.db.rollbackTransaction();
      console.error(`Failed to credit funds for ${event.transactionId}:`, error);
      // Publish FundsCreditFailedEvent for compensation
      const failureEvent: FundsCreditFailedEvent = {
        transactionId: event.transactionId,
        senderAccountId: event.senderAccountId, // This should come from the initial request or be passed along
        amount: event.amount,
        reason: error.message,
      };
      await kafkaProducer.send({
        topic: 'payment-events',
        messages: [{ value: JSON.stringify(failureEvent), key: failureEvent.transactionId }],
      });
    }
  }
}

// payment_saga_orchestrator.ts
import { kafkaProducer } from './kafka';
import { v4 as uuid } from 'uuid';

interface SagaState {
  transactionId: string;
  status: 'STARTED' | 'DEBITED' | 'CREDITED' | 'COMPLETED' | 'FAILED' | 'COMPENSATING';
  senderAccountId: string;
  recipientAccountId: string;
  amount: number;
  currentStep: number;
}

export class PaymentSagaOrchestrator {
  private sagaStates: Map<string, SagaState> = new Map(); // In-memory for example; use persistent storage in prod

  constructor() {
    // Subscribe to events from participant services
    // kafkaConsumer.subscribe to 'payment-service-events', 'ledger-service-events'
    // And handle incoming events to update sagaState and send next commands
  }

  async startPaymentTransfer(senderAccountId: string, recipientAccountId: string, amount: number): Promise<string> {
    const transactionId = uuid();
    const sagaState: SagaState = {
      transactionId,
      status: 'STARTED',
      senderAccountId,
      recipientAccountId,
      amount,
      currentStep: 0,
    };
    this.sagaStates.set(transactionId, sagaState);

    console.log(`Orchestrator: Starting saga ${transactionId}`);
    await this.sendDebitFundsCommand(sagaState);
    return transactionId;
  }

  private async sendDebitFundsCommand(sagaState: SagaState) {
    sagaState.status = 'DEBITING';
    // Persist sagaState update (e.g., to a database)
    await kafkaProducer.send({
      topic: 'payment-commands',
      messages: [{
        value: JSON.stringify({
          type: 'DebitFundsCommand',
          transactionId: sagaState.transactionId,
          accountId: sagaState.senderAccountId,
          amount: sagaState.amount,
        }),
        key: sagaState.transactionId
      }]
    });
    console.log(`Orchestrator: Sent DebitFundsCommand for ${sagaState.transactionId}`);
  }

  private async sendCreditFundsCommand(sagaState: SagaState) {
    sagaState.status = 'CREDITING';
    // Persist sagaState update
    await kafkaProducer.send({
      topic: 'ledger-commands',
      messages: [{
        value: JSON.stringify({
          type: 'CreditFundsCommand',
          transactionId: sagaState.transactionId,
          accountId: sagaState.recipientAccountId,
          amount: sagaState.amount,
        }),
        key: sagaState.transactionId
      }]
    });
    console.log(`Orchestrator: Sent CreditFundsCommand for ${sagaState.transactionId}`);
  }

  private async sendCompensateDebitCommand(sagaState: SagaState) {
    sagaState.status = 'COMPENSATING';
    // Persist sagaState update
    await kafkaProducer.send({
      topic: 'payment-commands',
      messages: [{
        value: JSON.stringify({
          type: 'CompensateDebitCommand',
          transactionId: sagaState.transactionId,
          accountId: sagaState.senderAccountId,
          amount: sagaState.amount,
        }),
        key: sagaState.transactionId
      }]
    });
    console.log(`Orchestrator: Sent CompensateDebitCommand for ${sagaState.transactionId}`);
  }

  // This method would be called by Kafka consumer when an event arrives
  async handleParticipantEvent(event: any) {
    const sagaState = this.sagaStates.get(event.transactionId);
    if (!sagaState) {
      console.warn(`Orchestrator: Saga state not found for ${event.transactionId}`);
      return;
    }

    switch (event.type) {
      case 'FundsDebitedEvent':
        sagaState.status = 'DEBITED';
        console.log(`Orchestrator: FundsDebitedEvent received for ${event.transactionId}`);
        await this.sendCreditFundsCommand(sagaState);
        break;
      case 'DebitFailedEvent':
        sagaState.status = 'FAILED';
        console.error(`Orchestrator: DebitFailedEvent received for ${event.transactionId}. Saga failed.`);
        // No compensation needed if debit failed at first step
        break;
      case 'FundsCreditedEvent':
        sagaState.status = 'COMPLETED';
        console.log(`Orchestrator: FundsCreditedEvent received for ${event.transactionId}. Saga completed.`);
        // Potentially send notification command
        break;
      case 'CreditFailedEvent':
        sagaState.status = 'FAILED';
        console.error(`Orchestrator: CreditFailedEvent received for ${event.transactionId}. Initiating compensation.`);
        await this.sendCompensateDebitCommand(sagaState);
        break;
      case 'DebitCompensatedEvent':
        sagaState.status = 'FAILED'; // The original transfer failed and was compensated
        console.log(`Orchestrator: DebitCompensatedEvent received for ${event.transactionId}. Saga ended in failure with compensation.`);
        break;
      // ... handle other events and compensation steps
    }
    // Persist updated sagaState
  }
}

-- SQL schema for an outbox table
CREATE TABLE outbox (
    id UUID PRIMARY KEY,
    aggregate_id UUID NOT NULL,
    aggregate_type VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    processed_at TIMESTAMP WITH TIME ZONE,
    status VARCHAR(50) NOT NULL DEFAULT 'PENDING' -- PENDING, SENT, FAILED
);

// payment.service.ts (with transactional outbox logic)
import { Database } from './database'; // Assuming a database wrapper
import { v4 as uuid } from 'uuid';

interface TransferRequest {
  senderAccountId: string;
  recipientAccountId: string;
  amount: number;
}

export class PaymentServiceWithOutbox {
  constructor(private db: Database) {}

  async debitFunds(request: TransferRequest): Promise<string> {
    const transactionId = uuid();
    await this.db.beginTransaction();
    try {
      // 1. Business Logic: Debit sender's account
      await this.db.execute(`UPDATE accounts SET balance = balance - $1 WHERE id = $2`, [request.amount, request.senderAccountId]);
      await this.db.execute(`INSERT INTO transactions (id, type, status, sender, recipient, amount) VALUES ($1, 'debit', 'pending', $2, $3, $4)`, [transactionId, request.senderAccountId, request.recipientAccountId, request.amount]);

      // 2. Outbox Logic: Insert event into outbox table
      const eventPayload = {
        transactionId: transactionId,
        senderAccountId: request.senderAccountId,
        recipientAccountId: request.recipientAccountId,
        amount: request.amount,
        type: 'FundsDebitedEvent'
      };
      await this.db.execute(
        `INSERT INTO outbox (id, aggregate_id, aggregate_type, event_type, payload) VALUES ($1, $2, 'Payment', 'FundsDebitedEvent', $3)`,
        [uuid(), transactionId, JSON.stringify(eventPayload)]
      );

      await this.db.commitTransaction();
      console.log(`Funds debited and event saved to outbox for ${transactionId}.`);
      return transactionId;
    } catch (error) {
      await this.db.rollbackTransaction();
      console.error(`Failed to debit funds and save event for ${transactionId}:`, error);
      throw error;
    }
  }
}

// A separate OutboxRelayService (could be a simple poller or CDC)
// outbox_relay.service.ts
import { kafkaProducer } from './kafka';

export class OutboxRelayService {
  constructor(private db: Database) {}

  async pollAndPublishEvents() {
    // Fetch unprocessed events from the outbox
    const events = await this.db.query(`SELECT * FROM outbox WHERE status = 'PENDING' ORDER BY created_at ASC FOR UPDATE SKIP LOCKED`);

    for (const event of events) {
      try {
        // Publish event to Kafka
        await kafkaProducer.send({
          topic: 'payment-events',
          messages: [{ value: JSON.stringify(event.payload), key: event.aggregate_id }],
        });

        // Mark event as sent
        await this.db.execute(`UPDATE outbox SET status = 'SENT', processed_at = NOW() WHERE id = $1`, [event.id]);
        console.log(`Published event ${event.id} for aggregate ${event.aggregate_id}`);
      } catch (error) {
        console.error(`Failed to publish event ${event.id}:`, error);
        // In a real system, handle retry logic, or mark as FAILED for manual review
      }
    }
  }

  startPolling(intervalMs: number = 1000) {
    setInterval(() => this.pollAndPublishEvents(), intervalMs);
    console.log(`Outbox relay service started, polling every ${intervalMs}ms.`);
  }
}

public interface PaymentServiceTCC {
    // Phase 1: Try - Reserve funds without final commit
    boolean tryDebit(String transactionId, String accountId, BigDecimal amount);

    // Phase 2: Confirm - Finalize debit
    void confirmDebit(String transactionId);

    // Phase 3: Cancel - Release reserved funds
    void cancelDebit(String transactionId);
}

public class PaymentTCCCoordinator {
    private final PaymentServiceTCC paymentService;
    private final InventoryServiceTCC inventoryService; // Another participant

    public PaymentTCCCoordinator(PaymentServiceTCC ps, InventoryServiceTCC is) {
        this.paymentService = ps;
        this.inventoryService = is;
    }

    public boolean executeOrderPayment(String transactionId, String userId, BigDecimal amount, String productId, int quantity) {
        boolean paymentTrySuccess = false;
        boolean inventoryTrySuccess = false;

        try {
            // Try Phase
            paymentTrySuccess = paymentService.tryDebit(transactionId, userId, amount);
            if (!paymentTrySuccess) {
                throw new RuntimeException("Payment Try failed");
            }
            inventoryTrySuccess = inventoryService.tryReserve(transactionId, productId, quantity);
            if (!inventoryTrySuccess) {
                throw new RuntimeException("Inventory Try failed");
            }

            // Confirm Phase (only if all Try succeeded)
            paymentService.confirmDebit(transactionId);
            inventoryService.confirmReserve(transactionId);
            System.out.println("Transaction " + transactionId + " completed successfully.");
            return true;

        } catch (Exception e) {
            System.err.println("Transaction " + transactionId + " failed. Initiating Cancel: " + e.getMessage());
            // Cancel Phase
            if (paymentTrySuccess) {
                paymentService.cancelDebit(transactionId);
            }
            if (inventoryTrySuccess) {
                inventoryService.cancelReserve(transactionId);
            }
            System.out.println("Transaction " + transactionId + " cancelled.");
            return false;
        }
    }
}

// payment.controller.ts (Idempotent API endpoint example)
import { Request, Response } from 'express';
import { PaymentService } from './payment.service';
import { IdempotencyStore } from './idempotency.store'; // e.g., Redis, a dedicated DB table

export class PaymentController {
  constructor(private paymentService: PaymentService, private idempotencyStore: IdempotencyStore) {}

  async processPayment(req: Request, res: Response) {
    const idempotencyKey = req.headers['x-idempotency-key'] as string;
    if (!idempotencyKey) {
      return res.status(400).send({ message: 'x-idempotency-key header is required' });
    }

    // 1. Check idempotency store
    const existingResponse = await this.idempotencyStore.get(idempotencyKey);
    if (existingResponse) {
      console.log(`Idempotent request for key ${idempotencyKey}. Returning cached response.`);
      return res.status(200).send(existingResponse);
    }

    try {
      const { accountId, amount } = req.body;
      // 2. Process the actual payment (e.g., call debitFunds method from PaymentService)
      const transactionResult = await this.paymentService.processDebit(accountId, amount);

      // 3. Store the successful response with the idempotency key
      await this.idempotencyStore.set(idempotencyKey, transactionResult, 60 * 60); // Cache for 1 hour

      res.status(200).send(transactionResult);
    } catch (error) {
      console.error(`Payment processing failed for key ${idempotencyKey}:`, error);
      // It's crucial NOT to store error responses as idempotent successes
      res.status(500).send({ message: 'Payment failed', error: error.message });
    }
  }
}

// Basic IdempotencyStore interface
interface IdempotencyStore {
  get(key: string): Promise<any | null>;
  set(key: string, value: any, ttlSeconds: number): Promise<void>;
}