CQRS & Event Sourcing

// ============================================================
// COMMAND SIDE — Write Model
// ============================================================

interface PlaceOrderCommand {
    type: 'PLACE_ORDER';
    customerId: string;
    items: { productId: string; quantity: number; price: number }[];
    shippingAddress: string;
}

class PlaceOrderCommandHandler {
    constructor(
        private readonly orderRepository: OrderWriteRepository,
        private readonly eventBus: EventBus
    ) {}

    async handle(command: PlaceOrderCommand): Promise<void> {
        // Validate command
        if (command.items.length === 0) {
            throw new Error('Order must have at least one item');
        }

        // Create aggregate and persist to Write DB (PostgreSQL)
        const order = Order.create({
            customerId: command.customerId,
            items: command.items,
            shippingAddress: command.shippingAddress,
        });

        await this.orderRepository.save(order);

        // Publish domain event to Kafka — Query side reads this
        await this.eventBus.publish({
            type: 'OrderPlaced',
            orderId: order.id,
            customerId: order.customerId,
            totalAmount: order.totalAmount,
            timestamp: new Date().toISOString(),
        });

        // Command handler returns void — no data back
    }
}

// ============================================================
// QUERY SIDE — Read Model
// ============================================================

interface GetOrderQuery {
    orderId: string;
}

interface OrderReadModel {
    orderId: string;
    customerName: string;   // Denormalized — no JOIN needed
    customerEmail: string;  // Denormalized — no JOIN needed
    items: { productName: string; quantity: number; price: number }[];
    totalAmount: number;
    status: string;
    createdAt: string;
}

class GetOrderQueryHandler {
    constructor(
        private readonly orderReadRepository: OrderReadRepository // Elasticsearch
    ) {}

    async handle(query: GetOrderQuery): Promise<OrderReadModel> {
        // Read from optimized Read DB — super fast, no JOIN
        const order = await this.orderReadRepository.findById(query.orderId);

        if (!order) {
            throw new Error(`Order ${query.orderId} not found`);
        }

        return order; // Returns full data object
    }
}

// ============================================================
// Domain Events
// ============================================================
interface AccountOpened {
    type: 'AccountOpened';
    accountId: string;
    ownerId: string;
    timestamp: string;
}

interface MoneyDeposited {
    type: 'MoneyDeposited';
    accountId: string;
    amount: number;
    timestamp: string;
}

interface MoneyWithdrawn {
    type: 'MoneyWithdrawn';
    accountId: string;
    amount: number;
    timestamp: string;
}

type BankAccountEvent = AccountOpened | MoneyDeposited | MoneyWithdrawn;

// ============================================================
// Aggregate — Events apply করে state manage করে
// ============================================================
class BankAccountAggregate {
    private accountId: string = '';
    private balance: number = 0;
    private ownerId: string = '';
    private isOpen: boolean = false;
    private uncommittedEvents: BankAccountEvent[] = [];

    // Reconstruct state from event history (replay)
    static rehydrate(events: BankAccountEvent[]): BankAccountAggregate {
        const account = new BankAccountAggregate();
        for (const event of events) {
            account.apply(event);  // Apply each event in order
        }
        return account;
    }

    // Command: Open account
    openAccount(accountId: string, ownerId: string): void {
        if (this.isOpen) throw new Error('Account already open');
        const event: AccountOpened = {
            type: 'AccountOpened',
            accountId, ownerId,
            timestamp: new Date().toISOString(),
        };
        this.apply(event);
        this.uncommittedEvents.push(event);
    }

    // Command: Deposit money
    deposit(amount: number): void {
        if (!this.isOpen) throw new Error('Account is closed');
        if (amount <= 0) throw new Error('Amount must be positive');
        const event: MoneyDeposited = {
            type: 'MoneyDeposited',
            accountId: this.accountId, amount,
            timestamp: new Date().toISOString(),
        };
        this.apply(event);
        this.uncommittedEvents.push(event);
    }

    // Command: Withdraw money
    withdraw(amount: number): void {
        if (!this.isOpen) throw new Error('Account is closed');
        if (amount > this.balance) throw new Error('Insufficient funds');
        const event: MoneyWithdrawn = {
            type: 'MoneyWithdrawn',
            accountId: this.accountId, amount,
            timestamp: new Date().toISOString(),
        };
        this.apply(event);
        this.uncommittedEvents.push(event);
    }

    // Internal: apply event to update in-memory state
    private apply(event: BankAccountEvent): void {
        switch (event.type) {
            case 'AccountOpened':
                this.accountId = event.accountId;
                this.ownerId = event.ownerId;
                this.isOpen = true;
                break;
            case 'MoneyDeposited':
                this.balance += event.amount;
                break;
            case 'MoneyWithdrawn':
                this.balance -= event.amount;
                break;
        }
    }

    getBalance(): number { return this.balance; }
    getUncommittedEvents(): BankAccountEvent[] { return this.uncommittedEvents; }
    clearUncommittedEvents(): void { this.uncommittedEvents = []; }
}

// ============================================================
// Event Store Schema (PostgreSQL — append-only table)
// ============================================================

/*
CREATE TABLE event_store (
    id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_id  VARCHAR(255) NOT NULL,      -- e.g., "ACC-001"
    aggregate_type VARCHAR(100) NOT NULL,     -- e.g., "BankAccount"
    event_type    VARCHAR(100) NOT NULL,      -- e.g., "MoneyDeposited"
    event_data    JSONB NOT NULL,             -- Serialized event payload
    version       INT NOT NULL,              -- Optimistic concurrency control
    metadata      JSONB,                     -- correlation_id, user_id, etc.
    timestamp     TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    -- Ensure version is sequential per aggregate
    UNIQUE(aggregate_id, version)
);

-- Index for loading events of a specific aggregate
CREATE INDEX idx_event_store_aggregate ON event_store(aggregate_id, version ASC);
*/

// ============================================================
// Event Store Implementation
// ============================================================
interface StoredEvent {
    id: string;
    aggregateId: string;
    aggregateType: string;
    eventType: string;
    eventData: unknown;
    version: number;
    metadata?: Record<string, unknown>;
    timestamp: string;
}

class EventStore {
    constructor(private readonly db: Database) {}

    // Save new events — append-only, never update
    async appendEvents(
        aggregateId: string,
        aggregateType: string,
        events: BankAccountEvent[],
        expectedVersion: number   // Optimistic concurrency
    ): Promise<void> {
        await this.db.transaction(async (tx) => {
            // Check for concurrent modification
            const currentVersion = await tx.query<{ max: number }>(
                'SELECT MAX(version) as max FROM event_store WHERE aggregate_id = $1',
                [aggregateId]
            );
            const latestVersion = currentVersion.rows[0]?.max ?? -1;

            if (latestVersion !== expectedVersion) {
                throw new ConcurrencyError(
                    `Expected version ${expectedVersion}, got ${latestVersion}`
                );
            }

            // Append each event
            for (let i = 0; i < events.length; i++) {
                await tx.query(
                    `INSERT INTO event_store
                     (aggregate_id, aggregate_type, event_type, event_data, version)
                     VALUES ($1, $2, $3, $4, $5)`,
                    [aggregateId, aggregateType, events[i].type,
                     JSON.stringify(events[i]), expectedVersion + i + 1]
                );
            }
        });
    }

    // Load all events for an aggregate (to reconstruct state)
    async loadEvents(aggregateId: string): Promise<StoredEvent[]> {
        const result = await this.db.query<StoredEvent>(
            `SELECT * FROM event_store
             WHERE aggregate_id = $1
             ORDER BY version ASC`,
            [aggregateId]
        );
        return result.rows;
    }

    // Time travel: load events up to a specific point in time
    async loadEventsUntil(aggregateId: string, until: Date): Promise<StoredEvent[]> {
        const result = await this.db.query<StoredEvent>(
            `SELECT * FROM event_store
             WHERE aggregate_id = $1 AND timestamp <= $2
             ORDER BY version ASC`,
            [aggregateId, until.toISOString()]
        );
        return result.rows;
    }
}

// ============================================================
// Projection: Order Status View
// Listens to events → builds denormalized read model
// ============================================================

interface OrderStatusReadModel {
    orderId: string;
    customerId: string;
    customerName: string;       // Denormalized
    status: 'pending' | 'confirmed' | 'shipped' | 'delivered' | 'cancelled';
    totalAmount: number;
    itemCount: number;
    shippingAddress: string;
    placedAt: string;
    shippedAt?: string;
    deliveredAt?: string;
    lastUpdated: string;
}

class OrderStatusProjection {
    constructor(
        private readonly readDb: ReadDatabase  // Elasticsearch or PostgreSQL read replica
    ) {}

    // Handles OrderPlaced event
    async onOrderPlaced(event: {
        orderId: string;
        customerId: string;
        customerName: string;
        items: { productId: string; quantity: number; price: number }[];
        shippingAddress: string;
        timestamp: string;
    }): Promise<void> {
        const totalAmount = event.items.reduce(
            (sum, item) => sum + item.price * item.quantity, 0
        );

        // Build and upsert the read model
        const readModel: OrderStatusReadModel = {
            orderId: event.orderId,
            customerId: event.customerId,
            customerName: event.customerName,  // Already denormalized in event
            status: 'pending',
            totalAmount,
            itemCount: event.items.length,
            shippingAddress: event.shippingAddress,
            placedAt: event.timestamp,
            lastUpdated: event.timestamp,
        };

        await this.readDb.upsert('order_status_view', event.orderId, readModel);
    }

    // Handles OrderShipped event
    async onOrderShipped(event: {
        orderId: string;
        trackingNumber: string;
        timestamp: string;
    }): Promise<void> {
        await this.readDb.update('order_status_view', event.orderId, {
            status: 'shipped',
            shippedAt: event.timestamp,
            lastUpdated: event.timestamp,
        });
    }

    // Handles OrderDelivered event
    async onOrderDelivered(event: {
        orderId: string;
        timestamp: string;
    }): Promise<void> {
        await this.readDb.update('order_status_view', event.orderId, {
            status: 'delivered',
            deliveredAt: event.timestamp,
            lastUpdated: event.timestamp,
        });
    }
}

// ============================================================
// Projection: Inventory View — same events, different read model
// ============================================================

class InventoryProjection {
    async onOrderPlaced(event: {
        items: { productId: string; quantity: number }[];
    }): Promise<void> {
        // Decrement available stock for each item
        for (const item of event.items) {
            await this.readDb.increment(
                'inventory_view',
                item.productId,
                { reserved_qty: item.quantity }
            );
        }
    }

    async onOrderDelivered(event: {
        items: { productId: string; quantity: number }[];
    }): Promise<void> {
        for (const item of event.items) {
            await this.readDb.increment(
                'inventory_view',
                item.productId,
                { sold_qty: item.quantity, reserved_qty: -item.quantity }
            );
        }
    }
}

// ============================================================
// Complete CQRS + Event Sourcing Flow
// ============================================================

// 1. Command arrives
const command: PlaceOrderCommand = {
    type: 'PLACE_ORDER',
    customerId: 'CUST-001',
    items: [
        { productId: 'PROD-A', quantity: 2, price: 500 },
        { productId: 'PROD-B', quantity: 1, price: 1200 },
    ],
    shippingAddress: 'Dhaka, Bangladesh',
};

// 2. Command Handler orchestrates everything
class PlaceOrderCommandHandler {
    constructor(
        private eventStore: EventStore,
        private eventBus: EventBus
    ) {}

    async handle(cmd: PlaceOrderCommand): Promise<void> {
        // Load Aggregate from Event Store (rehydrate from history)
        const pastEvents = await this.eventStore.loadEvents(`ORDER-${cmd.customerId}`);
        const order = OrderAggregate.rehydrate(pastEvents);

        // Execute command on aggregate — raises domain events
        order.placeOrder({
            customerId: cmd.customerId,
            items: cmd.items,
            shippingAddress: cmd.shippingAddress,
        });

        // Persist new events to Event Store (append-only)
        const newEvents = order.getUncommittedEvents();
        await this.eventStore.appendEvents(
            order.id,
            'Order',
            newEvents,
            order.version - newEvents.length  // optimistic concurrency
        );

        // Publish to Kafka — Projections will consume
        for (const event of newEvents) {
            await this.eventBus.publish('order-events', event);
        }

        // Return void — command has no return value
    }
}

// 3. Projection consumes event → updates Read DB
class OrderProjectionConsumer {
    async consume(event: DomainEvent): Promise<void> {
        switch (event.type) {
            case 'OrderPlaced':
                await this.readDb.upsert('orders_view', {
                    orderId: event.orderId,
                    customerId: event.customerId,
                    status: 'PENDING',
                    totalAmount: event.totalAmount,
                    placedAt: event.timestamp,
                });
                break;
            case 'OrderShipped':
                await this.readDb.update('orders_view', event.orderId, {
                    status: 'SHIPPED',
                    shippedAt: event.timestamp,
                });
                break;
        }
    }
}

// 4. Query Handler reads from optimized Read DB
class GetOrderQueryHandler {
    async handle(query: { orderId: string }) {
        // Instant read — no JOIN, pre-built view
        return await this.readDb.findById('orders_view', query.orderId);
    }
}