CQRS: Separating Reads from Writes

// Traditional CRUD: One model for reads and writes
const user = database.query('SELECT * FROM users WHERE id = 1');
user.name = 'Alice';
database.update('UPDATE users SET name = ? WHERE id = 1', [user.name]);

// CQRS: Separate write and read models
// Write model (normalized, enforces rules)
const command = new UpdateUserNameCommand(userId, newName);
commandHandler.handle(command); // Updates source of truth

// Read model (denormalized, optimized for queries)
const user = readDatabase.query('SELECT * FROM user_read_model WHERE id = 1');

class PlaceOrderCommand {
  constructor(customerId, items) {
    this.customerId = customerId;
    this.items = items; // [{productId, quantity}]
  }
}

class PlaceOrderCommandHandler {
  constructor(customerRepository, productRepository, eventPublisher) {
    this.customers = customerRepository;
    this.products = productRepository;
    this.eventPublisher = eventPublisher;
  }

  handle(command) {
    // Enforce business rules in the write model
    const customer = this.customers.getById(command.customerId);

    if (!customer.isActive()) {
      throw new Error("Customer account is inactive");
    }

    // Create aggregate
    const order = new Order(generateId(), customer);

    for (let item of command.items) {
      const product = this.products.getById(item.productId);

      if (product.discontinued) {
        throw new Error(`Product ${product.id} is discontinued`);
      }

      if (item.quantity <= 0) {
        throw new Error("Quantity must be positive");
      }

      order.addLineItem(product, item.quantity);
    }

    // Calculate totals, apply discounts, enforce rules
    order.calculateTotal();

    if (order.total.isZero()) {
      throw new Error("Order total must be greater than zero");
    }

    // Publish events (source of truth)
    const events = order.getDomainEvents();
    for (let event of events) {
      this.eventPublisher.publish(event);
    }

    return order.id;
  }
}

// Read model: denormalized for efficient queries
class OrderListQuery {
  constructor(customerId, pageSize, pageNumber) {
    this.customerId = customerId;
    this.pageSize = pageSize;
    this.pageNumber = pageNumber;
  }
}

class OrderListQueryHandler {
  constructor(readDatabase) {
    this.db = readDatabase;
  }

  handle(query) {
    // This read model might look completely different
    // from the write model's Order aggregate
    const sql = `
      SELECT
        id,
        status,
        total,
        createdAt,
        itemCount,
        customerEmail
      FROM order_list
      WHERE customerId = ?
      ORDER BY createdAt DESC
      LIMIT ? OFFSET ?
    `;

    return this.db.query(sql, [
      query.customerId,
      query.pageSize,
      query.pageSize * (query.pageNumber - 1)
    ]);
  }
}

// Another read model for a different query
class OrderDetailQuery {
  constructor(orderId) {
    this.orderId = orderId;
  }
}

class OrderDetailQueryHandler {
  constructor(readDatabase) {
    this.db = readDatabase;
  }

  handle(query) {
    const sql = `
      SELECT
        id,
        status,
        total,
        createdAt,
        items,
        shippingAddress,
        paymentMethod,
        trackingNumber
      FROM order_details
      WHERE id = ?
    `;

    return this.db.query(sql, [query.orderId]);
  }
}

class OrderEventProjector {
  constructor(readDatabase) {
    this.db = readDatabase;
  }

  projectOrderPlaced(event) {
    // When order is placed in write model, update read models
    this.db.insert('order_list', {
      id: event.orderId,
      customerId: event.customerId,
      status: 'placed',
      total: event.total,
      createdAt: event.timestamp,
      itemCount: event.items.length,
      customerEmail: event.customerEmail
    });

    for (let item of event.items) {
      this.db.insert('order_details_items', {
        orderId: event.orderId,
        productId: item.productId,
        quantity: item.quantity,
        price: item.price
      });
    }
  }

  projectOrderShipped(event) {
    // Update read models when order ships
    this.db.update('order_list', {
      status: 'shipped'
    }, { id: event.orderId });

    this.db.update('order_details', {
      trackingNumber: event.trackingNumber,
      status: 'shipped'
    }, { id: event.orderId });
  }

  // Handle other events...
}

class EventBus {
  constructor(projectors) {
    this.projectors = projectors;
  }

  publish(event) {
    // Write model publishes event
    // Read models project it
    for (let projector of this.projectors) {
      const methodName = `project${event.constructor.name}`;
      if (projector[methodName]) {
        projector[methodName](event);
      }
    }
  }
}

// Write model: stores events
class EventStore {
  append(event) {
    // Every command generates events that are stored
    // Event is the source of truth
    this.database.insert('events', {
      aggregateId: event.aggregateId,
      type: event.constructor.name,
      data: JSON.stringify(event),
      timestamp: new Date()
    });
  }

  getEvents(aggregateId) {
    // Reconstruct aggregate by replaying events
    return this.database.query(
      'SELECT * FROM events WHERE aggregateId = ? ORDER BY timestamp',
      [aggregateId]
    );
  }
}

class Order {
  static fromEvents(events) {
    const order = new Order();
    for (let event of events) {
      order.apply(event);
    }
    return order;
  }

  apply(event) {
    if (event instanceof OrderPlacedEvent) {
      this.id = event.orderId;
      this.customerId = event.customerId;
      this.items = event.items;
      this.status = 'placed';
    } else if (event instanceof OrderShippedEvent) {
      this.status = 'shipped';
      this.trackingNumber = event.trackingNumber;
    }
    // ...
  }
}

// Read model: uses events to build denormalized views
const eventStore = new EventStore(database);
const events = eventStore.getEvents(orderId);
const order = Order.fromEvents(events);

// Projectors subscribe to event stream and update read models
eventBus.subscribe(event => {
  orderListProjector.handle(event);
  orderDetailProjector.handle(event);
  analyticsProjector.handle(event);
});

// Write model: Order with line items (normalized)
class Order {
  id;
  customerId;
  status;
  items: OrderLineItem[];
}

class OrderLineItem {
  productId;
  quantity;
  price;
}

// But clients need denormalized read model:
// "Show me all orders with product name, customer name, and total"
// This requires joins across Order, Customer, and Product
// CQRS lets you denormalize this into a single table

// This doesn't benefit from CQRS
class SimpleTask {
  id;
  title;
  completed;
  dueDate;
}

// Everyone queries the same shape
// Traditional database is cleaner

// Client creates order
placeOrderCommand.execute();

// Immediately queries read model
// Might not see it yet (eventual consistency)
const orders = getOrderListQuery.execute();

// Solution: return order ID from write operation
// Client can query specific order immediately
const orderId = placeOrderCommand.execute();
const order = getOrderDetailQuery.execute(orderId);
// This order is in the write model, readable immediately

// If projector crashes mid-way through processing events:
// Read model is partially updated
// Needs recovery mechanism

class RobustProjector {
  handle(event) {
    try {
      // Update read model
      this.updateReadModel(event);
      // Mark event as processed
      this.recordProcessedEvent(event.id);
    } catch (error) {
      // Don't mark as processed
      // Retry will happen
      throw error;
    }
  }
}