Understanding CQRS and CDC: A Practical Guide with Real-World Analogies

restaurant-cqrs-cdc.js

/**
 * Restaurant CQRS and CDC Example
 *
 * This file demonstrates the Command Query Responsibility Segregation (CQRS)
 * and Change Data Capture (CDC) patterns using a restaurant analogy.
 */

// ===== Event Emitter (simulating a message bus) =====
const EventEmitter = require('events');

// ===== Command Bus (Write Path) =====
class CommandBus {
    constructor() {
        this.handlers = [];
        console.log('Command Bus initialized (like the order taking system)');
    }

    subscribe(handler) {
        this.handlers.push(handler);
        console.log(`Handler subscribed to Command Bus: ${handler.constructor.name}`);
    }

    publish(command) {
        console.log(`Command published: ${command.type}`, command.payload);
        // Simulate asynchronous behavior
        setTimeout(() => {
            this.handlers.forEach(handler => handler.handle(command));
        }, 10);
    }
}

// ===== Command Handler (Write Path) =====
class CommandHandler {
    constructor(database, cdcEventBus) {
        this.database = database;
        this.cdcEventBus = cdcEventBus;
        console.log('Command Handler initialized (like the kitchen staff)');
    }

    handle(command) {
        console.log(`Handling command: ${command.type}`);
        switch (command.type) {
            case 'CreateOrder':
                this.createOrder(command.payload);
                break;
            case 'UpdateOrder':
                this.updateOrder(command.payload);
                break;
            case 'CancelOrder':
                this.cancelOrder(command.payload);
                break;
            default:
                console.log(`Unknown command type: ${command.type}`);
        }
    }

    createOrder(orderData) {
        console.log(`Creating order: ${orderData.id}`);
        this.database.set(`order:${orderData.id}`, orderData);
        this.emitCDCEvent('ORDER_CREATED', orderData);
    }

    updateOrder(orderData) {
        console.log(`Updating order: ${orderData.id}`);
        this.database.set(`order:${orderData.id}`, orderData);
        this.emitCDCEvent('ORDER_UPDATED', orderData);
    }

    cancelOrder(orderData) {
        console.log(`Canceling order: ${orderData.id}`);
        this.database.delete(`order:${orderData.id}`);
        this.emitCDCEvent('ORDER_CANCELED', orderData);
    }

    emitCDCEvent(type, data) {
        console.log(`Emitting CDC event: ${type}`);
        this.cdcEventBus.emit('dataChanged', { type, data });
    }
}

// ===== Database (Write Path) =====
class Database {
    constructor() {
        this.data = new Map();
        console.log('Database initialized (like the main kitchen)');
    }

    set(key, value) {
        console.log(`Setting data: ${key}`);
        this.data.set(key, value);
    }

    get(key) {
        console.log(`Getting data: ${key}`);
        return this.data.get(key);
    }

    delete(key) {
        console.log(`Deleting data: ${key}`);
        this.data.delete(key);
    }

    getAll() {
        console.log('Getting all data');
        return Array.from(this.data.entries()).map(([key, value]) => ({ key, value }));
    }
}

// ===== CDC Processor (Synchronization) =====
class CDCProcessor {
    constructor(readReplica) {
        this.readReplica = readReplica;
        console.log('CDC Processor initialized (like the expeditor)');
    }

    handleCDCEvent(cdcEvent) {
        console.log(`Handling CDC event: ${cdcEvent.type}`);
        switch (cdcEvent.type) {
            case 'ORDER_CREATED':
                this.handleNewOrder(cdcEvent.data);
                break;
            case 'ORDER_UPDATED':
                this.handleOrderUpdate(cdcEvent.data);
                break;
            case 'ORDER_CANCELED':
                this.handleOrderCancel(cdcEvent.data);
                break;
            default:
                console.log(`Unknown CDC event type: ${cdcEvent.type}`);
        }
    }

    handleNewOrder(orderData) {
        console.log(`CDC: Adding new order to read replica: ${orderData.id}`);
        this.readReplica.set(`order:${orderData.id}`, orderData);
    }

    handleOrderUpdate(orderData) {
        console.log(`CDC: Updating order in read replica: ${orderData.id}`);
        this.readReplica.set(`order:${orderData.id}`, orderData);
    }

    handleOrderCancel(orderData) {
        console.log(`CDC: Removing canceled order from read replica: ${orderData.id}`);
        this.readReplica.delete(`order:${orderData.id}`);
    }
}

// ===== Read Model (Read Path) =====
class ReadModel {
    constructor() {
        this.data = new Map();
        console.log('Read Model initialized (like the service stations)');
    }

    set(key, value) {
        console.log(`Setting read data: ${key}`);
        this.data.set(key, value);
    }

    get(key) {
        console.log(`Getting read data: ${key}`);
        return this.data.get(key);
    }

    delete(key) {
        console.log(`Deleting read data: ${key}`);
        this.data.delete(key);
    }

    getAll() {
        console.log('Getting all read data');
        return Array.from(this.data.entries()).map(([key, value]) => ({ key, value }));
    }

    // Specialized query methods
    getDineInOrders() {
        console.log('Getting all dine-in orders');
        return this.getAll()
            .filter(item => item.value.type === 'dine-in')
            .map(item => item.value);
    }

    getDeliveryOrders() {
        console.log('Getting all delivery orders');
        return this.getAll()
            .filter(item => item.value.type === 'delivery')
            .map(item => item.value);
    }

    getTakeoutOrders() {
        console.log('Getting all takeout orders');
        return this.getAll()
            .filter(item => item.value.type === 'takeout')
            .map(item => item.value);
    }

    getOrdersByStatus(status) {
        console.log(`Getting all orders with status: ${status}`);
        return this.getAll()
            .filter(item => item.value.status === status)
            .map(item => item.value);
    }
}

// ===== Query Service (Read Path) =====
class QueryService {
    constructor(readModel) {
        this.readModel = readModel;
        console.log('Query Service initialized (like the server information system)');
    }

    query(query) {
        console.log(`Executing query: ${query.type}`);
        switch (query.type) {
            case 'GetOrderById':
                return this.readModel.get(`order:${query.payload.orderId}`);
            case 'GetAllOrders':
                return this.readModel.getAll().map(item => item.value);
            case 'GetDineInOrders':
                return this.readModel.getDineInOrders();
            case 'GetDeliveryOrders':
                return this.readModel.getDeliveryOrders();
            case 'GetTakeoutOrders':
                return this.readModel.getTakeoutOrders();
            case 'GetOrdersByStatus':
                return this.readModel.getOrdersByStatus(query.payload.status);
            default:
                console.log(`Unknown query type: ${query.type}`);
                return null;
        }
    }
}

// ===== Main Application =====
function runRestaurantExample() {
    console.log('\n===== Restaurant CQRS and CDC Example =====\n');

    // Initialize components
    const commandBus = new CommandBus();
    const database = new Database();
    const cdcEventBus = new EventEmitter();
    const readReplica = new ReadModel();
    const cdcProcessor = new CDCProcessor(readReplica);
    const commandHandler = new CommandHandler(database, cdcEventBus);
    const queryService = new QueryService(readReplica);

    // Subscribe handlers
    commandBus.subscribe(commandHandler);
    cdcEventBus.on('dataChanged', cdcProcessor.handleCDCEvent.bind(cdcProcessor));

    // Example 1: Create a dine-in order
    console.log('\n----- Example 1: Create a dine-in order -----');
    commandBus.publish({
        type: 'CreateOrder',
        payload: {
            id: 1,
            type: 'dine-in',
            table: 5,
            items: ['Pasta', 'Salad'],
            specialInstructions: 'No onions',
            status: 'pending',
            timestamp: new Date().toISOString()
        }
    });

    // Example 2: Create a delivery order
    console.log('\n----- Example 2: Create a delivery order -----');
    commandBus.publish({
        type: 'CreateOrder',
        payload: {
            id: 2,
            type: 'delivery',
            address: '123 Main St',
            items: ['Pizza', 'Wings'],
            specialInstructions: 'Extra cheese',
            status: 'pending',
            deliveryTime: new Date(Date.now() + 3600000).toISOString(),
            timestamp: new Date().toISOString()
        }
    });

    // Example 3: Create a takeout order
    console.log('\n----- Example 3: Create a takeout order -----');
    commandBus.publish({
        type: 'CreateOrder',
        payload: {
            id: 3,
            type: 'takeout',
            items: ['Burger', 'Fries'],
            specialInstructions: 'No pickles',
            status: 'pending',
            pickupTime: new Date(Date.now() + 1800000).toISOString(),
            timestamp: new Date().toISOString()
        }
    });

    // Wait for commands to be processed
    setTimeout(() => {
        // Example 4: Update an order status
        console.log('\n----- Example 4: Update an order status -----');
        commandBus.publish({
            type: 'UpdateOrder',
            payload: {
                id: 1,
                type: 'dine-in',
                table: 5,
                items: ['Pasta', 'Salad'],
                specialInstructions: 'No onions',
                status: 'preparing',
                timestamp: new Date().toISOString()
            }
        });

        // Wait for commands to be processed
        setTimeout(() => {
            // Example 5: Query all orders
            console.log('\n----- Example 5: Query all orders -----');
            const allOrders = queryService.query({ type: 'GetAllOrders' });
            console.log('All orders:', allOrders);

            // Example 6: Query dine-in orders
            console.log('\n----- Example 6: Query dine-in orders -----');
            const dineInOrders = queryService.query({ type: 'GetDineInOrders' });
            console.log('Dine-in orders:', dineInOrders);

            // Example 7: Query delivery orders
            console.log('\n----- Example 7: Query delivery orders -----');
            const deliveryOrders = queryService.query({ type: 'GetDeliveryOrders' });
            console.log('Delivery orders:', deliveryOrders);

            // Example 8: Query takeout orders
            console.log('\n----- Example 8: Query takeout orders -----');
            const takeoutOrders = queryService.query({ type: 'GetTakeoutOrders' });
            console.log('Takeout orders:', takeoutOrders);

            // Example 9: Query orders by status
            console.log('\n----- Example 9: Query orders by status -----');
            const pendingOrders = queryService.query({
                type: 'GetOrdersByStatus',
                payload: { status: 'pending' }
            });
            console.log('Pending orders:', pendingOrders);

            const preparingOrders = queryService.query({
                type: 'GetOrdersByStatus',
                payload: { status: 'preparing' }
            });
            console.log('Preparing orders:', preparingOrders);

            // Example 10: Cancel an order
            console.log('\n----- Example 10: Cancel an order -----');
            commandBus.publish({
                type: 'CancelOrder',
                payload: {
                    id: 2,
                    type: 'delivery',
                    status: 'canceled',
                    timestamp: new Date().toISOString()
                }
            });

            // Wait for commands to be processed
            setTimeout(() => {
                // Example 11: Final state of the system
                console.log('\n----- Example 11: Final state of the system -----');
                console.log('Main database state:');
                console.log(database.getAll());

                console.log('\nRead replica state:');
                console.log(readReplica.getAll());

                console.log('\n===== Restaurant CQRS and CDC Example Complete =====\n');
            }, 100);
        }, 100);
    }, 100);
}

// Run the example
runRestaurantExample();