How I Learned to Stop Worrying and Love Distributed Transactions (After Breaking Production Twice)

// This code haunts my dreams - never do this
async function processOrder(orderData) {
  const payment = await paymentService.charge(orderData.payment);
  const inventory = await inventoryService.reserve(orderData.items);
  const shipping = await shippingService.create(orderData.address);
  
  // What happens if shipping fails? Payment already went through!
  // What happens if the network fails after payment but before inventory?
  // What happens if this service crashes here?
  
  return { payment, inventory, shipping };
}

// The transaction coordinator from hell
class TransactionCoordinator {
  async execute(transactionSteps) {
    // Phase 1: Prepare all participants
    const prepared = await Promise.all(
      transactionSteps.map(step => step.prepare())
    );
    
    if (prepared.every(result => result.success)) {
      // Phase 2: Commit all participants
      return await Promise.all(
        transactionSteps.map(step => step.commit())
      );
    } else {
      // Rollback all participants
      await Promise.all(
        transactionSteps.map(step => step.rollback())
      );
    }
  }
}

// Event-driven saga that actually works in production
class OrderProcessingSaga {
  constructor(eventBus, sagaStore) {
    this.eventBus = eventBus;
    this.sagaStore = sagaStore;
    
    // Register event handlers - each step is independent
    this.eventBus.on('OrderCreated', this.handleOrderCreated.bind(this));
    this.eventBus.on('PaymentProcessed', this.handlePaymentProcessed.bind(this));
    this.eventBus.on('PaymentFailed', this.handlePaymentFailed.bind(this));
    this.eventBus.on('InventoryReserved', this.handleInventoryReserved.bind(this));
    this.eventBus.on('InventoryUnavailable', this.handleInventoryUnavailable.bind(this));
  }

  async handleOrderCreated(event) {
    const sagaId = event.orderId;
    
    // Always save saga state first - this saved me countless debugging hours
    await this.sagaStore.create(sagaId, {
      orderId: event.orderId,
      status: 'PAYMENT_PENDING',
      compensations: [] // Track what needs to be undone
    });

    // Emit event to trigger payment processing
    await this.eventBus.emit('ProcessPayment', {
      sagaId,
      orderId: event.orderId,
      amount: event.total,
      paymentMethod: event.paymentMethod
    });
  }

  async handlePaymentProcessed(event) {
    const saga = await this.sagaStore.get(event.sagaId);
    
    // Update saga state and add compensation action
    saga.status = 'INVENTORY_PENDING';
    saga.paymentId = event.paymentId;
    saga.compensations.push({
      action: 'RefundPayment',
      data: { paymentId: event.paymentId, amount: event.amount }
    });
    
    await this.sagaStore.update(saga);

    // Trigger inventory reservation
    await this.eventBus.emit('ReserveInventory', {
      sagaId: event.sagaId,
      orderId: saga.orderId,
      items: saga.items
    });
  }

  async handleInventoryUnavailable(event) {
    const saga = await this.sagaStore.get(event.sagaId);
    
    // Execute compensations in reverse order - critical for data integrity
    for (const compensation of saga.compensations.reverse()) {
      await this.executeCompensation(compensation);
    }
    
    saga.status = 'FAILED';
    await this.sagaStore.update(saga);

    // Notify order service of failure
    await this.eventBus.emit('OrderFailed', {
      orderId: saga.orderId,
      reason: 'INVENTORY_UNAVAILABLE'
    });
  }

  // The compensation logic that prevents data corruption
  async executeCompensation(compensation) {
    switch (compensation.action) {
      case 'RefundPayment':
        await this.eventBus.emit('RefundPayment', compensation.data);
        break;
      case 'ReleaseInventory':
        await this.eventBus.emit('ReleaseInventory', compensation.data);
        break;
      // Add more compensation actions as needed
    }
  }
}

// Event bus with guaranteed delivery and replay capability
class ReliableEventBus {
  constructor(redisClient) {
    this.redis = redisClient;
    this.consumers = new Map();
  }

  async emit(eventType, eventData) {
    const event = {
      id: this.generateEventId(),
      type: eventType,
      data: eventData,
      timestamp: Date.now(),
      version: 1
    };

    // Redis Streams provide durability and ordering guarantees
    await this.redis.xadd(
      `events:${eventType}`,
      '*',
      'event',
      JSON.stringify(event)
    );
  }

  async subscribe(eventType, handler, consumerGroup) {
    // Create consumer group if it doesn't exist
    try {
      await this.redis.xgroup('CREATE', `events:${eventType}`, consumerGroup, '$');
    } catch (err) {
      // Group already exists - that's fine
    }

    // Start consuming events
    const consumerId = `consumer-${process.pid}-${Date.now()}`;
    
    while (true) {
      try {
        const messages = await this.redis.xreadgroup(
          'GROUP', consumerGroup, consumerId,
          'COUNT', 10,
          'BLOCK', 5000,
          'STREAMS', `events:${eventType}`, '>'
        );

        for (const [stream, events] of messages) {
          for (const [messageId, fields] of events) {
            const event = JSON.parse(fields[1]);
            
            try {
              await handler(event.data);
              // Acknowledge successful processing
              await this.redis.xack(`events:${eventType}`, consumerGroup, messageId);
            } catch (error) {
              console.error(`Failed to process event ${messageId}:`, error);
              // Event will be retried due to lack of ACK
            }
          }
        }
      } catch (error) {
        console.error('Event bus error:', error);
        await this.sleep(1000); // Back off on errors
      }
    }
  }
}

// Long-running saga with timeout handling
class ShippingFulfillmentSaga {
  async handleOrderShipped(event) {
    const saga = await this.sagaStore.get(event.sagaId);
    
    // Set up timeout for delivery confirmation
    await this.scheduleTimeout({
      sagaId: event.sagaId,
      action: 'DeliveryTimeout',
      delay: 7 * 24 * 60 * 60 * 1000 // 7 days
    });
    
    saga.status = 'AWAITING_DELIVERY';
    await this.sagaStore.update(saga);
  }

  async handleDeliveryTimeout(event) {
    // Trigger customer service workflow for undelivered packages
    await this.eventBus.emit('InvestigateDelivery', {
      orderId: event.orderId,
      shippingId: event.shippingId
    });
  }
}

// Bulletproof saga persistence with full audit trail
class PostgresSagaStore {
  async create(sagaId, initialState) {
    const client = await this.pool.connect();
    
    try {
      await client.query('BEGIN');
      
      // Create saga record
      await client.query(
        'INSERT INTO sagas (id, status, created_at, updated_at) VALUES ($1, $2, NOW(), NOW())',
        [sagaId, initialState.status]
      );
      
      // Store initial state event
      await client.query(
        'INSERT INTO saga_events (saga_id, event_type, event_data, sequence) VALUES ($1, $2, $3, 1)',
        [sagaId, 'SagaCreated', JSON.stringify(initialState)]
      );
      
      await client.query('COMMIT');
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }

  async get(sagaId) {
    // Rebuild state from events - enables time travel debugging
    const events = await this.pool.query(
      'SELECT event_type, event_data FROM saga_events WHERE saga_id = $1 ORDER BY sequence',
      [sagaId]
    );
    
    let state = {};
    for (const event of events.rows) {
      state = this.applyEvent(state, event.event_type, JSON.parse(event.event_data));
    }
    
    return state;
  }
}

// Correlation IDs tie everything together across services
class EventTracer {
  static correlate(originalEvent, newEventData) {
    return {
      ...newEventData,
      correlationId: originalEvent.correlationId || originalEvent.id,
      parentEventId: originalEvent.id,
      traceId: originalEvent.traceId || this.generateTraceId()
    };
  }
}

// Idempotent event processing - this pattern is non-negotiable
async handlePaymentProcessed(event) {
  const deduplicationKey = `payment-${event.paymentId}-${event.orderId}`;
  
  // Check if we've already processed this exact event
  const existing = await this.redis.get(deduplicationKey);
  if (existing) {
    console.log(`Duplicate event detected: ${deduplicationKey}`);
    return; // Skip processing
  }
  
  // Process the event
  await this.processPayment(event);
  
  // Mark as processed with expiration
  await this.redis.setex(deduplicationKey, 3600, 'processed');
}