Entity Streams

An Entity Stream is an ordered sequence of events scoped to a specific ID (e.g., user_456). Ironflow uses these streams to maintain a “source of truth” for every change in your system.

Appending Events

Appending an event is the primary way to update an entity. The stream is automatically created on the first append.

TypeScript

import { createClient } from "@ironflow/node";

const ironflow = createClient({ apiKey: "..." });

const { entityVersion, eventId } = await ironflow.streams.append("order-123", {
  entityType: "order",
  name: "order.placed",
  data: { total: 100 },
});

Go

res, err := client.AppendStreamEvent(ctx, "order-123", ironflow.AppendEventInput{
    EntityType: "order",
    Name:       "order.placed",
    Data:       map[string]any{"total": 100},
})

Python

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

client = IronflowRPC(server_url="http://localhost:9123", api_key="ifkey_...")
res = client.streams.append_event(AppendEventRequest(
    entity_id="order-123", entity_type="order", event_name="order.placed",
    data=Struct.from_python({"total": 100}), expected_version=-1,
))
print(res.entity_version, res.event_id)

See the Python SDK reference for the generated Connect request and response types.

An entity ID is at most 128 characters from A-Z a-z 0-9 - . _ : ~, with no leading, trailing or doubled .. An entity type is at most 64 characters from A-Z a-z 0-9 - _ : ~, so it cannot hold a .. Either one fails with invalid_argument when it breaks the rule. An ID that starts with ironflow: (in any case) is reserved and fails with permission_denied. See Reserved names.

---

Optimistic Concurrency

To prevent “lost updates” when multiple workers try to update the same entity, use the expectedVersion option.

Value	Behavior
0	Create Only: Succeeds only if the stream doesn’t exist yet.
N	Exact Version: Succeeds only if the current version is exactly N.
-1	No Guard: Append regardless of version (default in the Node and Go SDKs).

The generated Python request leaves an unset expected_version at 0, which means create only. Pass expected_version=-1 to skip the guard.

import { ContendedError } from "@ironflow/node";

// Fetch current version (returns null if the stream has no events yet)
const info = await ironflow.streams.getInfo("order-123");
const version = info?.version ?? 0;

// Try to append with a guard
try {
  await ironflow.streams.append("order-123", { ... }, {
    expectedVersion: version
  });
} catch (err) {
  // If another write happened first, the server answers with ConnectRPC code
  // `aborted` and writes nothing. The SDK throws ContendedError for it.
  if (err instanceof ContendedError) {
    // re-read, rebuild, retry
  } else {
    throw err;
  }
}

// Fetch current version. A stream with no events yet returns a "stream not found"
// error: use version 0 for it instead of returning.
info, err := client.GetStreamInfo(ctx, "order-123")
if err != nil {
    return err
}

// Try to append with a guard
_, err = client.AppendStreamEvent(ctx, "order-123", ironflow.AppendEventInput{
    EntityType: "order",
    Name:       "order.updated",
    Data:       map[string]any{"status": "shipped"},
}, ironflow.WithExpectedVersion(int64(info.Version)))
if errors.Is(err, ironflow.ErrContended) {
    // another write happened first and nothing was written: re-read, rebuild, retry
} else if err != nil {
    return err
}

from protobuf.wkt import Struct
from ironflow.rpc import IronflowRPCError
from ironflow.rpc.v1 import AppendEventRequest, GetStreamInfoRequest

# Fetch current version. A stream with no events yet raises "not_found": use 0.
try:
    version = client.streams.get_info(GetStreamInfoRequest(entity_id="order-123")).version
except IronflowRPCError as err:
    if err.code != "not_found":
        raise
    version = 0

# Try to append with a guard
try:
    client.streams.append_event(AppendEventRequest(
        entity_id="order-123", entity_type="order", event_name="order.updated",
        data=Struct.from_python({"status": "shipped"}), expected_version=version,
    ))
except IronflowRPCError as err:
    if err.code == "aborted":  # another write happened first: "version conflict"
        ...  # re-read, rebuild, retry
    else:
        raise

---

Deleting a Stream

Deleting a stream appends a $stream.deleted tombstone at the stream’s current version + 1, under the same optimistic-concurrency check as any other append, and drops the stream’s snapshots. After that the stream refuses every append. It cannot be undone. The verb requires streams:delete, which only the admin role holds by default.

A legacy stream that holds more than one entity_type cannot be deleted this way: appending the tombstone is rejected for the same reason any other append to that stream is.

TypeScript

const { entityVersion } = await ironflow.streams.delete("order-123");
// entityVersion is the tombstone's version — the stream's final version

// Also delete the events below the tombstone
await ironflow.streams.delete("order-123", { purge: true });

Go

version, err := client.DeleteStream(ctx, "order-123", false)

// purge=true also deletes the events below the tombstone
version, err = client.DeleteStream(ctx, "order-123", true)

Python

from ironflow.rpc.v1 import DeleteStreamRequest

res = client.streams.delete(DeleteStreamRequest(entity_id="order-123", purge=False))
print(res.entity_version)

CLI

ironflow stream delete order-123          # prompts first
ironflow stream delete order-123 --purge --yes

What the tombstone looks like

The tombstone is an ordinary event on the stream, named $stream.deleted, with this payload:

{
  "entityId": "order-123",
  "deletedAt": "2026-09-21T10:15:30.123456789Z",
  "purge": false
}

Because it is a real event, it occupies the next version and shows up in read. Any later append returns ConnectRPC code failed_precondition — “entity stream is deleted”. createSnapshot returns the same code, so a snapshot cannot put state back onto a stream whose history was purged.

The tombstone’s emitted_by names who deleted the stream, for example apikey:<id> or user:<id>.

You cannot write the tombstone yourself. $ is a reserved event-name prefix: an appendEvent whose name starts with it is rejected with invalid_argument, whatever the name. Deleting a stream is streams:delete, which only admin holds by default, and reserving the prefix is what keeps an ordinary entities:append from reaching the same outcome by another door. All reserved names are listed in Reserved names.

purge

purge: true additionally deletes every event row below the tombstone. The tombstone row itself stays — it is what makes the stream refuse appends, so it can never be purged. The purge runs after the tombstone has committed, not in the same transaction; if it fails, the error says the stream is tombstoned and re-issuing the delete with purge: true runs the cleanup again.

Purging is the stronger choice in one more way: rows elsewhere that reference those events (a run, an outbox row, a pending debounce fire) are removed with them by foreign-key cascade. That is what you asked for by passing purge. The scheduled cleanup below makes the opposite choice — it skips any event another row still references.

Three limits on how far that cascade is allowed to reach:

• Active runs block the purge. If any run triggered by an event below the tombstone has not reached completed, failed or cancelled, the purge is refused with failed_precondition and nothing is deleted. The cascade does not check status, so without this an executing run would lose its row mid-flight. The tombstone is already in place by then, so nothing new can arrive — wait for the runs to finish and issue the purge again.
• On PostgreSQL, a write in progress can block the purge. The purge does not wait on those runs or their rows. It is refused with aborted while another call holds one of them or a row of one, for example a step report in progress, a run invoked from one of them, or the lease a cancelled run still holds. Nothing is deleted. The write ends on its own, so issue the purge again. The purge can still wait on the stream’s event rows, for example for a run redaction to commit, before it answers.
• Dead letters go with their events. outbox_dead_letter rows have no foreign key, so they would otherwise outlive the events they name, and every later ironflow outbox dlq requeue against them would fail: requeueing re-inserts into outbox, whose event_id does cascade. The purge deletes them in the same transaction instead.

Deleting a stream that is already deleted returns failed_precondition when purge is false. With purge: true it re-runs the purge and returns the existing tombstone’s version.

What still works

Reading does. read and getInfo keep answering after a delete, so the tombstone and whatever history survived stay visible. listStreams hides the stream.

Two fields mark a deleted stream, and both come from the ConnectRPC surface: the GetStreamInfo response carries deleted, true once the stream is tombstoned, and ListStreams takes includeDeleted to show tombstoned streams again. The Node SDK, the Go SDK and the CLI expose neither in this release; the generated Python SDK, which mirrors the protobuf messages, has both. The dashboard reads deleted and shows a red Deleted pill on the stream detail page, but its stream list sends no includeDeleted, so a tombstoned stream cannot be listed there either. Through the Node or Go SDK today, tell a deleted stream apart by reading it and looking for $stream.deleted as its last event.

A delete also publishes system.stream.{entity_id}.deleted, carrying entity_id, entity_type, entity_version, purge and timestamp, and writes an entity_stream.deleted audit row.

Projections are not rewritten

A projection sees the tombstone only if its event list matches it

Projection delivery is filtered by event name. A projection whose event list is ["order.*"] never receives $stream.deleted — a * expands to a subject wildcard under order., which the tombstone’s name does not sit under. (A catch-all ["*"] does match it.) Ironflow does not reach into projection state either way, so a projection that never sees the tombstone keeps everything its read model built from the deleted stream. To react to a delete, add $stream.deleted to the projection’s event list and handle it in the reducer.

Scheduled cleanup

Deleting a stream without purge leaves its events in place. Set IRONFLOW_ENTITY_EVENT_RETENTION_DAYS to have the nightly job remove them later: it only touches events below a tombstone, on streams that carry one, older than the window, and never one that a run, outbox row, pending debounce fire or outbox dead letter still references. A live stream is never pruned. See configuration. The window is server-wide, not per environment.

---

Redacting an Event’s Payload

Deleting a stream removes events. Redacting one keeps the event and removes only its payload, replacing data with a placeholder. The row keeps its id, name, sequence and entity version, so replay ordering and optimistic concurrency are unaffected — the stream still has the same number of events at the same versions, and a reducer still receives one at that position. It cannot be undone. The verb requires events:redact, which only the admin role holds by default.

TypeScript

await ironflow.redactEvent("ev_01JD2KQ7X8Y9Z");

Go

err := client.RedactEvent(ctx, "ev_01JD2KQ7X8Y9Z")

Python

from ironflow.rpc.v1 import RedactEventRequest

client.events.redact(RedactEventRequest(event_id="ev_01JD2KQ7X8Y9Z"))

CLI

ironflow event redact ev_01JD2KQ7X8Y9Z          # prompts first
ironflow event redact ev_01JD2KQ7X8Y9Z --yes

What the reducer sees

{
  "$redacted": true,
  "sha256": "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08",
  "redactedAt": "2026-09-21T10:15:30.123456789Z"
}

sha256 is of the original bytes; nothing recovers them from it. Reducers and upcasters receive this in the payload’s place, so a reducer that assumes its own shape will read undefined fields rather than crash — but it will also silently fold a missing value into state. Check first:

import { isRedacted } from "@ironflow/node";

function reduce(state, event) {
  if (isRedacted(event.data)) return state; // keep what we had
  return { ...state, email: event.data.email };
}

The Go SDK exposes ironflow.IsRedacted(data) and the predicate is the same everywhere: the payload is an object carrying "$redacted": true.

Snapshots are dropped

A snapshot holds state derived from the bytes being removed, so leaving it in place would keep getSnapshot serving the original content and a reducer resuming from a snapshot would never see the placeholder. Redacting an event on an entity stream therefore deletes every snapshot for that stream in the same transaction, and sweeps their object-storage prefix. The next read re-reduces from the events. Take a fresh snapshot afterwards if the stream is long enough to need one.

What redaction does not reach

• Projection state. A projection’s read model is yours and Ironflow never rewrites it — on a stream delete or on a redaction. A projection that folded the removed payload into its state keeps it until you rebuild the projection or handle the removal in your own reducer.
• The runs the event triggered. A run’s input holds a copy of the payload, and so does its run.created audit row. Redact those with ironflow run redact (or client.redactRun). audit_events carries no event id, so redacting an event scrubs no audit rows at all.
• A copy already published to a stream, which survives until its own retention window expires — seven days on PUBSUB; EVENTS has no age expiry and requires terminal reclamation. The worker reads each row just before it publishes. A publish past that read when the redaction commits counts as published. Any other unpublished event publishes the placeholder.
• An object-storage copy whose cleanup fails. Snapshot blobs are swept on a best-effort basis after the rows are scrubbed; a failed sweep is logged and the object is left behind with nothing to reclaim it.

Redaction is idempotent: a second call succeeds and leaves the first sha256 alone. It emits an event.redacted audit row carrying the event id only, and publishes no system.* frame. See ADR 0092.

---

Reading Stream History

You can read the history of an entity to reconstruct its state. A read returns every event when you pass no limit. To read a long stream in pages, pass limit and a start version.

TypeScript

// Forward (oldest first)
const { events } = await ironflow.streams.read("order-123", {
  direction: "forward",
  limit: 50,
});

// Backward (latest first)
const { events: latest } = await ironflow.streams.read("order-123", {
  direction: "backward",
  limit: 1,
});

Go

events, err := client.ReadStream(ctx, "order-123", ironflow.ReadStreamOpts{
    Direction: "forward",
})

Python

# Forward (oldest first)
page = client.streams.read_stream(ReadStreamRequest(entity_id="order-123", direction="forward", limit=50))
for event in page.events:
    print(event.name, event.data.to_python() if event.data else None)

# Backward (latest first)
latest = client.streams.read_stream(ReadStreamRequest(entity_id="order-123", direction="backward", limit=1))

# Current version, for the optimistic-concurrency guard above
info = client.streams.get_info(GetStreamInfoRequest(entity_id="order-123"))
print(info.version, info.event_count)

---

Real-Time Subscriptions

You can subscribe to an entity stream to receive live updates as new events are appended.

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

const sub = await ironflow.streams.subscribe("order-123", {
  entityType: "order",
  onEvent: (e) => console.log(`New event: ${e.name}`),
  replay: 10, // Replay last 10 events before going live
});

from ironflow.rpc.v1 import SubscribeOptions, SubscribeRequest

# Entity events use the pattern entity:{type}.{id}.{event_name}
for e in client.pubsub.subscribe(SubscribeRequest(
    pattern="entity:order.order-123.>",
    options=SubscribeOptions(replay=10),  # Replay last 10 events before going live
)):
    print(f"New event: {e.topic}")

Projections

If you find yourself reading streams frequently to calculate state (e.g., “Total Balance”), use Projections instead.

Projections subscribe to the global event log across many streams and are a different fold from aggregate rehydration — entity-stream snapshots accelerate aggregate load, but they do not accelerate projection rebuild. See Projections read the global event log for why.

Domain-Driven Design

Entity streams implement the Aggregate pattern from DDD — each stream is a consistency boundary with its own identity and event history. See Aggregates & Entity Streams for the full DDD mapping.