Deep Dive into the Saga Pattern: Choreography vs Orchestration, and Implementing Compensation Logic with Temporal

// order-service/order.consumer.ts
import { Kafka } from 'kafkajs';
 
const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();
const consumer = kafka.consumer({ groupId: 'order-service' });
 
export async function createOrderAndPublish(ctx: OrderContext): Promise<void> {
  await orderRepository.create({ id: ctx.orderId, status: 'PENDING' });
  await producer.send({
    topic: 'order-events',
    messages: [{ key: ctx.orderId, value: JSON.stringify({ type: 'OrderCreated', ...ctx }) }],
  });
}
 
// Subscribe to compensation event: cancel order after payment refund completes
await consumer.subscribe({ topics: ['payment-events'] });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString());
    if (event.type === 'PaymentRefunded') {
      await orderRepository.updateStatus(event.orderId, 'CANCELLED');
    }
  },
});

// payment-service/payment.consumer.ts
import { Kafka } from 'kafkajs';
 
const consumer = kafka.consumer({ groupId: 'payment-service' });
const producer = kafka.producer();
 
// Subscribe to both happy-path events and compensation events
await consumer.subscribe({ topics: ['order-events', 'inventory-events'] });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString());
 
    if (event.type === 'OrderCreated') {
      const result = await paymentGateway.charge(event);
      await producer.send({
        topic: 'payment-events',
        messages: [{
          key: event.orderId,
          value: JSON.stringify(
            result.success
              ? { type: 'PaymentProcessed', orderId: event.orderId, transactionId: result.transactionId }
              : { type: 'PaymentFailed', orderId: event.orderId, reason: result.reason }
          ),
        }],
      });
    }
 
    // Refund on inventory failure (compensating transaction)
    if (event.type === 'InventoryFailed') {
      await paymentGateway.refund({ orderId: event.orderId, transactionId: event.transactionId });
      await producer.send({
        topic: 'payment-events',
        messages: [{
          key: event.orderId,
          value: JSON.stringify({ type: 'PaymentRefunded', orderId: event.orderId }),
        }],
      });
    }
  },
});

// TypeORM + PostgreSQL example
async function createOrderWithOutbox(
  ctx: OrderContext,
  queryRunner: QueryRunner
): Promise<void> {
  await queryRunner.startTransaction();
  try {
    await queryRunner.manager.save(Order, {
      id: ctx.orderId,
      status: 'PENDING',
      amount: ctx.amount,
    });
 
    // Instead of publishing directly to Kafka, write to the outbox table (same transaction)
    await queryRunner.manager.save(OutboxEvent, {
      id: uuidv4(),
      aggregateType: 'Order',
      aggregateId: ctx.orderId,
      eventType: 'OrderCreated',
      payload: JSON.stringify(ctx),
      createdAt: new Date(),
    });
 
    await queryRunner.commitTransaction();
  } catch (err) {
    await queryRunner.rollbackTransaction();
    throw err;
  }
  // A CDC tool like Debezium detects changes to the outbox table → publishes to Kafka
}

pnpm add @temporalio/workflow @temporalio/activity @temporalio/worker @temporalio/client

// activities.ts
import { log, ApplicationFailure } from '@temporalio/activity';
 
export interface OrderContext {
  orderId: string;
  userId: string;
  amount: number;
  items: Array<{ productId: string; quantity: number }>;
}
 
export async function createOrder(ctx: OrderContext): Promise<void> {
  await orderRepository.create({ id: ctx.orderId, userId: ctx.userId, status: 'PENDING', amount: ctx.amount });
  log.info('Order created', { orderId: ctx.orderId });
}
 
export async function cancelOrder(orderId: string): Promise<void> {
  // Skip if order is already cancelled (idempotency guarantee)
  const order = await orderRepository.findById(orderId);
  if (order?.status === 'CANCELLED') return;
  await orderRepository.updateStatus(orderId, 'CANCELLED');
  log.info('Order cancelled', { orderId });
}
 
export async function processPayment(ctx: OrderContext): Promise<string> {
  const result = await paymentGateway.charge({
    orderId: ctx.orderId,
    userId: ctx.userId,
    amount: ctx.amount,
  });
 
  if (!result.success) {
    // Payment decline won't change outcome on retry, so mark as non-retryable
    throw ApplicationFailure.nonRetryable(`Payment failed: ${result.reason}`, 'PAYMENT_DECLINED');
  }
 
  return result.transactionId;
}
 
export async function refundPayment(orderId: string, transactionId: string): Promise<void> {
  const existing = await paymentRepository.findRefund(orderId);
  if (existing) {
    log.info('Already refunded, skipping', { orderId });
    return;
  }
  await paymentGateway.refund({ orderId, transactionId, idempotencyKey: `refund-${orderId}` });
  await paymentRepository.markRefunded(orderId, transactionId);
  log.info('Refund complete', { orderId });
}
 
export async function updateInventory(ctx: OrderContext): Promise<void> {
  // Note: if this Activity is retried, already-decremented items may be decremented again.
  // In production, use an idempotency key on each decrement call, or
  // check current inventory before decrementing to verify it has not already been processed.
  const decremented: typeof ctx.items = [];
 
  for (const item of ctx.items) {
    const success = await inventoryService.decrement(item.productId, item.quantity);
    if (!success) {
      // Insufficient inventory — restore already-decremented items within this Activity, then throw non-retryable
      for (const done of decremented) {
        await inventoryService.increment(done.productId, done.quantity);
      }
      throw ApplicationFailure.nonRetryable(`Insufficient inventory: ${item.productId}`, 'INSUFFICIENT_INVENTORY');
    }
    decremented.push(item);
  }
}
 
export async function restoreInventory(ctx: OrderContext): Promise<void> {
  for (const item of ctx.items) {
    await inventoryService.increment(item.productId, item.quantity);
  }
  log.info('Inventory restored', { orderId: ctx.orderId });
}
 
export async function scheduleShipping(orderId: string): Promise<void> {
  await shippingService.schedule({ orderId });
}
 
export async function cancelShipping(orderId: string): Promise<void> {
  await shippingService.cancel({ orderId });
  log.info('Shipping cancelled', { orderId });
}

// workflows.ts
import { proxyActivities, log } from '@temporalio/workflow';
import type * as activities from './activities';
import type { OrderContext } from './activities';
 
const {
  createOrder, cancelOrder,
  processPayment, refundPayment,
  updateInventory, restoreInventory,
  scheduleShipping, cancelShipping,
} = proxyActivities<typeof activities>({
  startToCloseTimeout: '10 minutes',
  retry: {
    maximumAttempts: 3,
    initialInterval: '1 second',
    backoffCoefficient: 2,
  },
});
 
export async function orderSagaWorkflow(ctx: OrderContext): Promise<void> {
  // Compensation stack: unshift() prepends to the array so for...of iteration automatically runs in reverse
  const compensations: Array<() => Promise<void>> = [];
 
  try {
    await createOrder(ctx);
    compensations.unshift(() => cancelOrder(ctx.orderId));
 
    const transactionId = await processPayment(ctx);
    // transactionId is captured in a closure — no need to manage it as separate state at compensation time
    compensations.unshift(() => refundPayment(ctx.orderId, transactionId));
 
    await updateInventory(ctx);
    compensations.unshift(() => restoreInventory(ctx));
 
    await scheduleShipping(ctx.orderId);
    compensations.unshift(() => cancelShipping(ctx.orderId));
 
    log.info('Order processing complete', { orderId: ctx.orderId });
  } catch (err) {
    log.error('Order processing failed, executing compensating transactions', { orderId: ctx.orderId, error: String(err) });
 
    for (const compensate of compensations) {
      try {
        await compensate();
      } catch (compensationErr) {
        // Compensation failed — the system may be left in an inconsistent state.
        // Send an alert or push to a DLQ to secure a manual recovery path.
        log.error('Compensating transaction failed', { error: String(compensationErr) });
        // Even if this compensation fails, continue attempting the remaining compensations
      }
    }
 
    throw err;
  }
}

// worker.ts
import { Worker } from '@temporalio/worker';
import * as activities from './activities';
 
async function run() {
  const worker = await Worker.create({
    workflowsPath: require.resolve('./workflows'),
    activities,
    taskQueue: 'order-saga',
  });
  await worker.run();
}
 
run().catch(console.error);

// client.ts
import { Client } from '@temporalio/client';
import { orderSagaWorkflow } from './workflows';
import type { OrderContext } from './activities';
 
const client = new Client();
const orderCtx: OrderContext = {
  orderId: 'ORD-20260712-001',
  userId: 'user-123',
  amount: 59000,
  items: [{ productId: 'PROD-456', quantity: 2 }],
};
 
const handle = await client.workflow.start(orderSagaWorkflow, {
  taskQueue: 'order-saga',
  // Use orderId as workflowId: prevents duplicate executions for the same order.
  // Even if start() is called twice due to network retries, Temporal guarantees idempotency.
  workflowId: `order-${orderCtx.orderId}`,
  args: [orderCtx],
});
 
console.log(`Workflow started: ${handle.workflowId}`);
await handle.result();

// Idempotency guarantee example — refund Activity
export async function refundPayment(orderId: string, transactionId: string): Promise<void> {
  const existing = await paymentRepository.findRefund(orderId);
  if (existing) {
    log.info('Already refunded, skipping', { orderId });
    return; // Called twice, but the refund happens only once
  }
 
  await paymentGateway.refund({
    orderId,
    transactionId,
    idempotencyKey: `refund-${orderId}`, // Pass idempotency key to the external payment API as well
  });
 
  await paymentRepository.markRefunded(orderId, transactionId);
}