Designing Event-Driven Microservices Architecture with Apache Kafka, Event Sourcing, and CQRS

import { Client } from 'pg';

export interface OrderCreatedEvent {
  orderId: string;
  customerId: string;
  totalAmount: number;
  createdAt: string;
}

export async function createOrderWithOutbox(
  dbClient: Client,
  orderData: { orderId: string; customerId: string; amount: number }
): Promise<void> {
  try {
    await dbClient.query('BEGIN');

    // 1. Persist domain entity into primary table
    await dbClient.query(
      `INSERT INTO orders (id, customer_id, amount, status) VALUES ($1, $2, $3, 'CREATED')`,
      [orderData.orderId, orderData.customerId, orderData.amount]
    );

    // 2. Write outbox event inside the EXACT SAME database transaction
    const eventPayload: OrderCreatedEvent = {
      orderId: orderData.orderId,
      customerId: orderData.customerId,
      totalAmount: orderData.amount,
      createdAt: new Date().toISOString(),
    };

    await dbClient.query(
      `INSERT INTO transactional_outbox (aggregate_type, aggregate_id, event_type, payload)
       VALUES ('ORDER', $1, 'ORDER_CREATED', $2)`,
      [orderData.orderId, JSON.stringify(eventPayload)]
    );

    await dbClient.query('COMMIT');
  } catch (error) {
    await dbClient.query('ROLLBACK');
    throw error;
  }
}