Event Versioning

As your system evolves, event schemas change. Ironflow supports event versioning with upcaster functions that automatically transform old events to the latest schema when they’re consumed.

How It Works

Events are stored with their original version. When a consumer reads them, upcasters transform the data through each version step until it reaches the latest schema.

Three language-specific rules to keep in mind:

• TypeScript — each defineEvent({ version: N, upcast }) registers the N-1 → N step. You do not specify the source version; it’s implicit.
• Go — upcasters.Register(name, from, to, fn) is explicit, but to must equal from + 1. The chain walks one step at a time; a registered 1 → 3 hop is unreachable.
• Python — UpcasterRegistry.register(name, from_version, to_version, fn) is explicit like Go, but a hop can skip versions: the chain jumps straight to the registered to_version (Node semantics). A redacted payload passes through unchanged, and a broken chain fails the job as non-retryable — unlike Go, which silently passes the old data through instead.

Defining Versioned Events

TypeScript

import { defineEvent } from "@ironflow/core";

// Version 1 — original schema
const userCreatedV1 = defineEvent({
  name: "user.created",
  version: 1,
});

// Version 2 — added email field
const userCreatedV2 = defineEvent({
  name: "user.created",
  version: 2,
  upcast: (v1Data) => ({
    ...v1Data,
    email: v1Data.email ?? "unknown@example.com",
  }),
});

// Version 3 — added role field
const userCreatedV3 = defineEvent({
  name: "user.created",
  version: 3,
  upcast: (v2Data) => ({
    ...v2Data,
    role: v2Data.role ?? "member",
  }),
});

Go

import (
    "encoding/json"
    "github.com/sahina/ironflow-go/ironflow"
)

upcasters := ironflow.NewUpcasterRegistry()

// v1 → v2: add email field
upcasters.Register("user.created", 1, 2, func(data json.RawMessage) (json.RawMessage, error) {
    var m map[string]any
    if err := json.Unmarshal(data, &m); err != nil {
        return nil, err
    }
    if _, ok := m["email"]; !ok {
        m["email"] = "unknown@example.com"
    }
    return json.Marshal(m)
})

// v2 → v3: add role field
upcasters.Register("user.created", 2, 3, func(data json.RawMessage) (json.RawMessage, error) {
    var m map[string]any
    if err := json.Unmarshal(data, &m); err != nil {
        return nil, err
    }
    if _, ok := m["role"]; !ok {
        m["role"] = "member"
    }
    return json.Marshal(m)
})

Python

from ironflow import UpcasterRegistry

upcasters = UpcasterRegistry()

# v1 → v2: add email field
upcasters.register("user.created", 1, 2, lambda d: {**d, "email": d.get("email", "unknown@example.com")})

# v2 → v3: add role field
upcasters.register("user.created", 2, 3, lambda d: {**d, "role": d.get("role", "member")})

Unlike Go, a Python register(name, from_version, to_version, fn) call can jump more than one version — to_version need not equal from_version + 1.

Registering Events

Register event definitions when setting up your serve handler or worker:

TypeScript

import { defineEvent, createEventDefinitionRegistry } from "@ironflow/core";
import { serve, createFunction } from "@ironflow/node";

const userCreatedV1 = defineEvent({ name: "user.created", version: 1 });
const userCreatedV2 = defineEvent({
  name: "user.created",
  version: 2,
  upcast: (data) => ({ ...data, email: data.email ?? "unknown@example.com" }),
});

const registry = createEventDefinitionRegistry();
registry.register(userCreatedV1);
registry.register(userCreatedV2);

const handleUser = createFunction(
  {
    id: "handle-user",
    triggers: [{ event: "user.created" }],
  },
  async ({ event }) => {
    // event.data is always v2 shape — upcasted automatically
    console.log(event.data.email); // always present
  },
);

export default serve({
  functions: [handleUser],
  eventDefinitions: registry,
});

Go

upcasters := ironflow.NewUpcasterRegistry()
upcasters.Register("user.created", 1, 2, func(data json.RawMessage) (json.RawMessage, error) {
    var m map[string]any
    if err := json.Unmarshal(data, &m); err != nil {
        return nil, err
    }
    m["email"] = "unknown@example.com"
    return json.Marshal(m)
})

// Pass upcasters to Serve or NewWorker
handler := ironflow.Serve(ironflow.ServeConfig{
    Functions: []ironflow.Function{HandleUser},
    Upcasters: upcasters,
})

Python

from ironflow import UpcasterRegistry
from ironflow.worker import Worker

upcasters = UpcasterRegistry()
upcasters.register("user.created", 1, 2, lambda d: {**d, "email": d.get("email", "unknown@example.com")})

worker = Worker(functions=[handle_user], upcasters=upcasters)
worker.run()

Worker and StreamingWorker both take upcasters. Projection runners do not apply upcasters — only function handlers do.

Emitting Versioned Events

Specify the version when emitting events:

TypeScript

// Emit a v2 event
await client.emit("user.created",
  { name: "Alice", email: "alice@example.com" },
  { version: 2 },
);

Go

_, err := client.Emit(ctx, "user.created", map[string]any{
    "name":  "Alice",
    "email": "alice@example.com",
}, ironflow.WithEmitVersion(2))

Python

POST /ironflow.v1.IronflowService/Emit — available through rpc.events.emit — takes an optional version since #1955. Omit it for 1:

from protobuf.wkt import Struct
from ironflow.rpc import v1
from ironflow import IronflowRPC
with IronflowRPC(server_url=client.server_url, api_key=client.api_key) as rpc:
    rpc.events.emit(v1.TriggerRequest(event='user.created', data=Struct.from_python({'name': 'Alice', 'email': 'alice@example.com'}), version=2))

Python also versions entity-stream events, which take version in the body — see Versioning with Entity Streams below — and it has the full schema registry surface through rpc.event_schemas.register, get, list, delete, list_history, get_revision, test_upcast and check_enforcement.

Events emitted without a version default to version 1.

Versioning with Entity Streams

Entity stream events also support versioning:

// Append a versioned event to an entity stream
await client.streams.append("user-123", {
  name: "user.profile_updated",
  data: { name: "Alice", email: "alice@example.com" },
  entityType: "user",
}, {
  expectedVersion: 2,
  version: 2, // event schema version
});

from protobuf.wkt import Struct
from ironflow import IronflowRPC
from ironflow.rpc.v1 import AppendEventRequest

rpc = IronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...")

# Append a versioned event to an entity stream
rpc.streams.append_event(AppendEventRequest(
    entity_id="user-123", entity_type="user", event_name="user.profile_updated",
    data=Struct.from_python({"name": "Alice", "email": "alice@example.com"}),
    expected_version=2,
    version=2,  # event schema version
))

When reading entity streams, events are returned with their original version. Upcasting is applied automatically inside push handlers (serve) and pull workers (createWorker) when an event triggers a function — not on the raw read path. Calling client.streams.read(...) returns event data as it was stored; if you need the latest schema there, run it through your registered upcasters yourself.

Best Practices

Always add, never remove

Add new fields with defaults. Never remove or rename fields in a new version — consumers on older code versions may still expect them.

// Good: add field with default
const v2 = defineEvent({
  name: "order.created",
  version: 2,
  upcast: (data) => ({ ...data, currency: data.currency ?? "USD" }),
});

// Bad: remove field (breaks older consumers)
// upcast: (data) => { delete data.oldField; return data; }

# Good: add field with default
upcasters.register("order.created", 1, 2, lambda d: {**d, "currency": d.get("currency", "USD")})

# Bad: remove field (breaks older consumers)
# upcasters.register("order.created", 1, 2, lambda d: {k: v for k, v in d.items() if k != "old_field"})

One change per version

Keep each version bump focused on a single schema change. This makes the upcaster chain easier to understand and debug.

Version numbers are linear

Use simple incrementing integers (1, 2, 3). The upcaster chain runs sequentially from the stored version to the latest registered version.

Test your upcasters

// Verify the full chain works
const v1Data = { name: "Alice" };
// After v1→v2: { name: "Alice", email: "unknown@example.com" }
// After v2→v3: { name: "Alice", email: "unknown@example.com", role: "member" }

# Verify the full chain works
assert upcasters.upcast_to_latest("user.created", {"name": "Alice"}, 1) == {
    "name": "Alice", "email": "unknown@example.com", "role": "member",
}

Event Schema Registry

Ironflow provides a server-side schema registry for storing JSON Schema definitions per event type and version. This enables schema validation, documentation, and discovery.

Validation is off until you turn it on

Registering a schema does not by itself validate anything. Enforcement is controlled by the IRONFLOW_EVENT_SCHEMA_ENFORCEMENT env var, read once at startup: off (the default — registered schemas are stored and never checked), warn (log the mismatch, let the event through) or reject (fail the emit). Engine-generated events such as cron ticks are exempt in every mode. See Event Schema Enforcement.

Registering Schemas

# Register a schema for order.created v1
curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/RegisterSchema \
  -H "Content-Type: application/json" \
  -d '{
    "event_name": "order.created",
    "version": 1,
    "schema_json": "{\"type\":\"object\",\"properties\":{\"orderId\":{\"type\":\"string\"},\"total\":{\"type\":\"number\"}},\"required\":[\"orderId\",\"total\"]}",
    "description": "Order creation event"
  }'

You can also manage schemas with the CLI: ironflow event schema register|list|get|delete|check|history. The ListSchemaHistory and GetSchemaRevision RPCs (ironflow event schema history) return archived definitions captured before a replacement or deletion. History is per event name and version. The SDKs wrap both RPCs: Schemas().ListHistory and GetRevision in Go, schemas.listHistory and getRevision in Node and the browser, and rpc.event_schemas.list_history and get_revision in Python.

Querying Schemas

# List schemas
curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/ListSchemas \
  -H "Content-Type: application/json" -d '{}'

# Filter by event name
curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/ListSchemas \
  -H "Content-Type: application/json" -d '{"eventName":"order.created"}'

# Read the latest version
curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/GetSchema \
  -H "Content-Type: application/json" -d '{"eventName":"order.created"}'

# Read a specific version
curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/GetSchema \
  -H "Content-Type: application/json" -d '{"eventName":"order.created","version":1}'

Deleting a Schema

Both the event name and the version are required:

curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/DeleteSchema \
  -H "Content-Type: application/json" -d '{"eventName":"order.created","version":1}'

Testing Upcasts

Test the upcaster chain for an event without processing real data:

curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/TestUpcast \
  -H "Content-Type: application/json" \
  -d '{
    "event_name": "order.created",
    "from_version": 1,
    "to_version": 3,
    "data": {"orderId": "123"}
  }'

This returns the version chain (stepsApplied) and the input data unchanged. The server does not transform the data: upcasters are SDK code, and they run in your workers when an event is read. To test the transformation itself, call your upcaster in a unit test.

The span is capped at 200 versions: each step in the chain costs one schema lookup, so to_version - from_version greater than 200 is refused with 400.

Checking what enforcement covers

CheckEnforcement reports the server’s current enforcement mode alongside, per registered schema, whether enforcement is actually active for it — so you can see what switching the mode on would and would not catch before you switch it:

# Every registered schema in the environment
curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/CheckEnforcement \
  -H "Content-Type: application/json" -d '{}'

# One event name, over a 72-hour traffic sample
curl -X POST http://localhost:9123/ironflow.v1.EventSchemaService/CheckEnforcement \
  -H "Content-Type: application/json" \
  -d '{"eventName":"order.created","trafficWindowHours":72}'

trafficWindowHours defaults to 24. The response carries the total count of matching schemas separately from the returned list, so a report truncated by the per-request cap is visible rather than silent — narrow with eventName when it is.

Domain-Driven Design

Event versioning is a critical part of maintaining domain events in a DDD system — as your domain model evolves, upcasters ensure consumers always see the latest event schema. See Commands, Events & Reactions for how domain events fit into the broader DDD picture.