Saga Pattern: Quản lý giao dịch phân tán trong kiến trúc Microservices

// order-service/src/events/handlers.ts
import { Kafka, Consumer } from 'kafkajs';
import { OrderRepository } from '../repositories/order-repository';
import { OrderStatus } from '../models/order';

export class OrderEventHandlers {
  private consumer: Consumer;
  private orderRepository: OrderRepository;
  
  constructor(kafka: Kafka, orderRepository: OrderRepository) {
    this.consumer = kafka.consumer({ groupId: 'order-service' });
    this.orderRepository = orderRepository;
  }
  
  async start(): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'payment-refunded', fromBeginning: true });
    
    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        if (topic === 'payment-refunded') {
          const event = JSON.parse(message.value!.toString());
          await this.handlePaymentRefunded(event);
        }
      },
    });
  }
  
  private async handlePaymentRefunded(event: { orderId: string }): Promise<void> {
    const order = await this.orderRepository.findById(event.orderId);
    if (!order) {
      console.error(`Order ${event.orderId} not found`);
      return;
    }
    
    order.status = OrderStatus.CANCELLED;
    await this.orderRepository.save(order);
    console.log(`Order ${event.orderId} cancelled due to inventory shortage`);
  }
  
  async stop(): Promise<void> {
    await this.consumer.disconnect();
  }
}

// payment-service/src/events/handlers.ts
import { Kafka, Consumer, Producer } from 'kafkajs';
import { PaymentRepository } from '../repositories/payment-repository';
import { RefundService } from '../services/refund-service';

export class PaymentEventHandlers {
  private consumer: Consumer;
  private producer: Producer;
  private paymentRepository: PaymentRepository;
  private refundService: RefundService;
  
  constructor(
    kafka: Kafka, 
    paymentRepository: PaymentRepository, 
    refundService: RefundService
  ) {
    this.consumer = kafka.consumer({ groupId: 'payment-service' });
    this.producer = kafka.producer();
    this.paymentRepository = paymentRepository;
    this.refundService = refundService;
  }
  
  async start(): Promise<void> {
    await this.consumer.connect();
    await this.producer.connect();
    await this.consumer.subscribe({ topic: 'inventory-failed', fromBeginning: true });
    
    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        if (topic === 'inventory-failed') {
          const event = JSON.parse(message.value!.toString());
          await this.handleInventoryFailed(event);
        }
      },
    });
  }
  
  private async handleInventoryFailed(event: { orderId: string }): Promise<void> {
    // Hoàn tiền
    const payment = await this.paymentRepository.findByOrderId(event.orderId);
    if (!payment) {
      console.error(`Payment for order ${event.orderId} not found`);
      return;
    }
    
    await this.refundService.refundPayment(payment);
    
    // Phát event
    await this.producer.send({
      topic: 'payment-refunded',
      messages: [
        { 
          key: event.orderId, 
          value: JSON.stringify({ orderId: event.orderId }) 
        },
      ],
    });
    
    console.log(`Payment refunded for order ${event.orderId}`);
  }
  
  async stop(): Promise<void> {
    await this.consumer.disconnect();
    await this.producer.disconnect();
  }
}

// order-saga/src/sagas/order-saga.ts
import { call, put, takeLatest } from 'redux-saga/effects';
import axios from 'axios';
import { 
  createOrderSuccess, 
  createOrderFailure,
  processPaymentSuccess,
  processPaymentFailure,
  // ... các action creators khác
} from '../actions/order-actions';

// Types
interface OrderRequest {
  customerId: string;
  items: Array<{
    productId: string;
    quantity: number;
    price: number;
  }>;
  totalAmount: number;
}

enum OrderState {
  STARTED = 'STARTED',
  ORDER_CREATED = 'ORDER_CREATED',
  PAYMENT_PROCESSED = 'PAYMENT_PROCESSED',
  INVENTORY_RESERVED = 'INVENTORY_RESERVED',
  DELIVERY_SCHEDULED = 'DELIVERY_SCHEDULED',
  COMPLETED = 'COMPLETED',
  CANCELLED = 'CANCELLED',
}

// Saga Workers
function* createOrder(action: { type: string, payload: OrderRequest }) {
  try {
    const { data } = yield call(
      axios.post, 
      'http://order-service/orders', 
      action.payload
    );
    yield put(createOrderSuccess(data));
    yield put({ type: 'PROCESS_PAYMENT', payload: { 
      orderId: data.id, 
      amount: action.payload.totalAmount 
    }});
  } catch (error) {
    yield put(createOrderFailure(error));
    // Không cần compensating transaction ở bước đầu tiên
  }
}

function* processPayment(action: { type: string, payload: { orderId: string, amount: number }}) {
  try {
    const { data } = yield call(
      axios.post, 
      'http://payment-service/payments', 
      {
        orderId: action.payload.orderId,
        amount: action.payload.amount,
      }
    );
    yield put(processPaymentSuccess(data));
    yield put({ type: 'RESERVE_INVENTORY', payload: { 
      orderId: action.payload.orderId,
      // Thêm dữ liệu cần thiết
    }});
  } catch (error) {
    yield put(processPaymentFailure(error));
    // Compensating transaction: hủy đơn hàng
    yield put({ type: 'CANCEL_ORDER', payload: { orderId: action.payload.orderId }});
  }
}

function* reserveInventory(action: { type: string, payload: { orderId: string }}) {
  try {
    const { data } = yield call(
      axios.post, 
      'http://inventory-service/inventory/reserve', 
      {
        orderId: action.payload.orderId,
        // Thêm dữ liệu cần thiết
      }
    );
    yield put({ type: 'RESERVE_INVENTORY_SUCCESS', payload: data });
    yield put({ type: 'SCHEDULE_DELIVERY', payload: { 
      orderId: action.payload.orderId,
      // Thêm dữ liệu cần thiết
    }});
  } catch (error) {
    yield put({ type: 'RESERVE_INVENTORY_FAILURE', payload: error });
    // Compensating transaction: hoàn tiền và hủy đơn hàng
    yield put({ type: 'REFUND_PAYMENT', payload: { orderId: action.payload.orderId }});
    yield put({ type: 'CANCEL_ORDER', payload: { orderId: action.payload.orderId }});
  }
}

// Compensating transaction workers
function* cancelOrder(action: { type: string, payload: { orderId: string }}) {
  try {
    yield call(
      axios.put, 
      `http://order-service/orders/${action.payload.orderId}/cancel`
    );
    yield put({ type: 'CANCEL_ORDER_SUCCESS', payload: action.payload });
  } catch (error) {
    yield put({ type: 'CANCEL_ORDER_FAILURE', payload: error });
    console.error('Failed to cancel order', error);
    // Có thể cần một cơ chế retry hoặc manual intervention
  }
}

function* refundPayment(action: { type: string, payload: { orderId: string }}) {
  try {
    yield call(
      axios.post, 
      'http://payment-service/payments/refund', 
      { orderId: action.payload.orderId }
    );
    yield put({ type: 'REFUND_PAYMENT_SUCCESS', payload: action.payload });
  } catch (error) {
    yield put({ type: 'REFUND_PAYMENT_FAILURE', payload: error });
    console.error('Failed to refund payment', error);
    // Cần một cơ chế retry hoặc manual intervention
  }
}

// Saga Watcher
export function* orderSaga() {
  yield takeLatest('CREATE_ORDER', createOrder);
  yield takeLatest('PROCESS_PAYMENT', processPayment);
  yield takeLatest('RESERVE_INVENTORY', reserveInventory);
  yield takeLatest('SCHEDULE_DELIVERY', scheduleDelivery);
  yield takeLatest('CANCEL_ORDER', cancelOrder);
  yield takeLatest('REFUND_PAYMENT', refundPayment);
  // ... các saga watchers khác
}