Skip to content
ironflow
Ironflow is pre-1.0: APIs can change between releases. Pin a version. What is ready →

CQRS + Event Sourcing

A practical walkthrough you can follow step-by-step. We take one example — placing an order — and carry it through every CQRS layer using Ironflow’s primitives.

If you want the conceptual mapping first, read CQRS with Projections. This tutorial shows the code.

The Scenario

Section titled “The Scenario”

A customer places an order. We want to:

  • Validate the command and produce events
  • Persist events atomically with optimistic concurrency
  • Build two projections: OrderDetailView (customer-facing order page) and CustomerOrdersList (account dashboard)
  • Serve queries from those projections

How Ironflow Shapes This

Section titled “How Ironflow Shapes This”

Ironflow ships the pieces you’d otherwise build yourself. A few differences from a generic CQRS stack are worth calling out up front:

  • No aggregate class. Ironflow has no Aggregate primitive. The consistency boundary lives in an entity stream, and the “aggregate method” is a pattern you write inside a function or handler: read the stream, fold state, validate, append.
  • No event-store table to manage. streams.append is the event store. expectedVersion is optimistic concurrency. The stream is the source of truth.
  • Built-in publication outbox for Ironflow events. streams.append writes the event and an outbox row in one transaction; Ironflow’s built-in outbox worker drains to NATS JetStream asynchronously. Projections consume from there with automatic checkpoints.
  • No checkpoint table to own. Managed projections checkpoint automatically. External projections just implement an idempotent handler.
  • Rebuild is one API call. POST /ironflow.v1.ProjectionService/RebuildProjection.

What’s still on you: choosing causation and correlation IDs and passing them through event metadata. Command completion recovery is application-owned. Step 3 combines ironflow.commandDedup() reservations with idempotent stream appends and causation receipts. Read-your-writes is handled differently per SDK — the Node client exposes projections.waitForEvent(eventId, projectionName) and projections.waitForCatchup(projectionName, { minSeq }); the browser client uses subscribeToProjection for live updates (Step 10).

Project Layout

Section titled “Project Layout”

Below are recommended layouts that match each stack’s conventions. The walkthrough’s snippets map directly onto these paths, and each step notes which files it lives in.

Feature-sliced — one folder per bounded context, with thin top-level composition files:

src/
├── orders/                              # "orders" bounded context
│   ├── commands/
│   │   └── place-order.ts               # PlaceOrderCommand + placeOrderHandler (Steps 1, 3)
│   ├── domain/
│   │   ├── aggregate.ts                 # foldOrder + placeOrder + OrderState (Step 4)
│   │   └── events.ts                    # OrderPlacedEvent, OrderShippedEvent, … (Step 5)
│   ├── projections/
│   │   ├── order-detail-view.ts         # managed projection (Step 8)
│   │   └── customer-orders-list.ts      # partitioned projection (Step 8)
│   ├── read/
│   │   └── queries.ts                   # wraps client.projections.get for the UI (Step 9)
│   └── infra/
│       ├── customer-repo.ts             # enrichment (Step 3)
│       ├── product-catalog.ts
│       └── command-dedup.ts             # KV-backed idempotency (Step 3)
├── api/
│   └── orders.ts                        # HTTP POST handler, returns 202 (Step 2)
├── worker.ts                            # createWorker(...) — projections + functions (Step 8)
└── ironflow.ts                          # createClient() shared by api + worker

Conventions:

  • kebab-case.ts for filenames, PascalCase for types, camelCase for functions.
  • One command per file under commands/; the handler lives beside its command type.
  • Each projection is its own small module exporting a createProjection(...) call.
  • worker.ts is the composition root for the read side; api/orders.ts is the write-side entry.
  • For Next.js, app/api/orders/route.ts forwards to src/api/orders.ts, and app/orders/[orderId]/page.tsx consumes src/orders/read/queries.ts.

These layouts scale from a small service to dozens of bounded contexts without shuffling files. Start small — the cqrs-order example in the repo uses a flat lib/ for readability; graduate to one of the above once you have more than one context.

Step 1 — Define the Command

Section titled “Step 1 — Define the Command”

Commands are intent. Imperative, named in present tense, carrying IDs and enough context to attempt the action — not enriched data.

type PlaceOrderCommand = {
  // Domain payload
  data: {
    orderId: string;
    customerId: string;
    items: { productId: string; qty: number }[];
    shippingAddress: { street: string; city: string; zip: string; country: string };
  };
  // Plumbing — not a domain concern, propagates onto events
  metadata: {
    commandId: string;        // idempotency key
    issuedAt: string;
    issuedBy: string;         // auth context
    correlationId?: string;   // originating request/workflow
    tenantId?: string;
    traceId?: string;
  };
};
from dataclasses import dataclass, field




@dataclass
class PlaceOrderData:
    order_id: str
    customer_id: str
    items: list[dict]             # [{"productId": ..., "qty": ...}]
    shipping_address: dict        # {"street", "city", "zip", "country"}




@dataclass
class CommandMetadata:
    command_id: str               # idempotency key
    issued_at: str
    issued_by: str                # auth context
    correlation_id: str | None = None   # originating request/workflow
    tenant_id: str | None = None
    trace_id: str | None = None




@dataclass
class PlaceOrderCommand:
    data: PlaceOrderData          # domain payload
    metadata: CommandMetadata     # plumbing — not a domain concern, propagates onto events

Keep the metadata split deliberate. data is what the domain needs to decide; metadata is cross-cutting plumbing that rides along on the resulting events. Generate commandId at the client or API boundary for dedup.

Step 2 — API Endpoint Receives the Command

Section titled “Step 2 — API Endpoint Receives the Command”

Expose the command via a thin HTTP handler. In Ironflow you have two patterns:

  • Next.js / Express / Lambda route that authenticates the user, constructs the command, and calls your handler directly.
  • Ironflow function triggered by an event if you want durable retries on the dispatch itself. The route emits the command as an event and the function runs the handler.

Either way, the endpoint is thin: authenticate, deserialize, dispatch, return 202 Accepted.

// app/api/orders/route.ts (Next.js)
export async function POST(req: Request) {
  const user = await authenticate(req);
  const body = await req.json();


  const command: PlaceOrderCommand = {
    data: {
      orderId: body.orderId,
      customerId: user.id,
      items: body.items,
      shippingAddress: body.shippingAddress,
    },
    metadata: {
      commandId: crypto.randomUUID(),
      issuedAt: new Date().toISOString(),
      issuedBy: user.id,
      correlationId: req.headers.get("x-request-id") ?? undefined,
      tenantId: user.orgId,
      traceId: req.headers.get("traceparent") ?? undefined,
    },
  };


  await placeOrderHandler(command);
  return Response.json(
    { orderId: command.data.orderId, commandId: command.metadata.commandId },
    { status: 202 }
  );
}

The same endpoint in Python:

# app/api.py (FastAPI)
import uuid
from datetime import datetime, timezone


from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


api = FastAPI()




@api.post("/api/orders")
async def post_order(req: Request):
    user = await authenticate(req)
    body = await req.json()


    command = PlaceOrderCommand(
        data=PlaceOrderData(
            order_id=body["orderId"],
            customer_id=user.id,
            items=body["items"],
            shipping_address=body["shippingAddress"],
        ),
        metadata=CommandMetadata(
            command_id=str(uuid.uuid4()),
            issued_at=datetime.now(timezone.utc).isoformat(),
            issued_by=user.id,
            correlation_id=req.headers.get("x-request-id"),
            tenant_id=user.org_id,
            trace_id=req.headers.get("traceparent"),
        ),
    )


    place_order_handler(command)
    return JSONResponse(
        {"orderId": command.data.order_id, "commandId": command.metadata.command_id},
        status_code=202,
    )

Return 202 Accepted because read models are eventually consistent. The write succeeded; the projection might not be updated yet. See Step 10.

Step 3 — Command Handler

Section titled “Step 3 — Command Handler”

The handler reserves the command, checks the stream for a completion receipt, fetches enrichment data, decides, and appends. Business rules live in the domain function called from the handler — not the handler itself.

import { createClient, NonRetryableError, type StreamEvent } from "@ironflow/node";


const ironflow = createClient({ apiKey: process.env.IRONFLOW_API_KEY! });
type DedupEntry = {
  orderId: string;
  claimedAt: string;
  completedAt?: string;
  entityVersion?: number;
};
const dedup = ironflow.commandDedup<DedupEntry>("order-commands");


type PlaceOrderResult = {
  orderId: string;
  entityVersion: number;
  dedup: boolean;
};


async function placeOrderHandler(
  cmd: PlaceOrderCommand,
): Promise<PlaceOrderResult> {
  const { data, metadata } = cmd;


  const claim: DedupEntry = {
    orderId: data.orderId,
    claimedAt: new Date().toISOString(),
  };
  const prior = await dedup.tryClaim(metadata.commandId, claim);
  if (prior !== null) {
    if (prior.orderId !== data.orderId) {
      throw new NonRetryableError("Command ID already used for another order");
    }
    // Empty defaults ("" / 0) in a serialized pending record do not prove completion.
    const { completedAt, entityVersion } = prior;
    if (
      typeof completedAt === "string" && completedAt !== "" &&
      typeof entityVersion === "number" && Number.isSafeInteger(entityVersion) && entityVersion > 0
    ) {
      return { orderId: prior.orderId, entityVersion, dedup: true };
    }
  }


  const finish = async (entityVersion: number, recovered: boolean): Promise<PlaceOrderResult> => {
    await dedup.finalize(metadata.commandId, {
      ...claim,
      claimedAt: prior?.claimedAt ?? claim.claimedAt,
      completedAt: new Date().toISOString(),
      entityVersion,
    });
    return { orderId: data.orderId, entityVersion, dedup: recovered };
  };
  const isReceipt = (ev: StreamEvent) =>
    ev.name === "order.placed" && ev.metadata?.causationId === metadata.commandId;
  const { events } = await ironflow.streams.read(data.orderId);
  const receipt = events.find(isReceipt);
  if (receipt) return finish(receipt.entityVersion, true);


  const customer = await customerRepo.getById(data.customerId);
  const products = await productCatalog.getMany(data.items.map((i) => i.productId));
  const state = foldOrder(events);
  const [ev] = placeOrder(state, data, customer, products, metadata.issuedAt);
  const eventMetadata = {
    causationId: metadata.commandId,
    correlationId: metadata.correlationId ?? metadata.commandId,
    tenantId: metadata.tenantId,
    traceId: metadata.traceId,
    issuedBy: metadata.issuedBy,
  };


  let ack: { eventId: string };
  try {
    ack = await ironflow.streams.append(
      data.orderId,
      { entityType: "order", name: ev.name, data: ev.data },
      {
        expectedVersion: state.version,
        idempotencyKey: `cqrs-order:place-order:${encodeURIComponent(data.orderId)}:${encodeURIComponent(metadata.commandId)}`,
        metadata: eventMetadata,
      },
    );
  } catch (err) {
    const { events: committed } = await ironflow.streams.read(data.orderId);
    const receipt = committed.find(isReceipt);
    if (receipt) return finish(receipt.entityVersion, true);
    throw err;
  }
  // An acknowledgement is not proof: the engine's environment-wide key index can answer
  // with a non-entity event or another stream's event. Only this command's receipt is.
  const { events: committed } = await ironflow.streams.read(data.orderId);
  const placed = committed.find(isReceipt);
  if (!placed) {
    throw new NonRetryableError("Append acknowledged without a committed entity version for this command");
  }
  // Finalization failures leave the receipt recoverable on the next retry.
  return finish(placed.entityVersion, placed.id !== ack.eventId);
}

The same recovery flow in Python:

import os
from datetime import datetime, timezone
from urllib.parse import quote


from protobuf.wkt import Struct
from ironflow import IronflowClient, IronflowRPC
from ironflow.rpc.v1 import AppendEventRequest, ReadStreamRequest


client = IronflowClient(api_key=os.environ["IRONFLOW_API_KEY"])
rpc = IronflowRPC(api_key=os.environ["IRONFLOW_API_KEY"])
dedup = client.command_dedup("order-commands")




def place_order_handler(cmd: PlaceOrderCommand):
    data, metadata = cmd.data, cmd.metadata
    claim = {"orderId": data.order_id, "claimedAt": datetime.now(timezone.utc).isoformat()}
    prior = dedup.try_claim(metadata.command_id, claim)
    if prior is not None:
        if prior["orderId"] != data.order_id:
            raise ValueError("Command ID already used for another order")
        # Empty defaults ("" / 0) in a serialized pending record do not prove completion.
        completed_at, version = prior.get("completedAt"), prior.get("entityVersion")
        if isinstance(completed_at, str) and completed_at and type(version) is int and 0 < version <= 2**53 - 1:
            return {"orderId": prior["orderId"], "entityVersion": version, "dedup": True}


    def finish(entity_version, recovered):
        dedup.finalize(metadata.command_id, {
            **claim, "completedAt": datetime.now(timezone.utc).isoformat(), "entityVersion": entity_version,
        })
        return {"orderId": data.order_id, "entityVersion": entity_version, "dedup": recovered}


    def find_receipt(events):
        for ev in events:
            meta = ev.metadata.to_python() if ev.metadata is not None else {}
            if ev.name == "order.placed" and meta.get("causationId") == metadata.command_id:
                return ev
        return None


    events = rpc.streams.read_stream(ReadStreamRequest(entity_id=data.order_id)).events
    receipt = find_receipt(events)
    if receipt is not None:
        return finish(receipt.entity_version, True)


    customer = customer_repo.get_by_id(data.customer_id)
    products = product_catalog.get_many([i["productId"] for i in data.items])
    state = fold_order(events)
    ev, = place_order(state, data, customer, products, metadata.issued_at)
    event_metadata = Struct.from_python({k: v for k, v in {
        "causationId": metadata.command_id,
        "correlationId": metadata.correlation_id or metadata.command_id,
        "tenantId": metadata.tenant_id,
        "traceId": metadata.trace_id,
        "issuedBy": metadata.issued_by,
    }.items() if v is not None})
    try:
        res = rpc.streams.append_event(AppendEventRequest(
            entity_id=data.order_id, entity_type="order", event_name=ev["name"],
            data=Struct.from_python(ev["data"]), expected_version=state["version"],
            idempotency_key=f"cqrs-order:place-order:{quote(data.order_id, safe='')}:{quote(metadata.command_id, safe='')}", metadata=event_metadata,
        ))
    except Exception:
        committed = rpc.streams.read_stream(ReadStreamRequest(entity_id=data.order_id)).events
        receipt = find_receipt(committed)
        if receipt is not None:
            return finish(receipt.entity_version, True)
        raise
    # An acknowledgement is not proof: the engine's environment-wide key index can answer
    # with a non-entity event or another stream's event. Only this command's receipt is.
    placed = find_receipt(rpc.streams.read_stream(ReadStreamRequest(entity_id=data.order_id)).events)
    if placed is None:
        raise ValueError("Append acknowledged without a committed entity version for this command")
    return finish(placed.entity_version, placed.id != res.event_id)

Step 4 — The “Aggregate” (a Pattern, Not a Class)

Section titled “Step 4 — The “Aggregate” (a Pattern, Not a Class)”

A pure function that folds events into state, and a pure function that decides new events given current state + command. No framework class required.

type OrderState = {
  id: string;
  status: "placed" | "shipped" | "cancelled" | null;
  items: LineItem[];
  customerId: string | null;
  totalAmount: number;
  version: number;
};


function foldOrder(events: StreamEvent[]): OrderState {
  const initial: OrderState = {
    id: "", status: null, items: [], customerId: null, totalAmount: 0, version: 0
  };
  return events.reduce((state, ev) => {
    switch (ev.name) {
      case "order.placed":
        return { ...state, id: ev.data.orderId, status: "placed",
                 items: ev.data.items, customerId: ev.data.customer.id,
                 totalAmount: ev.data.totalAmount, version: ev.entityVersion };
      case "order.shipped":
        return { ...state, status: "shipped", version: ev.entityVersion };
      case "order.cancelled":
        return { ...state, status: "cancelled", version: ev.entityVersion };
      default:
        return { ...state, version: ev.entityVersion };
    }
  }, initial);
}


function placeOrder(
  state: OrderState,
  data: PlaceOrderCommand["data"],
  customer: Customer,
  products: Product[],
  occurredAt: string,
) {
  if (state.status !== null) throw new Error("Order already exists");
  if (customer.isBlocked) throw new Error("Customer cannot place orders");
  if (products.some(p => !p.isAvailable)) throw new Error("Product unavailable");


  const lineItems = products.map(p => ({
    productId: p.id,
    name: p.name,              // embedded snapshot — frozen forever
    price: p.currentPrice,     // embedded snapshot
    qty: data.items.find(i => i.productId === p.id)!.qty,
  }));
  const total = lineItems.reduce((sum, li) => sum + li.price * li.qty, 0);


  // Domain-only payload. Plumbing (causation/correlation/tenant/trace)
  // is added by the handler when it appends — see Step 3 and Step 5.
  return [{
    name: "order.placed",
    data: {
      orderId: data.orderId,
      customer: { id: customer.id, name: customer.name, email: customer.email },
      items: lineItems,
      shippingAddress: data.shippingAddress,
      totalAmount: total,
      occurredAt,
    },
  }];
}

The same pair in Python:

def fold_order(events) -> dict:
    state = {"id": "", "status": None, "items": [], "customerId": None, "totalAmount": 0, "version": 0}
    for ev in events:
        data = ev.data.to_python() if ev.data is not None else {}
        if ev.name == "order.placed":
            state = {**state, "id": data["orderId"], "status": "placed", "items": data["items"],
                     "customerId": data["customer"]["id"], "totalAmount": data["totalAmount"]}
        elif ev.name == "order.shipped":
            state = {**state, "status": "shipped"}
        elif ev.name == "order.cancelled":
            state = {**state, "status": "cancelled"}
        state = {**state, "version": ev.entity_version}
    return state




def place_order(state, data, customer, products, occurred_at):
    if state["status"] is not None:
        raise ValueError("Order already exists")
    if customer.is_blocked:
        raise ValueError("Customer cannot place orders")
    if any(not p.is_available for p in products):
        raise ValueError("Product unavailable")


    qty = {i["productId"]: i["qty"] for i in data.items}
    line_items = [
        {"productId": p.id, "name": p.name, "price": p.current_price, "qty": qty[p.id]}  # frozen snapshot
        for p in products
    ]
    total = sum(li["price"] * li["qty"] for li in line_items)


    # Domain-only payload. Plumbing is added by the handler when it appends.
    return [{
        "name": "order.placed",
        "data": {
            "orderId": data.order_id,
            "customer": {"id": customer.id, "name": customer.name, "email": customer.email},
            "items": line_items,
            "shippingAddress": data.shipping_address,
            "totalAmount": total,
            "occurredAt": occurred_at,
        },
    }]

The event payload embeds customer.name, price, name, etc. Those values are frozen in the event forever — this is what makes rebuilds work.

Step 5 — Event Shape and Metadata

Section titled “Step 5 — Event Shape and Metadata”

Every event has three parts: payload (data, domain facts), user metadata (metadata, cross-cutting plumbing like tenant/trace IDs), and platform metadata (stamped by Ironflow).

{
  // Platform-assigned (returned from read/subscribe)
  id: "evt_xyz789",           // eventId
  name: "order.placed",
  entityVersion: 1,           // position within this entity's stream
  version: 1,                 // schema version of this event type
  timestamp: "2026-04-17T14:22:01Z",
  source: "api",              // api | cron | webhook


  // You control — domain payload
  data: {
    orderId, customer, items, shippingAddress, totalAmount,
  },


  // You control — cross-cutting metadata
  metadata: {
    causationId: "cmd_a1b2c3",
    correlationId: "req_789",
    tenantId: "org_123",
    traceId: "otel_abc",
    issuedBy: "user_456",
  }
}

SDK support today

Section titled “SDK support today”

  • ironflow.emit(name, data, { metadata }) — first-class. Metadata is delivered to push-mode function handlers as event.metadata.

    await ironflow.emit("order.placed", orderData, {
      metadata: { causationId: metadata.commandId, correlationId: metadata.commandId, tenantId: metadata.tenantId },
    });

  • ironflow.streams.append(entityId, { entityType, name, data }, { expectedVersion, metadata }) — first-class. Metadata rides alongside the event (not inside data) and is persisted to the store:

    await ironflow.streams.append(data.orderId, {
      entityType: "order",
      name: "order.placed",
      data: domainPayload,
    }, {
      expectedVersion: state.version,
      metadata: { causationId: metadata.commandId, correlationId: metadata.commandId, tenantId: metadata.tenantId },
    });

    streams.read returns metadata on each event, and downstream handlers and projections read it via event.metadata.

    In Python, both requests take a metadata Struct:

    meta = Struct.from_python({"causationId": metadata.command_id, "tenantId": metadata.tenant_id})
    rpc.events.emit(TriggerRequest(event="order.placed", data=Struct.from_python(order_data), metadata=meta))
    rpc.streams.append_event(AppendEventRequest(
        entity_id=data.order_id, entity_type="order", event_name="order.placed",
        data=Struct.from_python(domain_payload), expected_version=state["version"], metadata=meta,
    ))

    Python projection handlers read it as event.metadata, and Python function handlers as ctx.event.metadata.

  • Pull-mode handlers and projections — event.metadata (Go: event.Metadata) is delivered to push-mode functions, pull-mode workers, and projection handlers alike. Causation/correlation/tenant IDs you stamp on emit() or streams.append() are available everywhere.

Why keep metadata separate

Section titled “Why keep metadata separate”

Metadata is for plumbing, not domain semantics. Projections usually ignore it; logs, audits, and tracing consume it. Keeping it out of data stops cross-cutting concerns from polluting your domain events — and lets the SDK thread it uniformly across push/pull/projection handlers.

For schema evolution (adding fields, restructuring), use upcasters — see the Event Versioning guide.

Step 6 — Persisting Events

Section titled “Step 6 — Persisting Events”

streams.append is the event store. One call per event, atomic, with optimistic concurrency.

const { entityVersion, eventId } = await ironflow.streams.append(
  "order-123",
  { entityType: "order", name: "order.placed", data: { /* ... */ } },
  { expectedVersion: 0 }   // -1 (default) = no guard, 0 = create-only, N = exact prior version
);
res = rpc.streams.append_event(AppendEventRequest(
    entity_id="order-123", entity_type="order", event_name="order.placed",
    data=Struct.from_python({...}),
    expected_version=0,  # -1 = no guard, 0 = create-only, N = exact prior version
))
entity_version, event_id = res.entity_version, res.event_id
res, err := client.AppendStreamEvent(ctx, "order-123", ironflow.AppendEventInput{
    EntityType: "order",
    Name:       "order.placed",
    Data:       map[string]any{ /* ... */ },
}, ironflow.WithExpectedVersion(0)) // -1 (default) = no guard, 0 = create-only, N = exact prior version
if err != nil {
    return err
}
entityVersion, eventID := res.EntityVersion, res.EventID

The Go client covers the rest of the write side with the same shapes: client.ReadStream (Step 3), ironflow.WithAppendMetadata (Step 5), ironflow.NewCommandDedup (Step 3), ironflow.CreateProjection with ProjectionConfig.PartitionKey (Step 8), and client.WaitForEvent (Step 10). See the Go SDK reference.

AppendOptions.expectedVersion defaults to -1 (no guard). Pass 0 to require an empty stream (create-only), or pass the version observed when you read the stream to guard against concurrent writers.

Behavior:

  • Create-only: expectedVersion: 0 fails if the stream already has events → HTTP 409.
  • Optimistic concurrency: expectedVersion: N fails if another writer advanced the stream first. Catch the 409 and retry (reload + re-decide + re-append).
  • Atomicity: the append is atomic per event. Keep the per-command event count small — most CQRS decisions emit one event anyway.
  • sequence: not returned at all today (undefined) because the outbox publishes asynchronously. Use eventId with projections.waitForEvent() for read-your-writes (Step 10).

That’s the whole event store. No global_position column to maintain, no unique constraint to add, no migration.

Step 7 — Publishing through Ironflow’s outbox

Section titled “Step 7 — Publishing through Ironflow’s outbox”

This example uses Ironflow entity streams as authoritative state. Ironflow owns the outbox that publishes those events to NATS; no application-owned publisher is needed for this path.

When streams.append returns success, the event is committed to the store and an outbox row is written in the same transaction. A per-node outbox worker drains the outbox asynchronously to NATS JetStream, where projections consume the entity notifications with durable NATS consumers. Checkpoints are tracked automatically. Because publish is async, streams.append does not return a sequence at all — use projections.waitForEvent(eventId, ...) when you need read-your-writes (see Step 10 and Implementation Order → Step 4).

A mutation in a separate application database does not join this transaction, even through the same SDK object. Idempotency keys protect retries, not cross-database atomicity. If your application database is authoritative, use an application-owned transactional outbox and relay with stable keys, acknowledged only after confirmed Ironflow acceptance. Alternatively, use idempotent steps with compensation or eventual repair. Compensation is not ACID rollback. See transaction boundaries and integration choices for the two-transaction diagram.

Step 8 — Projectors Build Read Models

Section titled “Step 8 — Projectors Build Read Models”

Define each projection with createProjection. Ironflow subscribes, feeds events, and stores the state you return.

OrderDetailView — managed projection

Section titled “OrderDetailView — managed projection”
import { createProjection, createWorker } from "@ironflow/node";


// The handler receives (state, event, ctx). The typed event surface is
// { name, data }; timestamps and ids come off ctx.event (a Date for ctx.event.timestamp).
// Managed reducers must return FRESH state — never mutate the argument. Rebuild
// and the live tail can both apply the same event, so build new objects along the
// path you touch and leave untouched siblings shared by reference. See the
// Reducer Contract in /explanation/projections/.
const orderDetail = createProjection({
  name: "order-detail-view",
  events: ["order.placed", "order.shipped", "order.cancelled"],
  initialState: () => ({ orders: {} as Record<string, OrderDetail> }),
  handler: (state, event, ctx) => {
    const data = event.data as any;
    const ts = ctx.event.timestamp.toISOString();
    const orderId = data.orderId;
    const existing = state.orders[orderId] ?? {};


    switch (event.name) {
      case "order.placed":
        return { ...state, orders: { ...state.orders, [orderId]: {
          orderId,
          customerId: data.customer.id,
          customerName: data.customer.name,
          customerEmail: data.customer.email,
          items: data.items,
          shippingAddress: data.shippingAddress,
          totalAmount: data.totalAmount,
          status: "placed",
          placedAt: ts,
        } } };
      case "order.shipped":
        return { ...state, orders: { ...state.orders, [orderId]:
                 { ...existing, status: "shipped",
                   trackingNumber: data.trackingNumber,
                   shippedAt: ts } } };
      case "order.cancelled":
        return { ...state, orders: { ...state.orders, [orderId]:
                 { ...existing, status: "cancelled",
                   cancelledAt: ts,
                   cancellationReason: data.reason } } };
      default:
        // Unrelated event — return state unchanged.
        return state;
    }
  },
});

In Python:

from ironflow.projection import create_projection




# Managed reducers must return FRESH state — never mutate the argument.
def order_detail_handler(state, event, ctx):
    data, ts = event.data, ctx.event.timestamp  # ISO string
    order_id = data["orderId"]
    existing = state["orders"].get(order_id, {})


    if event.name == "order.placed":
        row = {
            "orderId": order_id,
            "customerId": data["customer"]["id"],
            "customerName": data["customer"]["name"],
            "customerEmail": data["customer"]["email"],
            "items": data["items"],
            "shippingAddress": data["shippingAddress"],
            "totalAmount": data["totalAmount"],
            "status": "placed",
            "placedAt": ts,
        }
    elif event.name == "order.shipped":
        row = {**existing, "status": "shipped", "trackingNumber": data.get("trackingNumber"), "shippedAt": ts}
    elif event.name == "order.cancelled":
        row = {**existing, "status": "cancelled", "cancelledAt": ts, "cancellationReason": data.get("reason")}
    else:
        return state  # Unrelated event — return state unchanged.
    return {**state, "orders": {**state["orders"], order_id: row}}




order_detail = create_projection(
    name="order-detail-view",
    events=["order.placed", "order.shipped", "order.cancelled"],
    initial_state=lambda: {"orders": {}},
    handler=order_detail_handler,
)

CustomerOrdersList — partitioned managed projection

Section titled “CustomerOrdersList — partitioned managed projection”

Partition by customerId so each customer’s dashboard is a separate state blob.

const customerOrders = createProjection({
  name: "customer-orders-list",
  events: ["order.placed", "order.shipped", "order.cancelled"],
  partitionKey: "$.customer.id",
  initialState: () => ({ orders: [] as OrderSummary[] }),
  handler: (state, event, ctx) => {
    const data = event.data as any;
    if (event.name === "order.placed") {
      const row = {
        orderId: data.orderId,
        placedAt: ctx.event.timestamp.toISOString(),
        totalAmount: data.totalAmount,
        status: "placed",
        summary: `${data.items.length} items`,
      };
      // Filter first, then prepend: a plain unshift would add a second copy of
      // the row if the same order.placed is redelivered or replayed by a rebuild.
      return { ...state,
               orders: [row, ...state.orders.filter(o => o.orderId !== data.orderId)] };
    }
    return { ...state, orders: state.orders.map(o =>
      o.orderId === data.orderId
        ? { ...o, status: event.name.split(".")[1] }
        : o) };
  },
});
def customer_orders_handler(state, event, ctx):
    data = event.data
    if event.name == "order.placed":
        row = {
            "orderId": data["orderId"],
            "placedAt": ctx.event.timestamp,
            "totalAmount": data["totalAmount"],
            "status": "placed",
            "summary": f"{len(data['items'])} items",
        }
        # Filter first, then prepend, so a redelivered order.placed adds no second row.
        return {**state, "orders": [row] + [o for o in state["orders"] if o["orderId"] != data["orderId"]]}
    status = event.name.split(".")[1]
    return {**state, "orders": [{**o, "status": status} if o["orderId"] == data["orderId"] else o
                                for o in state["orders"]]}




customer_orders = create_projection(
    name="customer-orders-list",
    events=["order.placed", "order.shipped", "order.cancelled"],
    partition_key="$.customer.id",
    initial_state=lambda: {"orders": []},
    handler=customer_orders_handler,
)

Start the worker

Section titled “Start the worker”
const worker = createWorker({
  functions: [],
  projections: [orderDetail, customerOrders],
});
// Not awaited: worker.start() is the poll loop and resolves only on stop().
worker.start();
from ironflow.worker import Worker


# run() owns the event loop and returns only when the worker stops.
Worker(functions=[], projections=[order_detail, customer_orders]).run()

Same events, two independent read models, independent checkpoints. Add, remove, or rebuild projections without touching the write side.

For side-effect sync (Elasticsearch, Algolia, Redis) use an external projection — omit initialState and make the handler async (event, ctx) => Promise<void>. Use ctx.event.id as the idempotency key so replays are safe.

Step 9 — Queries Hit Read Models

Section titled “Step 9 — Queries Hit Read Models”

Queries become trivial — no aggregate loading, no event folding.

import { ironflow } from "@ironflow/browser";


// Call once at app startup. Browser `ironflow` is a singleton.
ironflow.configure({ serverUrl: "https://api.example.com" });


const detail = await ironflow.getProjection("order-detail-view");
const order = (detail.state as any).orders["order-123"];


const dashboard = await ironflow.getProjection("customer-orders-list", {
  partition: "cust-456",
});

Real-time updates use subscribeToProjection, which pushes new state over the browser client’s transport (ConnectRPC server streaming by default; transport: "websocket" is opt-in) as events arrive:

const sub = await ironflow.subscribeToProjection("order-detail-view", {
  onUpdate: (state, event) => {
    // state is the new projection state; event is { id, name }
  },
});
// sub.unsubscribe() when you're done.

Step 10 — Handling Eventual Consistency in the UI

Section titled “Step 10 — Handling Eventual Consistency in the UI”

Because Step 2 returns 202 before projections are updated, the client might redirect to /orders/{orderId} and get a 404. Options:

  • Wait for the write on the server (recommended for entity-stream events) — call ironflow.projections.waitForEvent(eventId, "order-detail-view", { partition, timeoutMs: 5000 }) using the eventId returned from streams.append, then return the 202 only after the projection has applied it. In Go, call client.WaitForEvent. In Python, call rpc.projections.wait_for_event(v1.WaitForEventRequest(event_id=..., projection=...)). This is the canonical read-your-writes path because streams.append returns no sequence at all (only entityVersion and eventId), so you have no minSeq to hand to waitForCatchup yourself. waitForEvent resolves the eventId for you. It returns the resolved sequence as targetSeq, which is the minSeq that waitForCatchup takes.
  • Optimistic rendering — reconstruct the expected shape from the command, render it, then reconcile when the projection catches up. Best fit for SPAs.
  • Subscribe after redirect — call ironflow.subscribeToProjection("order-detail-view", { onUpdate }) from the browser and wait for the orderId key to appear in the pushed state. Usually <200ms.
  • Just accept the lag — for many domains, 200–500ms is invisible.

Step 11 — Rebuilding Projections

Section titled “Step 11 — Rebuilding Projections”

The payoff of the whole architecture. When a projection schema changes or a bug is fixed, you rebuild:

Terminal window
# Reset and rebuild from the beginning of the event log
curl -X POST $IRONFLOW_SERVER_URL/ironflow.v1.ProjectionService/RebuildProjection -H 'Content-Type: application/json' -d '{"name":"order-detail-view"}'

Ironflow clears the materialized state, resets the checkpoint, and replays all historical events through your handler. The projection status transitions rebuilding → active when it catches up; the dashboard Projections page shows lag in real time. Poll POST /ironflow.v1.ProjectionService/GetRebuildJob or watch the ironflow_projection_rebuild_* Prometheus metrics for progress.

For a safer cutover, keep the old projection online while you rebuild a v2 alongside it:

  1. Deploy order-detail-view-v2 with the new handler/schema.
  2. Wait for the v2 lag to reach zero.
  3. Flip your query code from v1 to v2.
  4. Delete v1: DELETE /api/v1/projections/order-detail-view.

Zero downtime, no data loss, and you can diff outputs during cutover. This is why events carry embedded snapshots — v2 can rebuild itself from the event log without depending on whatever customer/product data happens to exist today.

The Cheat Sheet

Section titled “The Cheat Sheet”

Write side: API endpoint → command handler (dedup, enrich, fold, decide) → streams.append with expectedVersion. Events hit NATS automatically.

Read side: createProjection(...) × N, each with independent checkpoints → queries hit getProjection or REST, with no joins or logic.

Enrichment: events carry ID and snapshot values for decision-relevant attributes. Rebuilds stay self-contained.

Consistency boundary: one entity stream is strongly consistent via expectedVersion. Everything downstream is eventually consistent.

Rebuild: POST /ironflow.v1.ProjectionService/RebuildProjection. Treat it as a first-class capability.

What Ironflow Doesn’t Do (Yet)

Section titled “What Ironflow Doesn’t Do (Yet)”

Be honest with your team:

  • Choreography sagas — Ironflow does orchestrated sagas via function compensation. Chain emits manually if you need choreography.

Implementation Order

Section titled “Implementation Order”

Ship velocity matters, but a few disciplines belong in the foundation rather than bolted on later. Concurrency control, idempotency, and a testing idiom are what separate event sourcing that scales from event sourcing that becomes a regret. The ordering below front-loads those, then layers capabilities outward.

Step 0 — Model before you code

Section titled “Step 0 — Model before you code”

Event-storm the slice with the people who know the domain. Name the aggregate. Define the consistency boundary (what must be transactionally consistent vs. what can lag). Pick past-tense, domain-language event names (OrderPlaced, not CreateOrderEvent). Skipping this is how teams end up with OrderManager god-streams that span half the business.

References: Eric Evans (DDD), Alberto Brandolini (Event Storming), Vaughn Vernon (“Implementing Domain-Driven Design”).

Step 1 — Walking skeleton with discipline baked in

Section titled “Step 1 — Walking skeleton with discipline baked in”

One stream, one event type, one projection, one query — end-to-end. From day one include:

  • expectedVersion on every append. Optimistic concurrency is the defining invariant of an event-sourced aggregate. Without it you have an event log, not event sourcing. Even with one writer, ship the guard.
  • Given-When-Then aggregate tests. Given prior events, when command, then new events. This is the ES testing idiom — your domain logic should be unit-testable without a database, a server, or a clock.

Step 2 — Command idempotency at the API boundary

Section titled “Step 2 — Command idempotency at the API boundary”

At-least-once delivery is a day-one concern. The moment you have an HTTP endpoint and a retrying client, duplicates happen. A KV reservation keyed by commandId does not prove completion. Pair it with a stable append idempotency key and recover pending claims from the committed event, as Step 3 shows. Do not treat idempotency as an optimization; it is correctness.

Step 3 — Second projection + correlation IDs on every event

Section titled “Step 3 — Second projection + correlation IDs on every event”

Add a second projection on the same events to prove the “one event, many views” principle. While you’re touching the write side, stamp causationId, correlationId, and tenantId on every event (Step 5 of the walkthrough). Debugging an event-sourced system without correlation chains is brutal; adding them later is harder than it sounds because you can’t backfill what wasn’t recorded.

Step 4 — Concurrency retry loop + read-your-writes UX

Section titled “Step 4 — Concurrency retry loop + read-your-writes UX”

Wrap the load–decide–append cycle in a retry that catches the 409 from streams.append, reloads the stream, re-runs the aggregate, and re-appends. For read-your-writes, use ironflow.projections.waitForEvent(eventId, projectionName, { partition, timeoutMs: 5000 }) — pass the eventId returned by streams.append. The server resolves the event ID to its NATS sequence once the outbox worker publishes, then blocks until the projection has processed it (or the deadline fires). Under the transactional outbox (#487), streams.append does not return a usable sequence anymore; it is omitted entirely (undefined) because the publish happens asynchronously after the transaction commits. This is where product usually pushes back on CQRS — have an answer ready.

Step 5 — Second event type + upcasters + rebuild rehearsal

Section titled “Step 5 — Second event type + upcasters + rebuild rehearsal”

You will version events on day one whether you planned to or not. Before your second deploy, add an upcaster for one event and rehearse a full projection rebuild against prod-volume data. Teams discover at 3 a.m. that their rebuild takes 40 hours. Measure it now, while the stakes are low.

Step 6 — External projections (search, cache, notifications)

Section titled “Step 6 — External projections (search, cache, notifications)”

Now that the core loop is solid, add side-effect projections (Elasticsearch, Algolia, Redis, email). These need idempotent handlers — use ctx.event.id as the dedup key so replays are safe. The discipline you built in Steps 1–4 is what makes these tractable.

Step 7 — Process managers when cross-aggregate workflows appear

Section titled “Step 7 — Process managers when cross-aggregate workflows appear”

The first time one aggregate needs to react to another’s events, introduce a process manager (saga). In Ironflow this is an orchestrated function with compensation. Resist the urge to do this earlier — premature sagas are worse than no sagas.

Step 8 — Partition high-volume projections; snapshot if measured

Section titled “Step 8 — Partition high-volume projections; snapshot if measured”

Partition the projection that’s actually hot (use the partition browser to see lag per key). Add snapshots only when fold latency exceeds your budget on a real stream — most streams stay short enough that snapshotting is unnecessary complexity. Measure first.

Infrastructure concerns like clustering and multi-node projections come after the core loop is rock-solid. See the Event Sourcing how-to guides for each piece in depth.