Build Event-Sourced Inventory Management

// src/events/inventory-events.ts — Immutable inventory event definitions
import { z } from "zod";

// Base event structure — all events share these fields
const BaseEvent = z.object({
  eventId: z.string().uuid(),
  sku: z.string(),
  warehouseId: z.string(),
  timestamp: z.date(),
  correlationId: z.string(),    // links related events (e.g., order → shipment)
  causedBy: z.string(),          // user ID or system name
  metadata: z.record(z.any()).optional(),
});

// Specific event types
export const StockReceived = BaseEvent.extend({
  type: z.literal("StockReceived"),
  quantity: z.number().int().positive(),
  purchaseOrderId: z.string(),
  unitCost: z.number().positive(),
  supplier: z.string(),
});

export const StockReserved = BaseEvent.extend({
  type: z.literal("StockReserved"),
  quantity: z.number().int().positive(),
  orderId: z.string(),
  expiresAt: z.date(),            // reservation expires if not fulfilled
});

export const StockShipped = BaseEvent.extend({
  type: z.literal("StockShipped"),
  quantity: z.number().int().positive(),
  orderId: z.string(),
  shipmentId: z.string(),
});

export const StockReturned = BaseEvent.extend({
  type: z.literal("StockReturned"),
  quantity: z.number().int().positive(),
  orderId: z.string(),
  returnId: z.string(),
  reason: z.enum(["defective", "wrong_item", "customer_changed_mind", "damaged_in_transit"]),
  condition: z.enum(["resellable", "damaged", "dispose"]),
});

export const StockTransferred = BaseEvent.extend({
  type: z.literal("StockTransferred"),
  quantity: z.number().int().positive(),
  fromWarehouseId: z.string(),
  toWarehouseId: z.string(),
  transferId: z.string(),
});

export const StockAdjusted = BaseEvent.extend({
  type: z.literal("StockAdjusted"),
  quantityChange: z.number().int(), // positive = gain, negative = loss
  reason: z.enum(["physical_count", "damage", "expiry", "theft", "system_correction"]),
  notes: z.string(),
  previousQuantity: z.number().int(),
  newQuantity: z.number().int(),
});

export const ReservationExpired = BaseEvent.extend({
  type: z.literal("ReservationExpired"),
  quantity: z.number().int().positive(),
  orderId: z.string(),
});

export type InventoryEvent = z.infer<typeof StockReceived>
  | z.infer<typeof StockReserved>
  | z.infer<typeof StockShipped>
  | z.infer<typeof StockReturned>
  | z.infer<typeof StockTransferred>
  | z.infer<typeof StockAdjusted>
  | z.infer<typeof ReservationExpired>;

// src/store/event-store.ts — Append-only event store with optimistic concurrency
import { pool } from "../db";
import { InventoryEvent } from "../events/inventory-events";
import { Kafka } from "kafkajs";

const kafka = new Kafka({ brokers: [process.env.KAFKA_BROKER!] });
const producer = kafka.producer();

// SQL schema for the event store
// CREATE TABLE inventory_events (
//   id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
//   stream_id VARCHAR(200) NOT NULL,        -- e.g., "inventory:SKU123:warehouse-1"
//   version BIGINT NOT NULL,                -- monotonically increasing per stream
//   event_type VARCHAR(50) NOT NULL,
//   data JSONB NOT NULL,
//   correlation_id VARCHAR(100),
//   caused_by VARCHAR(100),
//   created_at TIMESTAMPTZ DEFAULT NOW(),
//   UNIQUE (stream_id, version)             -- optimistic concurrency
// );
// CREATE INDEX idx_events_stream ON inventory_events (stream_id, version);
// CREATE INDEX idx_events_correlation ON inventory_events (correlation_id);
// CREATE INDEX idx_events_created ON inventory_events (created_at);

export class EventStore {
  async append(
    streamId: string,
    events: InventoryEvent[],
    expectedVersion: number
  ): Promise<number> {
    const client = await pool.connect();

    try {
      await client.query("BEGIN");

      // Check current version (optimistic concurrency)
      const { rows } = await client.query(
        "SELECT COALESCE(MAX(version), 0) as current_version FROM inventory_events WHERE stream_id = $1 FOR UPDATE",
        [streamId]
      );
      const currentVersion = Number(rows[0].current_version);

      if (currentVersion !== expectedVersion) {
        await client.query("ROLLBACK");
        throw new ConcurrencyError(
          `Expected version ${expectedVersion} but found ${currentVersion} for stream ${streamId}`
        );
      }

      // Append events with sequential versions
      let version = currentVersion;
      for (const event of events) {
        version++;
        await client.query(
          `INSERT INTO inventory_events (stream_id, version, event_type, data, correlation_id, caused_by)
           VALUES ($1, $2, $3, $4, $5, $6)`,
          [streamId, version, event.type, JSON.stringify(event), event.correlationId, event.causedBy]
        );
      }

      await client.query("COMMIT");

      // Publish to Kafka for projections and downstream consumers
      await producer.send({
        topic: "inventory-events",
        messages: events.map((event, i) => ({
          key: streamId,
          value: JSON.stringify({ ...event, streamId, version: currentVersion + i + 1 }),
        })),
      });

      return version;
    } catch (error) {
      await client.query("ROLLBACK");
      throw error;
    } finally {
      client.release();
    }
  }

  async getStream(streamId: string, fromVersion?: number): Promise<InventoryEvent[]> {
    const { rows } = await pool.query(
      `SELECT data FROM inventory_events 
       WHERE stream_id = $1 ${fromVersion ? "AND version > $2" : ""}
       ORDER BY version ASC`,
      fromVersion ? [streamId, fromVersion] : [streamId]
    );
    return rows.map((r) => r.data as InventoryEvent);
  }

  // Temporal query: get events for a stream up to a specific point in time
  async getStreamAt(streamId: string, asOf: Date): Promise<InventoryEvent[]> {
    const { rows } = await pool.query(
      "SELECT data FROM inventory_events WHERE stream_id = $1 AND created_at <= $2 ORDER BY version ASC",
      [streamId, asOf]
    );
    return rows.map((r) => r.data as InventoryEvent);
  }

  // Get all events for a correlation ID (e.g., all events related to order #123)
  async getByCorrelation(correlationId: string): Promise<InventoryEvent[]> {
    const { rows } = await pool.query(
      "SELECT data FROM inventory_events WHERE correlation_id = $1 ORDER BY created_at ASC",
      [correlationId]
    );
    return rows.map((r) => r.data as InventoryEvent);
  }
}

export class ConcurrencyError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "ConcurrencyError";
  }
}

// src/projections/stock-level.ts — Real-time stock level projection from events
import { InventoryEvent } from "../events/inventory-events";
import { Redis } from "ioredis";
import { pool } from "../db";

const redis = new Redis(process.env.REDIS_URL!);

export interface StockLevel {
  sku: string;
  warehouseId: string;
  available: number;      // can be sold
  reserved: number;       // held for pending orders
  incoming: number;       // expected from purchase orders
  total: number;          // available + reserved
  lastUpdated: Date;
}

// Rebuild stock level from events (used for initialization and verification)
export function projectStockLevel(events: InventoryEvent[]): StockLevel {
  let available = 0;
  let reserved = 0;
  const sku = events[0]?.sku || "";
  const warehouseId = events[0]?.warehouseId || "";

  for (const event of events) {
    switch (event.type) {
      case "StockReceived":
        available += event.quantity;
        break;
      case "StockReserved":
        available -= event.quantity;
        reserved += event.quantity;
        break;
      case "StockShipped":
        reserved -= event.quantity;
        break;
      case "StockReturned":
        if (event.condition === "resellable") {
          available += event.quantity;
        }
        // damaged/dispose items don't return to available stock
        break;
      case "StockTransferred":
        if (event.fromWarehouseId === warehouseId) {
          available -= event.quantity;
        }
        if (event.toWarehouseId === warehouseId) {
          available += event.quantity;
        }
        break;
      case "StockAdjusted":
        available += event.quantityChange;
        break;
      case "ReservationExpired":
        reserved -= event.quantity;
        available += event.quantity; // return to available pool
        break;
    }
  }

  return {
    sku,
    warehouseId,
    available: Math.max(0, available),
    reserved: Math.max(0, reserved),
    incoming: 0,
    total: Math.max(0, available + reserved),
    lastUpdated: events[events.length - 1]?.timestamp || new Date(),
  };
}

// Real-time projection updater — processes events from Kafka
export class StockLevelProjection {
  async handleEvent(event: InventoryEvent): Promise<void> {
    const key = `stock:${event.sku}:${event.warehouseId}`;

    // Update Redis projection incrementally (fast path)
    switch (event.type) {
      case "StockReceived":
        await redis.hincrby(key, "available", event.quantity);
        break;
      case "StockReserved":
        await redis.hincrby(key, "available", -event.quantity);
        await redis.hincrby(key, "reserved", event.quantity);
        break;
      case "StockShipped":
        await redis.hincrby(key, "reserved", -event.quantity);
        break;
      case "StockReturned":
        if (event.condition === "resellable") {
          await redis.hincrby(key, "available", event.quantity);
        }
        break;
      case "StockAdjusted":
        await redis.hincrby(key, "available", event.quantityChange);
        break;
      case "ReservationExpired":
        await redis.hincrby(key, "reserved", -event.quantity);
        await redis.hincrby(key, "available", event.quantity);
        break;
    }

    await redis.hset(key, "lastUpdated", new Date().toISOString());

    // Also update the read-model table for complex queries
    await pool.query(
      `INSERT INTO stock_levels (sku, warehouse_id, available, reserved, updated_at)
       VALUES ($1, $2, 
         COALESCE((SELECT available FROM stock_levels WHERE sku = $1 AND warehouse_id = $2), 0),
         COALESCE((SELECT reserved FROM stock_levels WHERE sku = $1 AND warehouse_id = $2), 0),
         NOW())
       ON CONFLICT (sku, warehouse_id) DO UPDATE SET
         available = stock_levels.available + $3,
         reserved = stock_levels.reserved + $4,
         updated_at = NOW()`,
      [event.sku, event.warehouseId,
        event.type === "StockReceived" ? event.quantity : 0,
        event.type === "StockReserved" ? event.quantity : 0]
    );
  }

  // Query current stock (from Redis projection)
  async getStockLevel(sku: string, warehouseId: string): Promise<StockLevel | null> {
    const key = `stock:${sku}:${warehouseId}`;
    const data = await redis.hgetall(key);
    if (!data.available) return null;

    return {
      sku,
      warehouseId,
      available: parseInt(data.available || "0"),
      reserved: parseInt(data.reserved || "0"),
      incoming: parseInt(data.incoming || "0"),
      total: parseInt(data.available || "0") + parseInt(data.reserved || "0"),
      lastUpdated: new Date(data.lastUpdated || Date.now()),
    };
  }

  // "Time travel" — what was the stock level at a specific point in time?
  async getStockLevelAt(sku: string, warehouseId: string, asOf: Date): Promise<StockLevel> {
    const eventStore = new (await import("../store/event-store")).EventStore();
    const streamId = `inventory:${sku}:${warehouseId}`;
    const events = await eventStore.getStreamAt(streamId, asOf);
    return projectStockLevel(events);
  }
}

// src/routes/inventory.ts — Inventory API with command/query separation
import { Hono } from "hono";
import { randomUUID } from "node:crypto";
import { EventStore, ConcurrencyError } from "../store/event-store";
import { StockLevelProjection } from "../projections/stock-level";

const eventStore = new EventStore();
const stockProjection = new StockLevelProjection();
const app = new Hono();

// Query: Get current stock level
app.get("/stock/:sku/:warehouseId", async (c) => {
  const { sku, warehouseId } = c.req.param();
  const stock = await stockProjection.getStockLevel(sku, warehouseId);
  if (!stock) return c.json({ error: "Not found" }, 404);
  return c.json(stock);
});

// Query: Stock level at a point in time (for investigations)
app.get("/stock/:sku/:warehouseId/at", async (c) => {
  const { sku, warehouseId } = c.req.param();
  const asOf = new Date(c.req.query("date")!);
  const stock = await stockProjection.getStockLevelAt(sku, warehouseId, asOf);
  return c.json(stock);
});

// Command: Receive stock from supplier
app.post("/stock/receive", async (c) => {
  const body = await c.req.json();
  const streamId = `inventory:${body.sku}:${body.warehouseId}`;

  // Retry loop for optimistic concurrency
  for (let attempt = 0; attempt < 3; attempt++) {
    try {
      const events = await eventStore.getStream(streamId);
      const version = events.length;

      await eventStore.append(streamId, [{
        type: "StockReceived" as const,
        eventId: randomUUID(),
        sku: body.sku,
        warehouseId: body.warehouseId,
        quantity: body.quantity,
        purchaseOrderId: body.purchaseOrderId,
        unitCost: body.unitCost,
        supplier: body.supplier,
        timestamp: new Date(),
        correlationId: body.purchaseOrderId,
        causedBy: c.get("userId"),
      }], version);

      return c.json({ success: true }, 201);
    } catch (error) {
      if (error instanceof ConcurrencyError && attempt < 2) continue;
      throw error;
    }
  }
});

// Command: Reserve stock for an order
app.post("/stock/reserve", async (c) => {
  const body = await c.req.json();
  const streamId = `inventory:${body.sku}:${body.warehouseId}`;

  for (let attempt = 0; attempt < 3; attempt++) {
    try {
      const events = await eventStore.getStream(streamId);
      const stock = (await import("../projections/stock-level")).projectStockLevel(events);

      // Business rule: can't reserve more than available
      if (stock.available < body.quantity) {
        return c.json({
          error: "Insufficient stock",
          available: stock.available,
          requested: body.quantity,
        }, 409);
      }

      await eventStore.append(streamId, [{
        type: "StockReserved" as const,
        eventId: randomUUID(),
        sku: body.sku,
        warehouseId: body.warehouseId,
        quantity: body.quantity,
        orderId: body.orderId,
        expiresAt: new Date(Date.now() + 30 * 60 * 1000), // 30 min reservation
        timestamp: new Date(),
        correlationId: body.orderId,
        causedBy: "order-service",
      }], events.length);

      return c.json({ success: true, reserved: body.quantity }, 201);
    } catch (error) {
      if (error instanceof ConcurrencyError && attempt < 2) continue;
      throw error;
    }
  }
});

// Query: Full event history for a SKU (for investigations)
app.get("/stock/:sku/:warehouseId/history", async (c) => {
  const { sku, warehouseId } = c.req.param();
  const streamId = `inventory:${sku}:${warehouseId}`;
  const events = await eventStore.getStream(streamId);
  return c.json({ events, count: events.length });
});

// Query: All events for an order (cross-stream correlation)
app.get("/orders/:orderId/inventory-events", async (c) => {
  const { orderId } = c.req.param();
  const events = await eventStore.getByCorrelation(orderId);
  return c.json({ events });
});

export default app;