Ego

eGo is a protobuf-first framework for building event-sourced and durable-state CQRS applications in Go. It runs on Go-Akt and adds persistence, projections, publishers, sagas, encryption, and observability to an actor system that your application owns.

eGo deliberately does not hide the actor runtime. Your application creates and operates the Go-Akt actor system, including clustering, discovery, remoting, TLS, supervision, and non-eGo actors. eGo contributes the extensions and actor kinds needed for its persistence model.

#Table of contents

• Features
• Requirements
• Installation
• Quick start
• Modeling entities
  • Event-sourced vs durable-state
• Configuration
• Snapshots and retention
• Event batching
• Performance tuning
• Projections
• Publishers
• Sagas and process managers
• Clustering
• Persistence
• Encryption and schema evolution
• Observability
• Reliability and operations
• Testing
• Examples
• Upgrading
• Contributing

#Features

• Event-sourced entities with deterministic recovery
• Durable-state entities that persist only the latest state
• Named, independently configured CQRS projections
• Snapshots, retention policies, and event batching
• Event adapters for protobuf schema evolution
• Event and state publishers for Kafka, NATS, Pulsar, and WebSocket
• Saga/process-manager support with compensation
• OpenTelemetry traces and metrics
• AES-256-GCM event and snapshot encryption
• Entity passivation, placement, relocation, and supervision controls
• In-memory stores, behavior scenarios, and generated mocks for testing

#Requirements

• Go 1.26 or later
• Basic familiarity with Go-Akt
• Protobuf messages for commands, events, and state

For production use, provide durable implementations of the stores your application needs. The stores in testkit are intended for tests and local examples.

#Installation

go get github.com/tochemey/ego/v4

#Quick start

Build one ego.Config, use it to construct the Go-Akt actor system, start that system, and then plug in the eGo engine:

package main

import (
    "context"
    "log"
    "time"

    accountpb "example.com/myapp/gen/account/v1"
    goakt "github.com/tochemey/goakt/v4/actor"
    "github.com/tochemey/ego/v4"
    "github.com/tochemey/ego/v4/projection"
    "github.com/tochemey/ego/v4/testkit"
)

func main() {
    ctx := context.Background()

    // Use a durable EventsStore in production.
    eventsStore := testkit.NewEventsStore()
    if err := eventsStore.Connect(ctx); err != nil {
        log.Fatal(err)
    }
    defer eventsStore.Disconnect(ctx)

    offsetStore := testkit.NewOffsetStore()
    if err := offsetStore.Connect(ctx); err != nil {
        log.Fatal(err)
    }
    defer offsetStore.Disconnect(ctx)

    cfg := ego.NewConfig(eventsStore,
        ego.WithOffsetStore(offsetStore),
        ego.WithProjection("account-balances", &projection.Options{
            Handler:      NewAccountBalancesProjection(),
            BufferSize:   100,
            PullInterval: 500 * time.Millisecond,
        }),
        ego.WithProjection("account-audit", &projection.Options{
            Handler:      NewAccountAuditProjection(),
            BufferSize:   100,
            PullInterval: time.Second,
        }),
    )

    sys, err := goakt.NewActorSystem("accounts", cfg.GoaktOptions()...)
    if err != nil {
        log.Fatal(err)
    }

    if err := sys.Start(ctx); err != nil {
        log.Fatal(err)
    }
    defer sys.Stop(ctx)

    engine, err := ego.NewEngine(sys, cfg)
    if err != nil {
        log.Fatal(err)
    }

    if err := engine.Start(ctx); err != nil {
        log.Fatal(err)
    }
    defer engine.Stop(ctx)

    if err := engine.StartProjection(ctx, "account-balances"); err != nil {
        log.Fatal(err)
    }
    
    if err := engine.StartProjection(ctx, "account-audit"); err != nil {
        log.Fatal(err)
    }

    account := NewAccountBehavior("account-123")
    if err := engine.Entity(ctx, account); err != nil {
        log.Fatal(err)
    }

    state, revision, err := engine.SendCommand(
        ctx,
        account.ID(),
        &accountpb.OpenAccount{InitialBalance: 1000},
        5*time.Second,
    )
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("state=%v revision=%d", state, revision)
}

NewEngine requires a running actor system built with cfg.GoaktOptions(). The same Config must be passed to both calls so the engine and actor-system extensions stay in sync.

NewAccountBalancesProjection and NewAccountAuditProjection represent application handlers that each implement projection.Handler. Each named projection keeps independent offsets and runtime settings.

The engine does not own the actor system. Stop the engine before stopping the actor system, as the deferred calls above do.

#Modeling entities

All commands, events, and states are protobuf messages.

An event-sourced behavior implements ego.EventSourcedBehavior:

type EventSourcedBehavior interface {
    InitialState() ego.State
    HandleCommand(context.Context, ego.Command, ego.State) ([]ego.Event, error)
    HandleEvent(context.Context, ego.Event, ego.State) (ego.State, error)
}

HandleCommand validates a command and returns zero or more events. eGo persists those events before committing the resulting state. HandleEvent must be deterministic because it is also used during recovery.

A durable-state behavior implements ego.DurableStateBehavior:

type DurableStateBehavior interface {
    InitialState() ego.State
    HandleCommand(context.Context, ego.Command, uint64, ego.State) (newState ego.State, newVersion uint64, err error)
}

Configure a state store and spawn the behavior with DurableStateEntity:

cfg := ego.NewConfig(nil, ego.WithStateStore(stateStore))

// Build and start the actor system and engine as shown above.
if err := engine.DurableStateEntity(ctx, behavior); err != nil {
    return err
}

A durable-state command handler deletes the entity's state by returning egopb.DeletedState as the new state with the next version: the state store keeps a tombstone carrying that version and no state, the deletion is published to the state subscribers under that version, and the entity continues from its initial state at that version, so its versions keep increasing across the deletion. A later recovery finds the tombstone and continues the same way.

Behavior values are Go-Akt dependencies. In addition to the methods above, they provide an ID and binary marshalling methods so they can travel with cluster spawn requests. See the event-sourced, durable-state, and saga examples for complete implementations.

#Event-sourced vs durable-state

#Configuration

Engine-wide options are passed to ego.NewConfig:

• WithStateStore enables durable-state entities.
• WithSnapshotStore enables event-sourced snapshots.
• WithOffsetStore supplies durable projection offsets.
• WithProjection registers a named projection and its handler.
• WithEventAdapters applies schema transformations during recovery and projection consumption.
• WithTelemetry enables OpenTelemetry instrumentation.
• WithEncryptor encrypts persisted event and snapshot payloads.
• WithKeyStore lets EraseEntity delete an entity's encryption key; pass the key store the encryptor was built with.
• WithPublishTimeout bounds each delivery attempt to an event or state publisher (default 30 seconds).
• WithEntityKinds registers behavior types on every cluster node.
• WithLogger configures logging for eGo and the underlying actor system.

Entity-specific options are passed when an entity is spawned:

• WithPassivateAfter
• WithRelocation
• WithSupervisorDirective
• WithPlacement

Event-sourced entities additionally support WithSnapshotInterval, WithRetentionPolicy, WithBatchThreshold, and WithBatchFlushWindow.

API details and defaults are documented on pkg.go.dev.

#Snapshots and retention

Snapshots reduce recovery work by restoring the most recent state and replaying only later events:

cfg := ego.NewConfig(eventsStore,
    ego.WithSnapshotStore(snapshotStore),
)

err := engine.Entity(ctx, behavior,
    ego.WithSnapshotInterval(100),
)

A snapshot interval of 0 disables automatic snapshots. Retention runs only after a snapshot has been successfully written:

err := engine.Entity(ctx, behavior,
    ego.WithSnapshotInterval(100),
    ego.WithRetentionPolicy(ego.RetentionPolicy{
        DeleteEventsOnSnapshot:    true,
        DeleteSnapshotsOnSnapshot: true,
        EventsRetentionCount:      200,
    }),
)

Your EventsStore and SnapshotStore implementations must support the corresponding delete operations.

#Event batching

Batching combines events produced by multiple commands into fewer store writes. It is disabled by default:

err := engine.Entity(ctx, behavior,
    ego.WithBatchThreshold(10),
    ego.WithBatchFlushWindow(5*time.Millisecond),
)

The threshold or flush window, whichever is reached first, triggers the write. If batching is enabled without a flush window, eGo uses a 5 ms default. Benchmark the settings with your command pattern and persistence backend; batching trades additional latency for fewer writes and does not have one ideal threshold.

#Performance tuning

eGo can sustain hundreds of thousands of commands per second on a single node with an in-memory store, and tens of thousands with durable backends like Postgres. This section outlines the recommended approach to maximize throughput and minimize memory cost.

#Enable event batching under concurrent load

Batching amortizes the cost of a single store write across multiple commands. It is most effective when:

• The entity receives commands concurrently (multiple goroutines or upstream services)
• The persistence store has non-trivial write latency (e.g. database round-trip > 100us)

Sequential command streams do not benefit from batching because each command waits for the flush window before the batch is written. For purely sequential workloads, leave batching disabled (the default).

#Choose the right batch threshold

#Minimize allocations for high throughput

eGo's hot path is optimized for low allocation overhead (~22 heap allocations per command round-trip). The dominant allocation cost comes from Protocol Buffers serialization, which is inherent to the persistence model. To keep allocation pressure low:

• Keep command and event protos small. Smaller messages reduce marshal/unmarshal cost.
• Use snapshots. They reduce recovery replay length and the number of events held in the store.
• Avoid large state protos. The state is serialized on every reply; smaller states mean fewer bytes and less GC pressure.

#Scale horizontally with clustering

For workloads beyond what a single node can handle, build a clustered Go-Akt actor system and plug eGo into it as described in Clustering.

#Projections

Each projection has its own name, handler, offsets, and recovery settings. Register projections on the Config, then start them after the engine:

cfg := ego.NewConfig(eventsStore,
    ego.WithOffsetStore(offsetStore),
    ego.WithProjection("account-balances", &projection.Options{
        Handler:      accountBalancesHandler,
        BufferSize:   100,
        PullInterval: 500 * time.Millisecond,
        Recovery:     projection.NewRecovery(
            projection.WithRecoveryPolicy(projection.RetryAndFail),
            projection.WithRetries(5),
            projection.WithRetryDelay(time.Second),
        ),
    }),
    ego.WithProjection("account-audit", &projection.Options{
        Handler:           accountAuditHandler,
        BufferSize:        250,
        PullInterval:      time.Second,
        Recovery:          projection.NewRecovery(
            projection.WithRecoveryPolicy(projection.RetryAndSkip),
            projection.WithRetries(3),
            projection.WithRetryDelay(2*time.Second),
        ),
        DeadLetterHandler: accountAuditDeadLetterHandler,
    }),
)

// Build and start the actor system and engine as shown above.
if err := engine.StartProjection(ctx, "account-balances"); err != nil {
    return err
}
if err := engine.StartProjection(ctx, "account-audit"); err != nil {
    return err
}

The two projections consume the same event journal independently. Each uses its own handler, buffer, pull interval, recovery policy, dead-letter handling, and offsets.

Projection handlers receive events with at-least-once delivery. They must be idempotent and safe for concurrent calls across different shards; events within one shard are delivered sequentially.

Offsets are event timestamps, taken just before the event is written by the node that persists it. The runner reads 100 ms behind the current time: an event reaches the handler only once it is at least that old, so an event that lands in the store after a pull already passed its timestamp, because its write was still in flight or because a peer's clock lags slightly, is still ahead of the committed offset when it becomes visible. A write that takes longer than that window, or skew beyond it, can still cause an event to be skipped, so keep node clocks synchronized. All events of one entity land on the same shard and are delivered in order; events of different entities carry no ordering guarantee.

The engine also supports:

• StopProjection to stop a running projection
• IsProjectionRunning to inspect its runtime state
• RebuildProjection to reset offsets and replay from a timestamp
• ProjectionLag to report lag by shard
• Recovery policies and dead-letter handlers for processing failures

Engine.Stop does not stop projection actors because they belong to the caller-owned Go-Akt actor system. In the normal shutdown sequence, call engine.Stop(ctx) and then sys.Stop(ctx); stopping the actor system terminates all projections. If the actor system must remain running, call engine.StopProjection(ctx, name) for each projection before stopping the engine.

In a cluster, a projection runs as a singleton. Every node must register the same named projections because the hosting node resolves each handler from its local Config.

#Publishers

Call AddEventPublishers or AddStatePublishers after the engine starts and before producing changes:

if err := engine.AddEventPublishers(eventPublisher); err != nil {
    return err
}

eGo includes connector modules for:

• Kafka
• NATS
• Pulsar
• WebSocket

You can also implement ego.EventPublisher or ego.StatePublisher. Publisher payload timestamps are Unix nanoseconds, and each payload includes its source shard.

Delivery to publishers is best-effort. Each Publish call runs under the timeout set with WithPublishTimeout (30 seconds by default) and is retried with backoff on failure. A payload that still cannot be delivered is dropped, as is any payload that arrives while a publisher already has 10,000 payloads waiting. Drops are logged and counted on the ego.publisher.dropped.total metric, labelled by publisher ID. Payloads produced before a publisher is added, or while the process is down, are never published.

When every event must reach its destination, publish from a projection instead: register one with WithProjection whose handler calls the publisher. It reads the journal, so it delivers at least once, in order per shard, resumes after an outage from the offset it recorded, runs as a single instance in a cluster, and keeps one set of offset rows like any other projection. State publishers have no such path, since durable state is not journaled.

#Sagas and process managers

eGo includes first-class saga support for long-running business processes that coordinate multiple entities. You can:

• Start a saga with Engine.Saga(...)
• Inspect it with Engine.SagaStatus(...)
• Model compensation logic for timeouts and failures
• Persist saga state using the same event-sourced foundations

A saga consumes the journal, starting at the moment it first ran, and records how far it has read in the offset store. Configure one with ego.WithOffsetStore(...); without it Engine.Saga returns ego.ErrOffsetStoreRequired. Reading the journal is what lets a saga see the events of every entity it coordinates, whichever cluster node persisted them: events written on the saga's own node reach it immediately, events written by a peer within the poll interval.

Delivery is at-least-once, so SagaBehavior.HandleEvent must be idempotent: an event already handled is handed to the saga again when it restarts before its progress was recorded. By default a saga that completes or fails leaves its offset rows in the offset store. Start the saga with ego.WithOffsetRemoval(), as in engine.Saga(ctx, behavior, timeout, ego.WithOffsetRemoval()), to have eGo delete them once the saga completes or fails: a settled saga never reads the journal again, so nothing consumes those rows.

Compensation commands must be idempotent as well: a saga restarted while it compensates sends again every compensation its journal does not record as applied, so a participant can receive one twice.

A saga is fed through the same runner as a projection, so the same read lag and ordering apply: the events of one entity reach the saga in the order they were persisted, but events of different entities may arrive in another order than they happened. A saga coordinating several entities has to tolerate a step arriving before the one it logically follows, for instance by tracking in its own state which steps it still expects.

See the fund-transfer saga example for a complete implementation.

#Clustering

Cluster, discovery, remoting, and TLS are configured with Go-Akt. Register eGo's actor kinds in the cluster configuration:

clusterConfig := goakt.NewClusterConfig().
    WithDiscovery(discoveryProvider).
    WithDiscoveryPort(gossipPort).
    WithPeersPort(peersPort).
    WithPartitionCount(partitions).
    WithMinimumPeersQuorum(quorum).
    WithReplicaCount(replicas).
    WithKinds(ego.ClusterKinds()...) // eGo's actor kinds, required for relocation

Also register every event-sourced, durable-state, and saga behavior type on every node:

cfg := ego.NewConfig(eventsStore,
    ego.WithEntityKinds(
        new(AccountBehavior),
        new(OrderBehavior),
        new(CheckoutSaga),
    ),
)

Then compose cfg.GoaktOptions() with Go-Akt's cluster, remote, TLS, and application-specific options when constructing the actor system:

sys, err := goakt.NewActorSystem("accounts",
    append(
        cfg.GoaktOptions(),
        goakt.WithCluster(clusterConfig),
        goakt.WithRemote(remote.NewConfig(host, remotingPort)),
        goakt.WithTLS(&tlsInfo),
    )...,
)

You retain full control over discovery, partitioning, quorum, replicas, TLS, remoting, and any additional cluster knobs Go-Akt exposes. eGo derives cluster behavior (e.g. running projections as singletons) directly from sys.InCluster() at runtime — no separate cluster flag to keep in sync.

Single-node deployments do not need ClusterKinds or WithEntityKinds.

#Remoting

Go-Akt speaks a multiplexed remoting protocol — per-peer lane connections, chunked large messages, and credit-based flow control. eGo requires no configuration for it: remote.NewConfig(host, remotingPort) negotiates it on its own, and eGo's remote surface is unchanged.

Two things follow from how eGo's traffic maps onto those lanes.

Entity placement travels on the control lane. Spawns, singleton placement, and death-watch are carried separately from user commands, so a burst of entity traffic can no longer delay them.

Entity commands share one ordinary lane by default. SendCommand and saga participant calls are asks, and every ask from one node to a given peer rides a single connection with one writer queue and one credit window. Raising the lane count shards receivers across connections so sends proceed in parallel:

goakt.WithRemote(remote.NewConfig(host, remotingPort,
    remote.WithOrdinaryLanes(4),
)),

Any lane count is safe for eGo. Ordering in eGo is per entity — each entity actor serializes its own mailbox — and Go-Akt pins a receiver to a lane by a stable hash of its address, so commands to one entity stay in order however many lanes exist. eGo never relies on ordering between different entities.

Slow entities degrade gracefully rather than stalling a connection: asks are multiplexed by correlation ID and dispatched on a bounded worker pool, so an entity waiting on its events store occupies a worker instead of blocking the socket. When that pool saturates, the affected request comes back as an unavailable error and the connection stays healthy.

Leave the protocol pin at its auto default while upgrading a running cluster. Nodes negotiate the multiplexed protocol with peers that support it and fall back to the legacy wire for those that do not, so a cluster rolls node by node without a flag day.

The Kubernetes cluster example demonstrates a three-node deployment with PostgreSQL, Kubernetes discovery, a singleton projection, OpenTelemetry, Prometheus, Jaeger, and Grafana.

#Persistence

eGo defines small interfaces for:

• persistence.EventsStore
• persistence.SnapshotStore
• persistence.StateStore
• offsetstore.OffsetStore

Applications may implement these interfaces directly. The ego-contrib project provides ready-to-use implementations:

• Event stores: memory, Postgres, SQLite
• Offset stores: memory, Postgres, SQLite
• Snapshot stores: Postgres, SQLite
• Durable state stores: memory, Postgres, SQLite, Cassandra, DynamoDB

To use a contrib store, import the relevant module alongside eGo:

import (
    "github.com/tochemey/ego-contrib/eventstore/postgres"
    "github.com/tochemey/ego-contrib/snapshotstore/postgres"
    "github.com/tochemey/ego-contrib/offsetstore/postgres"
)

Applications own store connectivity: connect stores before starting the actor system and disconnect them after the engine and actor system have stopped.

#Encryption and schema evolution

WithEncryptor transparently encrypts event and snapshot payloads before persistence and decrypts them during entity recovery and projection processing. The built-in encryption.AESEncryptor uses AES-256-GCM and a pluggable encryption.KeyStore.

WithEventAdapters registers transformations for events written with older protobuf schemas. Adapters run in registration order during recovery and projection consumption.

#Observability

WithTelemetry accepts an OpenTelemetry tracer and meter:

cfg := ego.NewConfig(eventsStore,
    ego.WithTelemetry(&ego.Telemetry{
        Tracer: tracer,
        Meter:  meter,
    }),
)

Instrumentation covers command dispatch and handling, event persistence, active entities and projections, projection processing, offsets, lag, and approximate events behind.

#Reliability and operations

eGo includes several production-focused capabilities:

• Faster recovery through snapshots
• Storage cleanup through retention policies
• At-rest encryption for events and snapshots
• GDPR-style erasure with Engine.EraseEntity(...): the live entity is stopped, then with full=false its encryption key is deleted through the store configured with WithKeyStore (crypto-shredding), and with full=true its events and snapshots are deleted as well, and the durable state from its store when one is configured with WithStateStore.
• Store write failures are handed to the entity's supervisor: the default RestartDirective replays the journal on the same PID, StopDirective stops the entity
• Pluggable structured logging via ego.WithLogger(...)

#Testing

The testkit package provides in-memory event, snapshot, state, offset, and key stores. Its scenario API tests behavior logic without starting an actor system:

testkit.ForEventSourcedBehavior(behavior).
    Given(priorState).
    When(command).
    ThenEvents(t, expectedEvents...).
    ThenState(t, expectedState)

Given states the entity's prior state — the same state the engine recovers from the journal and hands to HandleCommand — so the arrangement never depends on HandleEvent. Omit it to start from InitialState().

Where the history reads better than the folded state, GivenEvents arranges the entity from the events it has already recorded, replayed in order through HandleEvent just as the engine replays a journal:

testkit.ForEventSourcedBehavior(behavior).
    GivenEvents(accountCreated, accountCredited).
    When(command).
    ThenEvents(t, expectedEvents...)

The two compose — Given(snapshotState).GivenEvents(subsequentEvents...) mirrors an entity recovered from a snapshot and then replayed. An event HandleEvent rejects fails the scenario as a broken arrangement, reported as such by every assertion including ThenError, so a bad setup can never pass as a failed command.

The durable-state scenario reads the same way, with Given(priorState, priorVersion) and ThenState/ThenVersion.

Generated mocks for persistence, encryption, adapters, offsets, and publishers are available under mocks.

#Examples

• Event-sourced entity
• Durable-state entity
• Fund-transfer saga
• Three-node Kubernetes cluster with a projection and a fund-transfer saga
• Benchmarks

Run the local examples with:

make run-eventsourced
make run-durablestate
make run-saga

#Upgrading

See the changelog for breaking changes and version-specific migration instructions.

#Contributing

Contributions are welcome. Read the contribution guide before opening a pull request.