mink

package module
v1.1.15 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 19, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

README

go-mink 🦫

The batteries-included Event Sourcing & CQRS toolkit for Go.

Go Reference Go Report Card Build Status codecov License Go Version

Quality Gate Status Reliability Rating Security Rating Maintainability Rating

📖 Documentation · 🎓 Tutorial · 💡 Examples · 🔎 API Reference · 🗺️ Roadmap

go-mink lets you build event-sourced systems in Go without the usual boilerplate. Instead of storing current state, it stores every change as an immutable event and rebuilds state by replaying them — giving you a perfect audit log, time travel, and read models for free. It's inspired by MartenDB for .NET and ships with everything you need to go from prototype to production: a pluggable event store, a CQRS command bus, projections, sagas, an outbox, field-level encryption, and a full GDPR toolkit.

// Model your domain as events — go-mink handles the rest.
order := NewOrder("order-123")
order.Create("customer-456")
order.AddItem("SKU-001", 2, 29.99)

store.SaveAggregate(ctx, order)              // append events, optimistic concurrency
loaded := NewOrder("order-123")
store.LoadAggregate(ctx, loaded)             // rebuild state by replaying events

Why "go-mink"? Marten (the animal) inspired the .NET library's name, so we picked the mink — another member of the Mustelidae family — for its Go cousin. 🦫

Table of Contents

Why Event Sourcing?

Traditional CRUD apps overwrite data — the moment you UPDATE a row, the previous state is gone. Event sourcing keeps the entire history as an append-only log of facts. That single change unlocks a lot:

  • 🕰️ Complete audit trail — every state change is a durable, timestamped fact. Answer "who changed what, when, and why?" without bolt-on logging.
  • ⏪ Time travel & debugging — replay a stream to any point in time to see exactly how a bug happened.
  • 📊 Purpose-built read models — project the same events into as many optimized query models as you need (CQRS), and rebuild them from scratch whenever your requirements change.
  • 🔌 Natural integration — events are a first-class integration point. Publish them reliably to Kafka, webhooks, or SNS with the built-in outbox.
  • 🔐 Compliance by design — with field-level encryption, crypto-shredding, and a GDPR erasure/export toolkit, "right to be forgotten" is a feature, not a project.

The catch is usually the plumbing: optimistic concurrency, serialization, schema evolution, projection checkpoints, idempotency, sagas. go-mink is that plumbing — production-tested, pluggable, and zero-overhead when a feature is unused — so you can focus on your domain.

New to the pattern? Start with the Introduction and the 8-part Blog Series.

Features

Everything below is included, works together, and is covered by tests (CI enforces 90% coverage).

Area What you get
🎯 Event Store Append-only storage, optimistic concurrency, global ordering, snapshots, catch-up subscriptions, filtered feed reads (by type / stream / category)
🔌 Adapters Production PostgreSQL adapter and an in-memory adapter for tests — swap with one line
🧱 Aggregates AggregateBase with event application & uncommitted-event tracking
📋 Command Bus Full CQRS dispatch with a composable middleware pipeline
🧩 Middleware Validation, recovery, logging, correlation/causation IDs, idempotency, timeout, metrics, tracing, audit
📖 Projections Inline (strong), async (eventual), and live (real-time) projections with checkpoints, pause/resume & rebuild
📊 Read Models Generic repository + fluent query builder (filter/order/paginate) over memory or PostgreSQL
🔁 Sagas Process managers for long-running workflows with automatic compensation and timeouts
📤 Outbox Transactional outbox for reliable publishing to Webhook · Kafka · SNS, with retry & dead-letter
🔄 Versioning Schema evolution via upcasting (zero DB migration) + a schema-compatibility registry
📦 Serializers JSON (default), MessagePack, and Protobuf
🔐 Encryption Field-level encryption via AWS KMS · HashiCorp Vault · local AES-256-GCM (envelope encryption, per-tenant keys)
🧹 GDPR Toolkit Data export (Art. 15/20), erasure (Art. 17) via crypto-shredding, retention policies, anonymization, subject discovery
📝 Audit Logging Immutable, queryable audit trail of every command (who/what/when/outcome)
📈 Observability First-class Prometheus metrics and OpenTelemetry tracing
🛠️ CLI mink — project init, code generation, migrations, projection / stream / feed inspection, diagnostics, GDPR ops
🧪 Testing BDD fixtures, event assertions, projection/saga test harnesses, and Docker test containers

Installation

Requires Go 1.25+ (matches go.mod; CI builds on Go 1.25 and 1.26).

go get go-mink.dev                      # core library (memory adapter included)
go get go-mink.dev/adapters/postgres    # PostgreSQL adapter for production

Optionally grab the mink CLI for scaffolding and operations:

go install go-mink.dev/cmd/mink@latest
mink --help

Quick Start

Define events, model an aggregate, and let the store persist and replay them. This runs against the in-memory adapter — no infrastructure required.

package main

import (
	"context"
	"fmt"

	"go-mink.dev"
	"go-mink.dev/adapters/memory"
)

// 1. Events are plain structs — the immutable facts of your domain.
type OrderCreated struct{ OrderID, CustomerID string }
type ItemAdded struct {
	OrderID string
	SKU     string
	Qty     int
}

// 2. An aggregate embeds AggregateBase and applies events to build state.
type Order struct {
	mink.AggregateBase
	CustomerID string
	Items      int
}

func NewOrder(id string) *Order {
	o := &Order{}
	o.SetID(id)
	o.SetType("Order")
	return o
}

func (o *Order) Create(customerID string) {
	o.Apply(OrderCreated{OrderID: o.AggregateID(), CustomerID: customerID})
}

func (o *Order) AddItem(sku string, qty int) {
	o.Apply(ItemAdded{OrderID: o.AggregateID(), SKU: sku, Qty: qty})
}

// ApplyEvent mutates state — called both for new and replayed events.
func (o *Order) ApplyEvent(event interface{}) error {
	switch e := event.(type) {
	case OrderCreated:
		o.CustomerID = e.CustomerID
	case ItemAdded:
		o.Items += e.Qty
	}
	return nil
}

func main() {
	ctx := context.Background()
	store := mink.New(memory.NewAdapter())
	store.RegisterEvents(OrderCreated{}, ItemAdded{})

	// Write: emit events and persist them.
	order := NewOrder("order-123")
	order.Create("customer-456")
	order.AddItem("SKU-001", 2)
	if err := store.SaveAggregate(ctx, order); err != nil {
		panic(err)
	}

	// Read: rebuild state by replaying the stream.
	loaded := NewOrder("order-123")
	if err := store.LoadAggregate(ctx, loaded); err != nil {
		panic(err)
	}
	fmt.Printf("Order for %s has %d items (v%d)\n",
		loaded.CustomerID, loaded.Items, loaded.Version())
	// → Order for customer-456 has 2 items (v2)
}

Going to production? Swap the adapter — everything else stays the same:

adapter, _ := postgres.NewAdapter("postgres://localhost/mydb")
defer adapter.Close()
store := mink.New(adapter)

The PostgreSQL adapter auto-creates its schema (streams + events, plus snapshots/checkpoints) on first use. See the Event Store guide.

How It Fits Together

Commands flow through a middleware pipeline into handlers that load aggregates and emit events. The event store is the source of truth; projections, sagas, and the outbox all react to the same durable log.

flowchart LR
    CMD[Commands] --> BUS
    subgraph BUS[Command Bus]
      direction TB
      MW[validation · idempotency · correlation<br/>audit · metrics · tracing · recovery]
    end
    BUS --> H[Handlers]
    H --> AGG[Aggregates]
    AGG -- events --> ES[(Event Store<br/>optimistic concurrency)]
    ES --> PE[Projection Engine]
    ES --> SM[Saga Manager]
    ES --> OP[Outbox Processor]
    PE --> RM[(Read Models)]
    SM -- compensation --> BUS
    OP --> EXT[Kafka · Webhooks · SNS]
  • Command Bus dispatches commands through middleware to handlers (bus.go, middleware.go).
  • Aggregates run domain logic and produce events (aggregate.go).
  • Event Store appends events with optimistic concurrency and replays them (store.go).
  • Projection Engine builds read models — inline, async, or live (projection_engine.go).
  • Saga Manager orchestrates multi-step workflows and compensates on failure (saga_manager.go).
  • Outbox Processor reliably publishes events to external systems (outbox_processor.go).

Full write-up in the Architecture guide.

Guided Tour

⚙️ CQRS with the Command Bus 
// Commands are validated, deduplicated intents.
type CreateOrder struct {
	mink.CommandBase
	CustomerID string `json:"customerId"`
}

func (c CreateOrder) CommandType() string { return "CreateOrder" }
func (c CreateOrder) Validate() error {
	if c.CustomerID == "" {
		return mink.NewValidationError("CreateOrder", "CustomerID", "required")
	}
	return nil
}

bus := mink.NewCommandBus()
bus.Use(mink.RecoveryMiddleware())        // panics → errors
bus.Use(mink.ValidationMiddleware())      // reject invalid commands
bus.Use(mink.CorrelationIDMiddleware(nil))

// Idempotency: the same command is processed exactly once.
idem := memory.NewIdempotencyStore()
bus.Use(mink.IdempotencyMiddleware(mink.DefaultIdempotencyConfig(idem)))

bus.RegisterFunc("CreateOrder", func(ctx context.Context, cmd mink.Command) (mink.CommandResult, error) {
	c := cmd.(CreateOrder)
	// load aggregate, run domain logic, save events...
	return mink.NewSuccessResult("order-123", 1), nil
})

result, err := bus.Dispatch(ctx, CreateOrder{CustomerID: "cust-456"})

See the cqrs example and the Advanced Patterns guide.

📖 Projections & Read Models 
type OrderSummary struct {
	OrderID   string
	Status    string
	ItemCount int
	Total     float64
}

type OrderSummaryProjection struct {
	mink.ProjectionBase
	repo *mink.InMemoryRepository[OrderSummary]
}

func (p *OrderSummaryProjection) Apply(ctx context.Context, e mink.StoredEvent) error {
	// transform events into read-model updates...
	return nil
}

engine := mink.NewProjectionEngine(store, mink.WithCheckpointStore(memory.NewCheckpointStore()))
engine.RegisterInline(&OrderSummaryProjection{repo: repo})
engine.Start(ctx)
defer engine.Stop(ctx)

// Query with the fluent builder.
orders, _ := repo.Find(ctx, mink.NewQuery().
	Where("Status", mink.FilterOpEq, "Pending").
	OrderByDesc("Total").
	WithLimit(10))

// Rebuild from scratch whenever requirements change.
mink.NewProjectionRebuilder(store, checkpointStore).
	RebuildInline(ctx, projection, mink.RebuildOptions{BatchSize: 1000})

See the projections example and the Read Models guide.

🔁 Sagas / Process Managers

Orchestrate multi-step workflows that span aggregates, with automatic compensation when a step fails.

type OrderFulfillmentSaga struct {
	mink.SagaBase
	PaymentID   string
	ShipmentID  string
}

// React to events by emitting the next command; compensate on failure.
func (s *OrderFulfillmentSaga) HandleEvent(ctx context.Context, e mink.StoredEvent) ([]mink.Command, error) {
	switch e.Type {
	case "OrderPlaced":
		return []mink.Command{ProcessPayment{...}}, nil
	case "PaymentFailed":
		return []mink.Command{CancelOrder{...}}, nil // compensation
	}
	return nil, nil
}

See the sagas example and the Sagas section.

📤 Outbox — reliable publishing to Kafka / Webhooks / SNS 
outboxStore := memory.NewOutboxStore()
es := mink.NewEventStoreWithOutbox(store, outboxStore, []mink.OutboxRoute{
	{EventTypes: []string{"OrderCreated"}, Destination: "webhook:https://partner.example.com/events"},
	{Destination: "kafka:all-events"}, // every event
})

// Events and outbox messages are written atomically.
es.Append(ctx, "Order-123", []interface{}{OrderCreated{OrderID: "123"}})

// A background processor delivers them with retry + dead-letter.
processor := mink.NewOutboxProcessor(outboxStore,
	mink.WithPublisher(webhook.New()),
	mink.WithPollInterval(time.Second),
)
processor.Start(ctx)
defer processor.Stop(ctx)

Publishers: outbox/webhook, outbox/kafka, outbox/sns.

🔄 Event Versioning & Upcasting (zero DB migration) 
// Upcast v1 → v2 by adding a default Currency field.
type orderCreatedV1ToV2 struct{}

func (u orderCreatedV1ToV2) EventType() string { return "OrderCreated" }
func (u orderCreatedV1ToV2) FromVersion() int  { return 1 }
func (u orderCreatedV1ToV2) ToVersion() int    { return 2 }
func (u orderCreatedV1ToV2) Upcast(data []byte, m mink.Metadata) ([]byte, error) {
	var obj map[string]interface{}
	json.Unmarshal(data, &obj)
	obj["currency"] = "USD"
	return json.Marshal(obj)
}

chain := mink.NewUpcasterChain()
chain.Register(orderCreatedV1ToV2{})
store := mink.New(adapter, mink.WithUpcasters(chain))
// Old events are upcasted transparently on Load — the schema version lives in metadata.

See the versioning example and the Versioning guide.

🔐 Field-Level Encryption & Crypto-Shredding 
provider, _ := local.New(local.WithKey("master-1", myKey)) // KMS/Vault in production
defer provider.Close()

encConfig := mink.NewFieldEncryptionConfig(
	mink.WithEncryptionProvider(provider),
	mink.WithDefaultKeyID("master-1"),
	mink.WithEncryptedFields("CustomerCreated", "email", "phone", "ssn"),
	mink.WithTenantKeyResolver(func(tenantID string) string { return "tenant-" + tenantID }),
)

store := mink.New(adapter, mink.WithFieldEncryption(encConfig))
// PII is encrypted at rest and decrypted transparently on load.
// Revoking a key makes that data permanently unrecoverable — GDPR crypto-shredding.
store.EncryptionConfig().RevokeKey("tenant-42")

Providers: encryption/local, encryption/kms, encryption/vault. See the Security guide.

🧹 GDPR — Export, Erasure & Retention

go-mink treats data governance as a first-class concern.

// Right to access / portability (Article 15 & 20)
exporter := mink.NewDataExporter(store)
result, _ := exporter.Export(ctx, mink.ExportRequest{
	SubjectID: "user-123",
	Filter:    mink.FilterByUserID("user-123"),
})

// Right to erasure (Article 17) — one call: revoke keys, redact read models,
// run external-PII hooks, and emit a verifiable erasure certificate.
eraser := mink.NewDataEraser(store,
	mink.WithEraseSubjectResolver(resolver),
	mink.WithSharedKeyGuard(),                 // don't nuke a shared tenant key
	mink.WithReadModelRedactor(readModels...),
	mink.WithSubjectStore(mink.NewAuditSubjectEraser(auditStore)),
)
res, _ := eraser.Erase(ctx, mink.ErasureRequest{SubjectID: "user-123"})

// Scheduled retention: shred / redact / anonymize by age, category, or tenant.
rm := mink.NewRetentionManager(store, []mink.RetentionPolicy{
	{Name: "shred-old-pii", Category: "user", MaxAge: 365 * 24 * time.Hour, Action: mink.ActionShred},
})
report, _ := rm.Apply(ctx)

Explore export & encryption, or read the Security & GDPR guide. The mink gdpr CLI can discover, verify, erase, and dry-run retain.

📝 Audit Logging 
auditStore := memory.NewAuditStore() // PostgreSQL store for production
bus.Use(mink.AuditMiddleware(mink.DefaultAuditConfig(auditStore)))

ctx = mink.WithActor(ctx, "alice@example.com")
bus.Dispatch(ctx, cmd) // records who/what/when/outcome, immutably

entries, _ := auditStore.Find(ctx, mink.AuditQuery{CommandType: "WithdrawFunds"})

See the audit example and the Audit Logging guide.

📈 Observability — Prometheus & OpenTelemetry 
// Prometheus metrics
m := metrics.New(metrics.WithMetricsServiceName("order-service"))
bus.Use(m.CommandMiddleware())

// OpenTelemetry tracing
tracer := tracing.NewTracer(tracing.WithServiceName("order-service"))
bus.Use(tracing.CommandMiddleware(tracer))

See the metrics & tracing examples.

🧪 Testing utilities 
// BDD-style aggregate tests: Given → When → Then
bdd.Given(t, NewOrder("order-123")).
	When(func() error { return order.Create("customer-456") }).
	Then(OrderCreated{OrderID: "order-123", CustomerID: "customer-456"})

// Fluent event assertions
assertions.AssertEventTypes(t, events, "OrderCreated", "ItemAdded")

// Real PostgreSQL in tests via containers
container := containers.StartPostgres(t)
db := container.MustDB(ctx)

See the bdd-testing example and the Testing guide.

Examples

Fifteen runnable, self-contained examples live in examples/ — each with its own README. Most need zero infrastructure (go run ./examples/<name>); two use PostgreSQL via docker compose.

Example Demonstrates Infra
basic Minimal event sourcing: events, aggregates, save/load —
cqrs Command bus + full middleware pipeline —
cqrs-postgres CQRS backed by PostgreSQL 🐘
projections Inline & live read models, query builder —
sagas Process manager with compensation —
versioning Schema evolution via upcasting (v1→v2→v3) —
encryption Field-level encryption + crypto-shredding —
export GDPR data export strategies —
audit Command audit-logging middleware —
metrics Prometheus instrumentation —
tracing OpenTelemetry distributed tracing —
msgpack MessagePack serialization —
protobuf Protobuf serialization —
bdd-testing Given/When/Then aggregate testing —
full-ecommerce Everything together — a realistic system 🐘

New here? Read basic → cqrs → projections → full-ecommerce.

Performance

go-mink is benchmarked on every commit with a shared adapter benchmark suite. Representative figures from CI (ubuntu-latest):

Adapter Append (single) Append (batch 100) Load (1000 events) Concurrent (8 workers)
Memory ~2.5M events/sec ~6M events/sec ~60M events/sec ~1.5M events/sec
PostgreSQL ~5K events/sec ~30K events/sec ~300K events/sec ~15K events/sec

Scale tests: the memory adapter processes 1M events in under a second; the PostgreSQL adapter handles 100K events with p99 append latency under 2 ms.

make benchmark-adapters      # memory adapter (no infra)
make benchmark-adapters-pg   # PostgreSQL adapter (needs infra)

Details, latency percentiles, and a guide for benchmarking new adapters: Benchmarks.

Documentation

Full documentation lives at go-mink.dev.
🚀 Get started Introduction · Architecture · Tutorial
🧠 Core concepts Event Store · Read Models · Adapters
🔬 Advanced Advanced Patterns · Versioning · Security & GDPR · Audit Logging · API Design
🛠️ Guides CLI · Testing · Benchmarks
📚 Learn by reading Blog Series · Architecture Decision Records · Roadmap
🔎 Reference pkg.go.dev/go-mink.dev

Project Status

Current release: v1.0.0 — stable. All core features are complete and the public API is stable under Semantic Versioning. Active development continues on the develop branch (see the roadmap and CHANGELOG).

Releases are automated: every merge to develop publishes a release candidate (v1.0.4-rc.1, …), and merging develop into main cuts the next stable release. Opt into an RC with go get go-mink.dev@v1.0.4-rc.1.

Contributing

Contributions are very welcome — whether it's a bug report, a docs fix, a new adapter, or a feature. 🙌

  1. Read the Contributing Guide and Code of Conduct.
  2. Branch from develop (feature/your-thing or fix/your-thing).
  3. Write tests (we hold 90%+ coverage) and run the checks below.
  4. Open a PR against develop with a clear description.
make test-unit     # fast unit tests, no infrastructure needed
make test          # full suite (auto-starts PostgreSQL + Kafka via docker-compose)
make lint          # golangci-lint
make fmt           # go fmt ./...

Great first contributions: try an example, improve a doc page, add a benchmark, or implement an EventStoreAdapter for your favorite database. Browse good first issues to get started.

Community & Support

  • 💬 GitHub Discussions — questions, ideas, and show-and-tell
  • 🐛 Issues — bugs and feature requests
  • 🔐 Security Policy — report vulnerabilities privately
  • ⭐ Star the repo if go-mink is useful to you — it genuinely helps others discover it.

License

Apache License 2.0 © go-mink contributors.

go-mink — Event Sourcing for Go, done right. 🦫

Documentation

Overview

Package mink provides event sourcing and CQRS primitives for Go applications. It offers a simple, flexible API for building event-sourced systems with support for multiple database backends.

Package mink provides event sourcing and CQRS primitives for Go applications.

go-mink is an Event Sourcing library for Go that makes it easy to build applications using event sourcing patterns. It provides a simple API for storing events, loading aggregates, and projecting read models.

Quick Start

Create an event store with the in-memory adapter for development: 

import (
    "go-mink.dev"
    "go-mink.dev/adapters/memory"
)

store := mink.New(memory.NewAdapter())

For production, use the PostgreSQL adapter: 

import (
    "go-mink.dev"
    "go-mink.dev/adapters/postgres"
)

adapter, err := postgres.NewAdapter(ctx, connStr)
if err != nil {
    log.Fatal(err)
}
store := mink.New(adapter)

Defining Events

Events are simple structs that represent something that happened in your domain: 

type OrderCreated struct {
    OrderID    string `json:"orderId"`
    CustomerID string `json:"customerId"`
}

type ItemAdded struct {
    OrderID  string  `json:"orderId"`
    SKU      string  `json:"sku"`
    Quantity int     `json:"quantity"`
    Price    float64 `json:"price"`
}

Register events with the store so they can be serialized and deserialized: 

store.RegisterEvents(OrderCreated{}, ItemAdded{})

Defining Aggregates

Aggregates are domain objects that encapsulate business logic and generate events: 

type Order struct {
    mink.AggregateBase
    CustomerID string
    Items      []OrderItem
    Status     string
}

func NewOrder(id string) *Order {
    return &Order{
        AggregateBase: mink.NewAggregateBase(id, "Order"),
    }
}

func (o *Order) Create(customerID string) {
    o.Apply(OrderCreated{OrderID: o.AggregateID(), CustomerID: customerID})
    o.CustomerID = customerID
    o.Status = "Created"
}

func (o *Order) ApplyEvent(event interface{}) error {
    switch e := event.(type) {
    case OrderCreated:
        o.CustomerID = e.CustomerID
        o.Status = "Created"
    case ItemAdded:
        o.Items = append(o.Items, OrderItem{SKU: e.SKU, Quantity: e.Quantity, Price: e.Price})
    }
    // NOTE: Version is managed automatically by LoadAggregate and SaveAggregate.
    // You do NOT need to call IncrementVersion() here.
    return nil
}

Saving and Loading Aggregates

Save aggregates to persist their uncommitted events: 

order := NewOrder("order-123")
order.Create("customer-456")
order.AddItem("SKU-001", 2, 29.99)

err := store.SaveAggregate(ctx, order)

Load aggregates to rebuild state from events: 

loaded := NewOrder("order-123")
err := store.LoadAggregate(ctx, loaded)
// loaded.Status == "Created"
// loaded.Items contains the added item
// loaded.Version() returns the number of events in the stream

Low-Level Event Operations

Append events directly to a stream: 

events := []interface{}{
    OrderCreated{OrderID: "123", CustomerID: "456"},
    ItemAdded{OrderID: "123", SKU: "SKU-001", Quantity: 2, Price: 29.99},
}
err := store.Append(ctx, "Order-123", events)

Load events from a stream: 

events, err := store.Load(ctx, "Order-123")

Optimistic Concurrency

Use expected versions to prevent concurrent modifications: 

// Create new stream (must not exist)
err := store.Append(ctx, "Order-123", events, mink.ExpectVersion(mink.NoStream))

// Append to existing stream at specific version
err := store.Append(ctx, "Order-123", events, mink.ExpectVersion(1))

Version constants:

  • AnyVersion (-1): Skip version check
  • NoStream (0): Stream must not exist
  • StreamExists (-2): Stream must exist

Metadata

Add metadata to events for tracing and multi-tenancy: 

metadata := mink.Metadata{}.
    WithUserID("user-123").
    WithCorrelationID("corr-456").
    WithTenantID("tenant-789")

err := store.Append(ctx, "Order-123", events, mink.WithAppendMetadata(metadata))

Commands and CQRS (v0.2.0)

Define commands to encapsulate user intentions: 

type CreateOrder struct {
    mink.CommandBase
    CustomerID string `json:"customerId"`
}

func (c CreateOrder) CommandType() string { return "CreateOrder" }
func (c CreateOrder) Validate() error {
    if c.CustomerID == "" {
        return mink.NewValidationError("CustomerID", "required")
    }
    return nil
}

Create a command bus with middleware: 

bus := mink.NewCommandBus()
bus.Use(mink.ValidationMiddleware())
bus.Use(mink.RecoveryMiddleware(func(err error) { log.Error(err) }))
bus.Use(mink.LoggingMiddleware(logger, nil))

Register command handlers: 

bus.Register("CreateOrder", func(ctx context.Context, cmd mink.Command) (mink.CommandResult, error) {
    c := cmd.(CreateOrder)
    order := NewOrder(uuid.New().String())
    order.Create(c.CustomerID)
    if err := store.SaveAggregate(ctx, order); err != nil {
        return mink.NewErrorResult(err), err
    }
    return mink.NewSuccessResult(order.AggregateID(), order.Version()), nil
})

Dispatch commands: 

result, err := bus.Dispatch(ctx, CreateOrder{CustomerID: "cust-123"})

Idempotency

Prevent duplicate command processing with idempotency: 

idempotencyStore := memory.NewIdempotencyStore()
config := mink.DefaultIdempotencyConfig(idempotencyStore)
bus.Use(mink.IdempotencyMiddleware(config))

Make commands idempotent by implementing IdempotentCommand: 

func (c CreateOrder) IdempotencyKey() string { return c.RequestID }

Index

Constants

View Source 
const (
	// AuditOrderTimestampDesc returns the most recent entries first (default).
	AuditOrderTimestampDesc = adapters.AuditOrderTimestampDesc

	// AuditOrderTimestampAsc returns the oldest entries first.
	AuditOrderTimestampAsc = adapters.AuditOrderTimestampAsc
)

Re-export audit ordering constants for convenience.

View Source 
const (
	// AnyVersion skips version checking, allowing append regardless of current version.
	AnyVersion int64 = -1

	// NoStream indicates the stream must not exist (for creating new streams).
	NoStream int64 = 0

	// StreamExists indicates the stream must exist (for appending to existing streams).
	StreamExists int64 = -2
)

Version constants for optimistic concurrency control.

View Source 
const (
	OutboxPending    = adapters.OutboxPending
	OutboxProcessing = adapters.OutboxProcessing
	OutboxCompleted  = adapters.OutboxCompleted
	OutboxFailed     = adapters.OutboxFailed
	OutboxDeadLetter = adapters.OutboxDeadLetter
)

Outbox status constants.

View Source 
const (
	SagaStatusStarted            = adapters.SagaStatusStarted
	SagaStatusRunning            = adapters.SagaStatusRunning
	SagaStatusCompleted          = adapters.SagaStatusCompleted
	SagaStatusFailed             = adapters.SagaStatusFailed
	SagaStatusCompensating       = adapters.SagaStatusCompensating
	SagaStatusCompensated        = adapters.SagaStatusCompensated
	SagaStatusCompensationFailed = adapters.SagaStatusCompensationFailed
)

Re-export saga status constants from adapters.

View Source 
const (
	SagaStepPending     = adapters.SagaStepPending
	SagaStepRunning     = adapters.SagaStepRunning
	SagaStepCompleted   = adapters.SagaStepCompleted
	SagaStepFailed      = adapters.SagaStepFailed
	SagaStepCompensated = adapters.SagaStepCompensated
)

Re-export saga step status constants from adapters.

View Source 
const (

	// DefaultSchemaVersion is the schema version assumed for events without an explicit version.
	// All events stored before versioning was introduced are treated as version 1.
	DefaultSchemaVersion = 1
)
View Source 
const SubjectTagsKey = "$subjects"

SubjectTagsKey is the reserved Metadata.Custom key under which subject tags are recorded (a JSON array of subject ids). It is exported so adapters can implement a drift-free SubjectIndexAdapter by querying the events' own tags (e.g. a PostgreSQL JSONB query on metadata->'custom'->>'$subjects').

Variables

View Source 
var (
	// ErrEncryptionFailed indicates a field encryption operation failed.
	ErrEncryptionFailed = encryption.ErrEncryptionFailed

	// ErrDecryptionFailed indicates a field decryption operation failed.
	ErrDecryptionFailed = encryption.ErrDecryptionFailed

	// ErrKeyNotFound indicates the requested encryption key does not exist.
	ErrKeyNotFound = encryption.ErrKeyNotFound

	// ErrKeyRevoked indicates the encryption key has been revoked (crypto-shredding).
	ErrKeyRevoked = encryption.ErrKeyRevoked

	// ErrProviderClosed indicates the encryption provider has been closed.
	ErrProviderClosed = encryption.ErrProviderClosed
)

Encryption-related sentinel errors. These are aliases to the encryption package errors for compatibility.

View Source 
var (
	// ErrErasureFailed indicates a data erasure operation failed.
	ErrErasureFailed = errors.New("mink: erasure failed")

	// ErrErasureSubjectRequired indicates the subject ID was not provided.
	ErrErasureSubjectRequired = errors.New("mink: subject ID is required for erasure")

	// ErrNoErasureSources indicates none of streams, filter, or key IDs was provided.
	ErrNoErasureSources = errors.New("mink: streams, filter, or key IDs are required for erasure")

	// ErrErasureNotConfigured indicates the store has no field encryption, so there
	// is nothing to crypto-shred.
	ErrErasureNotConfigured = errors.New("mink: erasure requires field encryption (WithFieldEncryption)")

	// ErrErasureScanNotSupported indicates the adapter does not support event
	// scanning. Provide explicit stream IDs or key IDs instead.
	ErrErasureScanNotSupported = errors.New("mink: adapter does not support event scanning; provide explicit stream IDs or key IDs")

	// ErrSharedKeyRevocation indicates the blast-radius guard (WithSharedKeyGuard)
	// blocked an erasure because a key to revoke also protects events for other
	// subjects (e.g. a per-tenant key). Set AllowSharedKeyRevocation to proceed.
	ErrSharedKeyRevocation = errors.New("mink: erasure would revoke a key shared with other subjects (blast-radius guard); set AllowSharedKeyRevocation to proceed")
)

Erasure-related sentinel errors.

View Source 
var (
	// ErrStreamNotFound indicates the requested stream does not exist.
	ErrStreamNotFound = adapters.ErrStreamNotFound

	// ErrConcurrencyConflict indicates an optimistic concurrency violation.
	ErrConcurrencyConflict = adapters.ErrConcurrencyConflict

	// ErrEventNotFound indicates the requested event does not exist.
	ErrEventNotFound = errors.New("mink: event not found")

	// ErrSerializationFailed indicates event serialization/deserialization failed.
	ErrSerializationFailed = errors.New("mink: serialization failed")

	// ErrBinarySerializerUnsupported indicates a binary serializer (e.g.
	// serializer/msgpack or serializer/protobuf, whose output is not valid JSON
	// text) was paired with an event-store adapter that persists event data in a
	// JSON/JSONB column (see adapters.JSONDataAdapter, implemented by the
	// PostgreSQL adapter). New panics with an error wrapping this sentinel so the
	// misconfiguration surfaces at construction time instead of as a cryptic
	// driver error on the first Append.
	ErrBinarySerializerUnsupported = errors.New("mink: binary serializer is not supported by a JSON-backed event store adapter")

	// ErrEventTypeNotRegistered indicates an unknown event type was encountered.
	ErrEventTypeNotRegistered = errors.New("mink: event type not registered")

	// ErrUnregisteredEventType indicates an aggregate-replay event whose type is not
	// registered, so it deserialized to the map fallback and would be silently dropped
	// from the rebuilt aggregate state. Returned by LoadAggregate under WithStrictReplay.
	ErrUnregisteredEventType = errors.New("mink: unregistered event type on aggregate replay")

	// ErrNilAggregate indicates a nil aggregate was passed.
	ErrNilAggregate = errors.New("mink: nil aggregate")

	// ErrAggregateTypeNotRegistered indicates no factory is registered for an aggregate type.
	ErrAggregateTypeNotRegistered = errors.New("mink: aggregate type not registered")

	// ErrCommandUnsuccessful indicates a command returned an unsuccessful result with no error.
	ErrCommandUnsuccessful = errors.New("mink: command returned unsuccessful result")

	// ErrSagaTimedOut indicates a saga exceeded its configured timeout and was compensated.
	ErrSagaTimedOut = errors.New("mink: saga timed out")

	// ErrNilStore indicates a nil event store was passed.
	ErrNilStore = errors.New("mink: nil event store")

	// ErrNilAuditStore indicates the audit middleware was configured with a nil
	// Store. It is surfaced (only) when the middleware is fail-closed.
	ErrNilAuditStore = errors.New("mink: nil audit store")

	// ErrEmptyStreamID indicates an empty stream ID was provided.
	ErrEmptyStreamID = adapters.ErrEmptyStreamID

	// ErrNoEvents indicates no events were provided for append.
	ErrNoEvents = adapters.ErrNoEvents

	// ErrEventTooLarge indicates an event's serialized data exceeds the configured
	// maximum size (see WithMaxEventSize).
	ErrEventTooLarge = errors.New("mink: event too large")

	// ErrInvalidVersion indicates an invalid version number was provided.
	ErrInvalidVersion = adapters.ErrInvalidVersion

	// ErrAdapterClosed indicates the adapter has been closed.
	ErrAdapterClosed = adapters.ErrAdapterClosed

	// ErrSubscriptionNotSupported indicates the adapter does not support subscriptions.
	ErrSubscriptionNotSupported = errors.New("mink: adapter does not support subscriptions")

	// ErrFilteredFeedNotSupported indicates the adapter does not implement
	// FilteredFeedAdapter (the filtered load-from-position read).
	ErrFilteredFeedNotSupported = errors.New("mink: adapter does not support filtered feed reads")

	// ErrRewriteNotSupported indicates the adapter does not implement
	// adapters.EventRewriteAdapter (the in-place event data/metadata rewrite that
	// ReEncryptStreamInPlace needs).
	ErrRewriteNotSupported = errors.New("mink: adapter does not support in-place event rewrite")

	// ErrHandlerNotFound indicates no handler is registered for a command type.
	ErrHandlerNotFound = errors.New("mink: handler not found")

	// ErrValidationFailed indicates command validation failed.
	ErrValidationFailed = errors.New("mink: validation failed")

	// ErrCommandAlreadyProcessed indicates an idempotent command was already processed.
	ErrCommandAlreadyProcessed = errors.New("mink: command already processed")

	// ErrNilCommand indicates a nil command was passed.
	ErrNilCommand = errors.New("mink: nil command")

	// ErrHandlerPanicked indicates a handler panicked during execution.
	ErrHandlerPanicked = errors.New("mink: handler panicked")

	// ErrCommandBusClosed indicates the command bus has been closed.
	ErrCommandBusClosed = errors.New("mink: command bus closed")

	// ErrOutboxMessageNotFound indicates the requested outbox message does not exist.
	ErrOutboxMessageNotFound = adapters.ErrOutboxMessageNotFound

	// ErrOutboxStoreClosed indicates the outbox store has been closed.
	ErrOutboxStoreClosed = errors.New("mink: outbox store closed")

	// ErrPublisherNotFound indicates no publisher is registered for a destination.
	ErrPublisherNotFound = errors.New("mink: publisher not found for destination")

	// ErrOutboxProcessorRunning indicates the outbox processor is already running.
	ErrOutboxProcessorRunning = errors.New("mink: outbox processor already running")
)

Sentinel errors for common error conditions. Use errors.Is() to check for these errors. These errors are aliases to the adapters package errors for compatibility.

View Source 
var (
	// ErrNilProjection indicates a nil projection was passed.
	ErrNilProjection = errors.New("mink: nil projection")

	// ErrEmptyProjectionName indicates a projection has no name.
	ErrEmptyProjectionName = errors.New("mink: projection name is required")

	// ErrProjectionNotFound indicates the requested projection does not exist.
	ErrProjectionNotFound = errors.New("mink: projection not found")

	// ErrProjectionAlreadyRegistered indicates a projection with the same name is already registered.
	ErrProjectionAlreadyRegistered = errors.New("mink: projection already registered")

	// ErrProjectionEngineAlreadyRunning indicates the projection engine is already running.
	ErrProjectionEngineAlreadyRunning = errors.New("mink: projection engine already running")

	// ErrProjectionEngineStopped indicates the projection engine has been stopped.
	ErrProjectionEngineStopped = errors.New("mink: projection engine stopped")

	// ErrNoCheckpointStore indicates no checkpoint store was configured.
	ErrNoCheckpointStore = errors.New("mink: checkpoint store is required")

	// ErrNotImplemented indicates a method is not implemented.
	ErrNotImplemented = errors.New("mink: not implemented")

	// ErrProjectionFailed indicates a projection failed to process an event.
	ErrProjectionFailed = errors.New("mink: projection failed")

	// ErrTransient marks an error as a transient (retryable) infrastructure failure —
	// e.g. a dropped database connection or a mid-failover read — as opposed to a
	// deterministic "poison" failure that will never succeed on replay. A projection's
	// Apply/ApplyBatch may wrap an infrastructure error with it
	// (fmt.Errorf("...: %w", mink.ErrTransient)) to request retry independently of the
	// poison budget; DefaultErrorClassifier honors it via errors.Is. It carries no
	// behavior on its own — an AsyncOptions.ErrorClassifier must be configured for it to
	// take effect.
	ErrTransient = errors.New("mink: transient error")
)
View Source 
var (
	// ErrExportFailed indicates a data export operation failed.
	ErrExportFailed = errors.New("mink: export failed")

	// ErrSubjectIDRequired indicates the subject ID was not provided in the export request.
	ErrSubjectIDRequired = errors.New("mink: subject ID is required for export")

	// ErrNoExportSources indicates neither streams nor a filter was provided.
	ErrNoExportSources = errors.New("mink: either streams or filter is required for export")

	// ErrExportScanNotSupported indicates the adapter does not support event scanning.
	// Provide explicit stream IDs in the ExportRequest instead.
	ErrExportScanNotSupported = errors.New("mink: adapter does not support event scanning; provide explicit stream IDs")

	// ErrExportPartialFootprint indicates ExportStream auto-resolved a subject to an
	// incomplete footprint. ExportStream has no result object to carry a Partial flag, so
	// rather than silently stream partial data it fails with this error — use Export
	// (which reports ExportResult.Partial), or pass explicit Streams to acknowledge it.
	ErrExportPartialFootprint = errors.New("mink: subject footprint is partial; use Export for ExportResult.Partial, or pass explicit Streams")
)

Export-related sentinel errors.

View Source 
var (
	// ErrNotFound indicates the requested entity was not found.
	ErrNotFound = errors.New("mink: not found")

	// ErrAlreadyExists indicates the entity already exists.
	ErrAlreadyExists = errors.New("mink: already exists")

	// ErrInvalidQuery indicates the query is invalid.
	ErrInvalidQuery = errors.New("mink: invalid query")

	// ErrUnknownFilterField indicates a query filter names a field that does not
	// resolve to a known column / struct field. It wraps ErrInvalidQuery so a
	// non-empty filter set can never silently degrade to an unqualified match-all
	// (e.g. a mistyped field turning DeleteMany into a whole-table delete).
	ErrUnknownFilterField = fmt.Errorf("%w: filter names an unknown field", ErrInvalidQuery)
)

Repository errors

View Source 
var (
	// ErrSagaNotFound indicates the requested saga does not exist.
	ErrSagaNotFound = adapters.ErrSagaNotFound

	// ErrSagaAlreadyExists indicates a saga with the same ID already exists.
	ErrSagaAlreadyExists = adapters.ErrSagaAlreadyExists

	// ErrSagaCompleted indicates the saga has already completed.
	ErrSagaCompleted = errors.New("mink: saga already completed")

	// ErrSagaFailed indicates the saga has failed.
	ErrSagaFailed = errors.New("mink: saga failed")

	// ErrSagaCompensating indicates the saga is currently compensating.
	ErrSagaCompensating = errors.New("mink: saga is compensating")

	// ErrNoSagaHandler indicates no handler is registered for the event type.
	ErrNoSagaHandler = errors.New("mink: no saga handler for event")

	// ErrSagaNotRetryable indicates a saga cannot be re-driven by
	// SagaManager.RetrySaga / ResumeStalled. This is returned when the saga is in a
	// non-retryable status — a terminal Completed, or an in-flight
	// Started/Running/Compensating owned by the event loop or timeout sweep (see
	// SagaStatus.IsRetryable) — or when it has no captured trigger event to replay
	// (e.g. a saga created before this capability existed). Match it with
	// errors.Is(err, ErrSagaNotRetryable); inspect a *SagaNotRetryableError for the
	// saga id, its status, and the human-readable reason.
	ErrSagaNotRetryable = errors.New("mink: saga is not retryable")
)

Saga-related sentinel errors.

View Source 
var (
	// ErrUpcastFailed indicates an upcaster failed to transform event data.
	ErrUpcastFailed = errors.New("mink: upcast failed")

	// ErrSchemaVersionGap indicates a gap in the upcaster chain for an event type.
	ErrSchemaVersionGap = errors.New("mink: schema version gap")

	// ErrIncompatibleSchema indicates a schema change is not backward compatible.
	ErrIncompatibleSchema = errors.New("mink: incompatible schema")

	// ErrSchemaNotFound indicates the requested schema was not found.
	ErrSchemaNotFound = errors.New("mink: schema not found")
)

Sentinel errors for event versioning and upcasting.

View Source 
var ErrColumnValueRange = errors.New("mink: column value out of range for destination field")

ErrColumnValueRange indicates a non-NULL column value does not fit its destination struct field's numeric type — a value outside the field's range (e.g. beyond int8) or a negative value read into an unsigned field. errors.Is(err, ErrColumnValueRange) holds for a *ColumnValueRangeError.

View Source 
var ErrNilAuditEntry = adapters.ErrNilAuditEntry

ErrNilAuditEntry is returned by AuditStore.Append when the entry is nil.

View Source 
var ErrNullColumn = errors.New("mink: NULL read into a non-nullable scalar column")

ErrNullColumn indicates a stored SQL NULL was read into a non-pointer scalar field whose column is not declared `nullable`. errors.Is(err, ErrNullColumn) holds for a *NullColumnError; use errors.As to recover the column/field.

View Source 
var (
	// ErrRetentionMaxScanNeedsCheckpoint indicates WithRetentionMaxScan was configured
	// without WithRetentionCheckpoint. A per-run scan cap only makes sense with a
	// checkpoint to resume the remainder on the next run; without one it would re-scan the
	// same oldest events every run and never reach the aged tail. RetentionManager reports
	// this (non-fatally) in RetentionReport.Errors and runs the sweep unbounded rather than
	// silently capping and forgetting.
	ErrRetentionMaxScanNeedsCheckpoint = errors.New("mink: WithRetentionMaxScan requires WithRetentionCheckpoint to resume across runs; scanning unbounded")
)

Retention-related sentinel errors.

View Source 
var NewDecryptionError = encryption.NewDecryptionError

NewDecryptionError creates a new EncryptionError for a decrypt operation.

View Source 
var NewEncryptionError = encryption.NewEncryptionError

NewEncryptionError creates a new EncryptionError for an encrypt operation.

View Source 
var NewKeyNotFoundError = encryption.NewKeyNotFoundError

NewKeyNotFoundError creates a new KeyNotFoundError.

View Source 
var NewKeyRevokedError = encryption.NewKeyRevokedError

NewKeyRevokedError creates a new KeyRevokedError.

Functions

func ActorFromContext added in v1.1.0 

func ActorFromContext(ctx context.Context) string

ActorFromContext returns the audit actor from context, or "" if not set.

func BackfillSubjectIndex added in v1.1.3 

func BackfillSubjectIndex(ctx context.Context, store *EventStore, tagger SubjectTagger, writer SubjectIndexWriter, batchSize int) (int, error)

BackfillSubjectIndex scans the whole event store and records each event's subjects (derived by tagger, plus any already tagged on the event) into writer. This is the migration step that lets a subject index cover events written BEFORE subject tagging was enabled — making historical subjects fully resolvable, and therefore fully erasable, instead of permanently Partial. It reads RAW events (no decryption) and requires a scannable adapter (SubscriptionAdapter) — returns ErrSubscriptionNotSupported otherwise. Returns the number of events scanned. Safe to re-run (IndexSubjects is idempotent).

func BuildStreamID 

func BuildStreamID(aggregateType, aggregateID string) string

BuildStreamID creates a stream ID from an aggregate type and ID. This follows the convention: "{Type}-{ID}"

func CausationIDFromContext 

func CausationIDFromContext(ctx context.Context) string

CausationIDFromContext returns the causation ID from context.

func CorrelationIDFromContext 

func CorrelationIDFromContext(ctx context.Context) string

CorrelationIDFromContext returns the correlation ID from context.

func GenerateIdempotencyKey 

func GenerateIdempotencyKey(cmd Command) string

GenerateIdempotencyKey generates an idempotency key from a command. The key is based on the command type and its JSON-serialized content.

func GetCommandType 

func GetCommandType(cmd interface{}) string

GetCommandType returns the type name of a command using reflection. This is useful for commands that don't embed CommandBase.

func GetEncryptedFields 

func GetEncryptedFields(m Metadata) []string

GetEncryptedFields extracts the list of encrypted field names from event metadata.

func GetEncryptionAlgorithm added in v1.0.25 

func GetEncryptionAlgorithm(m Metadata) string

GetEncryptionAlgorithm extracts the encryption algorithm recorded in event metadata. It returns an empty string for legacy events written before the algorithm was stamped; callers should treat an empty value as the default AES-256-GCM algorithm.

func GetEncryptionKeyID 

func GetEncryptionKeyID(m Metadata) string

GetEncryptionKeyID extracts the encryption key ID from event metadata.

func GetEventType 

func GetEventType(event interface{}) string

GetEventType returns the event type name for the given event. It uses the struct name as the type name.

func GetIdempotencyKey 

func GetIdempotencyKey(cmd Command) string

GetIdempotencyKey returns the idempotency key for a command. If the command implements IdempotentCommand, it uses that key. Otherwise, it generates a key from the command content.

func GetSchemaVersion 

func GetSchemaVersion(m Metadata) int

GetSchemaVersion extracts the schema version from event metadata. Returns DefaultSchemaVersion if the version is not set or cannot be parsed.

func GetSubjectTags added in v1.1.3 

func GetSubjectTags(m Metadata) []string

GetSubjectTags returns the data-subject ids recorded on an event's metadata, or nil if none.

func IdempotencyKeyFromField 

func IdempotencyKeyFromField(fieldGetter func(Command) string) func(Command) string

IdempotencyKeyFromField extracts the idempotency key from a field in the command. If the field is empty, it falls back to GenerateIdempotencyKey.

func IdempotencyKeyPrefix 

func IdempotencyKeyPrefix(prefix string) func(Command) string

IdempotencyKeyPrefix is a convenience function to create a prefixed idempotency key.

func IsEncrypted 

func IsEncrypted(m Metadata) bool

IsEncrypted reports whether the event has encrypted fields.

func ReEncryptStream added in v1.1.3 

func ReEncryptStream(ctx context.Context, store *EventStore, srcStreamID, dstStreamID string) (copied int, oldKeyIDs []string, err error)

ReEncryptStream re-encrypts a stream's events into dstStreamID under the event store's CURRENT encryption configuration (e.g. a freshly rotated default key), WITHOUT mutating the source stream. This is the append-only way to re-encrypt after a suspected key compromise: events are read (decrypted under the old key), then re-appended to a new stream (re-encrypted under the new key).

IMPORTANT — this function does NOT erase anything. The SOURCE stream and its old-key-recoverable PII remain fully intact and decryptable afterward; a compromised old key still reads the source. Re-encryption is complete only when the caller ALSO retires the source stream and revokes the old key id(s) — which is why the old key ids are returned. Until then the PII exists under both keys.

The source stream must still be decryptable (its key not yet revoked). Non-PII metadata (correlation/causation/tenant/subject tags) is carried over; stale $encryption_* markers are stripped so the new append re-stamps them for the current key. The destination is appended with strict expected-version checks starting from NoStream, so a re-run against an existing destination fails instead of silently duplicating the copy. Returns the number of events copied and the distinct old key ids (for the caller to revoke).

func RegisterGenericHandler 

func RegisterGenericHandler[C Command](registry *HandlerRegistry, handler func(ctx context.Context, cmd C) (CommandResult, error))

RegisterGenericHandler is a convenience function to register a generic handler.

func ReplayEvents added in v1.0.25 

func ReplayEvents(applier EventApplier, target interface{}, events []interface{}) error

ReplayEvents applies a sequence of events to a target using the given EventApplier, stopping at the first error. It is a convenience for rebuilding state from events without an EventStore.

func SagaStateToJSON 

func SagaStateToJSON(state *SagaState) ([]byte, error)

SagaStateToJSON converts saga state to JSON for persistence.

func ShouldHandleEventType 

func ShouldHandleEventType(handledEvents []string, eventType string) bool

ShouldHandleEventType checks if an event type should be handled given a list of handled events. Returns true if handledEvents is empty (meaning all events are handled) or if eventType is in the list. This is a utility function used by projection engine and rebuilder to filter events.

func TenantIDFromContext 

func TenantIDFromContext(ctx context.Context) string

TenantIDFromContext returns the tenant ID from context.

func Version 

func Version() string

Version returns the library version string.

func WithActor added in v1.1.0 

func WithActor(ctx context.Context, actor string) context.Context

WithActor returns a context with the audit actor set.

func WithCausationID 

func WithCausationID(ctx context.Context, causationID string) context.Context

WithCausationID returns a context with the causation ID set.

func WithTenantID 

func WithTenantID(ctx context.Context, tenantID string) context.Context

WithTenantID returns a context with the tenant ID set.

Types

type ActorFunc added in v1.1.0 

type ActorFunc func(ctx context.Context, cmd Command) string

ActorFunc resolves the actor responsible for a command.

type Aggregate 

type Aggregate interface {
	// AggregateID returns the unique identifier for this aggregate instance.
	AggregateID() string

	// AggregateType returns the type/category of this aggregate (e.g., "Order", "Customer").
	AggregateType() string

	// Version returns the current version of the aggregate.
	// This is the number of events that have been applied.
	Version() int64

	// ApplyEvent applies an event to update the aggregate's state.
	// This method should be idempotent and deterministic.
	ApplyEvent(event interface{}) error

	// UncommittedEvents returns events that have been applied but not yet persisted.
	UncommittedEvents() []interface{}

	// ClearUncommittedEvents removes all uncommitted events after successful persistence.
	ClearUncommittedEvents()
}

Aggregate defines the interface for event-sourced aggregates. An aggregate is a domain object whose state is derived from a sequence of events.

type AggregateBase 

type AggregateBase struct {
	// contains filtered or unexported fields
}

AggregateBase provides a default partial implementation of the Aggregate interface. Embed this struct in your aggregate types to get default behavior.

func NewAggregateBase 

func NewAggregateBase(id, aggregateType string) AggregateBase

NewAggregateBase creates a new AggregateBase with the given ID and type.

func (*AggregateBase) AggregateID 

func (a *AggregateBase) AggregateID() string

AggregateID returns the aggregate's unique identifier.

func (*AggregateBase) AggregateType 

func (a *AggregateBase) AggregateType() string

AggregateType returns the aggregate type.

func (*AggregateBase) Apply 

func (a *AggregateBase) Apply(event interface{})

Apply records an event as uncommitted. This should be called by the aggregate after creating a new event. The aggregate should also update its internal state based on the event.

func (*AggregateBase) ClearUncommittedEvents 

func (a *AggregateBase) ClearUncommittedEvents()

ClearUncommittedEvents removes all uncommitted events.

func (*AggregateBase) GetID added in v1.0.25 

func (a *AggregateBase) GetID() string

GetID is an alias for AggregateID following DDD conventions. It lets aggregates embedding AggregateBase satisfy AggregateRoot.

func (*AggregateBase) HasUncommittedEvents 

func (a *AggregateBase) HasUncommittedEvents() bool

HasUncommittedEvents returns true if there are events waiting to be persisted.

func (*AggregateBase) IncrementVersion 

func (a *AggregateBase) IncrementVersion()

IncrementVersion increments the aggregate version by 1.

func (*AggregateBase) OriginalVersion added in v1.0.25 

func (a *AggregateBase) OriginalVersion() int64

OriginalVersion returns the stream version the aggregate had when it was last loaded or saved (informational), and lets aggregates embedding AggregateBase satisfy VersionedAggregate. Note: SaveAggregate uses Version() — not this — for the optimistic-concurrency check; for stock AggregateBase the two are equal because creating uncommitted events does not change Version().

func (*AggregateBase) SetID 

func (a *AggregateBase) SetID(id string)

SetID sets the aggregate's ID.

func (*AggregateBase) SetType 

func (a *AggregateBase) SetType(t string)

SetType sets the aggregate type.

func (*AggregateBase) SetVersion 

func (a *AggregateBase) SetVersion(v int64)

SetVersion sets the aggregate version.

func (*AggregateBase) StreamID 

func (a *AggregateBase) StreamID() StreamID

StreamID returns the stream ID for this aggregate. The stream ID is composed of the aggregate type and ID.

func (*AggregateBase) UncommittedEvents 

func (a *AggregateBase) UncommittedEvents() []interface{}

UncommittedEvents returns events that haven't been persisted yet.

func (*AggregateBase) Version 

func (a *AggregateBase) Version() int64

Version returns the current version of the aggregate.

type AggregateCommand 

type AggregateCommand interface {
	Command

	// AggregateID returns the ID of the aggregate this command targets.
	// Returns empty string for commands that create new aggregates.
	AggregateID() string
}

AggregateCommand is a command that targets a specific aggregate.

type AggregateFactory 

type AggregateFactory func(id string) Aggregate

AggregateFactory creates new aggregate instances.

type AggregateHandler 

type AggregateHandler[C AggregateCommand, A Aggregate] struct {
	// contains filtered or unexported fields
}

AggregateHandler is a handler that works with aggregates and an event store. It loads the aggregate, executes the command, and saves the results.

func NewAggregateHandler 

func NewAggregateHandler[C AggregateCommand, A Aggregate](config AggregateHandlerConfig[C, A]) *AggregateHandler[C, A]

NewAggregateHandler creates a new AggregateHandler.

func (*AggregateHandler[C, A]) CommandType 

func (h *AggregateHandler[C, A]) CommandType() string

CommandType returns the command type this handler processes.

func (*AggregateHandler[C, A]) Handle 

func (h *AggregateHandler[C, A]) Handle(ctx context.Context, cmd Command) (CommandResult, error)

Handle loads the aggregate, executes the command, and saves the aggregate.

type AggregateHandlerConfig 

type AggregateHandlerConfig[C AggregateCommand, A Aggregate] struct {
	Store     *EventStore
	Factory   func(id string) A
	Executor  func(ctx context.Context, agg A, cmd C) error
	NewIDFunc func() string
}

AggregateHandlerConfig configures an AggregateHandler.

type AggregateRegistry added in v1.0.25 

type AggregateRegistry struct {
	// contains filtered or unexported fields
}

AggregateRegistry maps aggregate type names to factories, enabling generic construction of aggregates by type (e.g. for replay tooling or CLI commands). It is safe for concurrent use.

func NewAggregateRegistry added in v1.0.25 

func NewAggregateRegistry() *AggregateRegistry

NewAggregateRegistry creates an empty AggregateRegistry.

func (*AggregateRegistry) Create added in v1.0.25 

func (r *AggregateRegistry) Create(aggregateType, id string) (Aggregate, error)

Create constructs a new aggregate of the given type with the given ID. Returns an error wrapping ErrAggregateTypeNotRegistered (with the type name) if no factory is registered for the type.

func (*AggregateRegistry) Register added in v1.0.25 

func (r *AggregateRegistry) Register(aggregateType string, factory AggregateFactory)

Register associates an aggregate type name with a factory.

type AggregateRoot 

type AggregateRoot interface {
	Aggregate

	// GetID is an alias for AggregateID for DDD conventions.
	GetID() string
}

AggregateRoot is an extended interface that includes domain-driven design patterns.

type Anonymizer added in v1.1.3 

type Anonymizer struct {
	// contains filtered or unexported fields
}

Anonymizer produces stable pseudonyms for PII values. It is deterministic (equal inputs yield equal pseudonyms, preserving referential/analytic continuity) and one-way (the original value cannot be recovered from the pseudonym). It is the transform behind ActionAnonymize, usable from retention Apply hooks and the eraser's read-model redactors.

func NewAnonymizer added in v1.1.3 

func NewAnonymizer(secret []byte, opts ...AnonymizerOption) *Anonymizer

NewAnonymizer creates an Anonymizer keyed by secret. The secret MUST be kept confidential — knowing it does not reveal originals, but it gates linkability.

func (*Anonymizer) Pseudonymize added in v1.1.3 

func (a *Anonymizer) Pseudonymize(scope, value string) string

Pseudonymize returns a stable, one-way pseudonym for value within scope (e.g. a field name or tenant id, so the same value pseudonymizes differently per scope).

type AnonymizerOption added in v1.1.3 

type AnonymizerOption func(*Anonymizer)

AnonymizerOption configures an Anonymizer.

func WithPseudonymLength added in v1.1.3 

func WithPseudonymLength(n int) AnonymizerOption

WithPseudonymLength caps the hex length of the pseudonym (default 32). A value <= 0 or >= 64 uses the full SHA-256 hex.

func WithPseudonymPrefix added in v1.1.3 

func WithPseudonymPrefix(prefix string) AnonymizerOption

WithPseudonymPrefix prepends a fixed prefix to every pseudonym (e.g. "anon_").

type AppendOption 

type AppendOption func(*appendConfig)

AppendOption configures an append operation.

func ExpectVersion 

func ExpectVersion(v int64) AppendOption

ExpectVersion sets the expected stream version for optimistic concurrency.

func WithAppendMetadata 

func WithAppendMetadata(m Metadata) AppendOption

WithMetadata sets metadata for all events in the append operation.

type AsyncOptions 

type AsyncOptions struct {
	// BatchSize is the maximum number of events to process in a batch.
	// Default: 100
	BatchSize int

	// BatchTimeout bounds how long loading and processing a single batch may take.
	// When > 0 the batch runs under a context with this deadline. 0 disables it.
	// Default: 1 second
	BatchTimeout time.Duration

	// PollInterval is how often to poll for new events when idle.
	// Default: 100ms
	PollInterval time.Duration

	// RetryPolicy defines how to handle errors. When set, its ShouldRetry method governs
	// the retry budget and MaxRetries is ignored.
	RetryPolicy RetryPolicy

	// MaxRetries bounds retries for a failing (poison) event before the engine gives up
	// on it, and is used only when RetryPolicy is nil. A positive value gives up after
	// that many attempts; a non-positive value (0 or negative) retries indefinitely — the
	// same convention the built-in ExponentialBackoffRetry/RetryForever policies follow.
	// Default: 3
	MaxRetries int

	// ErrorClassifier optionally classifies a processing error as transient or poison. An
	// error classified ErrorClassTransient is retried with backoff but does not consume
	// the poison budget (MaxRetries / RetryPolicy), so it never reaches OnPoisonEvent and
	// never faults the worker — infrastructure blips self-heal instead of exhausting the
	// budget. An ErrorClassPoison error is accounted exactly as before. nil (the default)
	// classifies every error as poison, making behavior and overhead identical to the
	// prior release. See DefaultErrorClassifier for a batteries-included implementation.
	ErrorClassifier func(error) ErrorClass

	// OnPoisonEvent, if set, is invoked when an event keeps failing after the
	// retry budget is exhausted. Returning nil tells the engine to skip the event
	// (advance past it) and continue; returning an error stops the worker. When
	// nil, the worker stops in the Faulted state after the retries are exhausted.
	OnPoisonEvent func(ctx context.Context, event StoredEvent, err error) error

	// RestartPolicy optionally restarts a Faulted worker with backoff, resuming strictly
	// from its persisted checkpoint (never reprocessing from position 0). nil (the
	// default) leaves a fault terminal — the worker goroutine exits, exactly as before —
	// so this is opt-in and zero-overhead when unset. Reprocessing history remains the
	// exclusive job of Rebuild.
	RestartPolicy RestartPolicy

	// StartFromBeginning starts processing from the beginning of the event stream on the
	// worker's first boot when no checkpoint exists. It never applies to a supervised or
	// manual restart, which always resume from the checkpoint.
	// If false, starts from the last checkpoint.
	// Default: false
	StartFromBeginning bool
}

AsyncOptions configures async projection behavior.

func DefaultAsyncOptions 

func DefaultAsyncOptions() AsyncOptions

DefaultAsyncOptions returns the default async projection options.

type AsyncProjection 

type AsyncProjection interface {
	Projection

	// Apply processes a single event asynchronously.
	Apply(ctx context.Context, event StoredEvent) error

	// ApplyBatch processes multiple events in a single batch for efficiency.
	// If not supported, implementations should process events sequentially.
	ApplyBatch(ctx context.Context, events []StoredEvent) error
}

AsyncProjection processes events asynchronously in the background. This provides eventual consistency but better write performance and scalability.

type AsyncProjectionBase 

type AsyncProjectionBase struct {
	ProjectionBase
}

AsyncProjectionBase provides a default implementation of AsyncProjection. Embed this struct and override Apply to create async projections.

func NewAsyncProjectionBase 

func NewAsyncProjectionBase(name string, handledEvents ...string) AsyncProjectionBase

NewAsyncProjectionBase creates a new AsyncProjectionBase.

func (*AsyncProjectionBase) ApplyBatch 

func (p *AsyncProjectionBase) ApplyBatch(ctx context.Context, events []StoredEvent) error

ApplyBatch provides a default implementation that processes events sequentially. Override this method for custom batch processing logic.

type AsyncResult 

type AsyncResult struct {
	// contains filtered or unexported fields
}

AsyncResult represents the result of an asynchronous operation. It provides methods to wait for completion and check the result.

func (*AsyncResult) Cancel 

func (r *AsyncResult) Cancel()

Cancel cancels the async operation's context. This can be used to stop a long-running operation early.

func (*AsyncResult) Context 

func (r *AsyncResult) Context() context.Context

Context returns the context for this async operation.

func (*AsyncResult) Done 

func (r *AsyncResult) Done() <-chan struct{}

Done returns a channel that is closed when the operation completes. Use this in select statements for non-blocking wait patterns.

Example: 

select {
case <-result.Done():
    if err := result.Err(); err != nil {
        log.Printf("Failed: %v", err)
    }
case <-time.After(5 * time.Second):
    result.Cancel()
    log.Println("Timed out")
}

func (*AsyncResult) Err 

func (r *AsyncResult) Err() error

Err returns the error from the completed operation, or nil if not yet complete or successful.

func (*AsyncResult) IsComplete 

func (r *AsyncResult) IsComplete() bool

IsComplete returns true if the operation has completed (successfully or with an error).

func (*AsyncResult) Wait 

func (r *AsyncResult) Wait() error

Wait blocks until the operation completes and returns any error. This is a convenience method equivalent to <-result.Done(); return result.Err()

func (*AsyncResult) WaitWithTimeout 

func (r *AsyncResult) WaitWithTimeout(timeout time.Duration) error

WaitWithTimeout blocks until the operation completes or the timeout expires. Returns context.DeadlineExceeded if the timeout is reached before completion.

type AuditConfig added in v1.1.0 

type AuditConfig struct {
	// Store is the audit store that persists the trail. Required.
	Store AuditStore

	// ActorFunc resolves the actor for each command. If nil, the actor is read
	// from the context via ActorFromContext.
	ActorFunc ActorFunc

	// SkipCommands lists command types that should not be audited.
	SkipCommands []string

	// FailClosed determines behavior when the audit store write fails. If true,
	// the audit write failure is surfaced as the command result/error (note: the
	// command's side effect has already run — auditing is not transactional). If
	// false (the default), the failure is ignored (fail-open) and the original
	// command result is returned.
	FailClosed bool

	// IncludeMetadata copies the command's metadata map into the audit entry when
	// the command exposes one via GetMetadataMap() map[string]string.
	IncludeMetadata bool
	// contains filtered or unexported fields
}

AuditConfig configures the audit logging middleware.

func DefaultAuditConfig added in v1.1.0 

func DefaultAuditConfig(store AuditStore) AuditConfig

DefaultAuditConfig returns a default audit configuration for the given store.

type AuditEntry added in v1.1.0 

type AuditEntry = adapters.AuditEntry

AuditEntry is a single immutable record in the command audit trail.

type AuditOrder added in v1.1.0 

type AuditOrder = adapters.AuditOrder

AuditOrder controls how audit entries are sorted when queried.

type AuditQuery added in v1.1.0 

type AuditQuery = adapters.AuditQuery

AuditQuery filters and paginates a query against the audit trail.

type AuditStore added in v1.1.0 

type AuditStore = adapters.AuditStore

AuditStore persists and queries the command audit trail.

type BinaryFormatReporter added in v1.1.6 

type BinaryFormatReporter interface {
	// BinaryFormat reports whether Serialize produces binary (non-JSON-text) output.
	BinaryFormat() bool
}

BinaryFormatReporter is an optional interface a Serializer may implement to declare whether its Serialize output is binary — i.e. not valid JSON text. The shipped serializer/msgpack and serializer/protobuf serializers implement it returning true; the default JSONSerializer returns false.

It is consulted by New, together with adapters.JSONDataAdapter, so that pairing a binary serializer with a JSON/JSONB-backed event store (such as the PostgreSQL adapter) fails fast at construction with an actionable error rather than deep in the driver on the first Append. A serializer that does not implement this interface is assumed to produce JSON-compatible text (the historical default), so existing custom serializers are unaffected.

type CatchupSubscription 

type CatchupSubscription struct {
	// contains filtered or unexported fields
}

CatchupSubscription provides catch-up subscription functionality. It first reads historical events from the event store, then switches to polling for new events. This ensures no events are missed during the transition.

func NewCatchupSubscription 

func NewCatchupSubscription(
	store *EventStore,
	fromPosition uint64,
	opts ...SubscriptionOptions,
) (*CatchupSubscription, error)

NewCatchupSubscription creates a new catch-up subscription. Call Start() to begin receiving events from the specified position.

func (*CatchupSubscription) Close 

func (s *CatchupSubscription) Close() error

Close stops the subscription.

func (*CatchupSubscription) Err 

func (s *CatchupSubscription) Err() error

Err returns any error that caused the subscription to close.

func (*CatchupSubscription) Events 

func (s *CatchupSubscription) Events() <-chan StoredEvent

Events returns the channel for receiving events.

func (*CatchupSubscription) Position 

func (s *CatchupSubscription) Position() uint64

Position returns the current position of the subscription.

func (*CatchupSubscription) Start 

func (s *CatchupSubscription) Start(ctx context.Context, pollInterval time.Duration) error

Start begins the catch-up subscription with the specified poll interval. It first catches up on historical events, then polls for new events.

type CategoryFilter 

type CategoryFilter struct {
	// contains filtered or unexported fields
}

CategoryFilter filters events by stream category.

func NewCategoryFilter 

func NewCategoryFilter(category string) *CategoryFilter

NewCategoryFilter creates a filter that only matches events from streams in the category.

func (*CategoryFilter) Matches 

func (f *CategoryFilter) Matches(event StoredEvent) bool

Matches returns true if the event's stream is in the category.

type CertificateSink added in v1.1.3 

type CertificateSink func(ctx context.Context, cert ErasureCertificate) error

CertificateSink receives an erasure certificate (no PII) for recording, e.g. via an AuditStore. Configure with WithCertificateSink.

type Checkpoint 

type Checkpoint struct {
	// ProjectionName is the name of the projection.
	ProjectionName string

	// Position is the global position of the last processed event.
	Position uint64

	// UpdatedAt is when the checkpoint was last updated.
	UpdatedAt time.Time
}

Checkpoint represents a stored checkpoint record.

type CheckpointStore 

type CheckpointStore interface {
	// GetCheckpoint returns the last processed position for a projection.
	// Returns 0 if no checkpoint exists.
	GetCheckpoint(ctx context.Context, projectionName string) (uint64, error)

	// SetCheckpoint stores the last processed position for a projection.
	SetCheckpoint(ctx context.Context, projectionName string, position uint64) error

	// DeleteCheckpoint removes the checkpoint for a projection.
	DeleteCheckpoint(ctx context.Context, projectionName string) error

	// GetAllCheckpoints returns checkpoints for all projections.
	GetAllCheckpoints(ctx context.Context) (map[string]uint64, error)
}

CheckpointStore manages projection checkpoints. Checkpoints track the last processed position for each projection.

type Clearable 

type Clearable interface {
	// Clear removes all data from the read model.
	Clear(ctx context.Context) error
}

Clearable is an interface for projections that can clear their read model.

type ColumnValueRangeError added in v1.1.11 

type ColumnValueRangeError struct {
	Column string // SQL column
	Field  string // destination struct field (empty if unresolved)
	GoType string // Go type of the field (empty if unresolved)
	Value  string // the out-of-range value as read from the column
}

ColumnValueRangeError reports that a non-NULL value read from a `nullable` scalar column overflows its destination field's Go type. The NULL-safe read path scans nullable integers/floats through a wide intermediate (int64 / float64) and copies the value back with reflection, which would otherwise silently truncate or wrap an out-of-range value; this error preserves the fail-loud behavior database/sql applies on a direct scan. Widen the field type (e.g. int8 -> int32, or a signed type for a value that can be negative) to resolve it.

func (*ColumnValueRangeError) Error added in v1.1.11 

func (e *ColumnValueRangeError) Error() string

func (*ColumnValueRangeError) Is added in v1.1.11 

func (e *ColumnValueRangeError) Is(target error) bool

Is reports a match against the ErrColumnValueRange sentinel.

func (*ColumnValueRangeError) Unwrap added in v1.1.11 

func (e *ColumnValueRangeError) Unwrap() error

Unwrap returns the ErrColumnValueRange sentinel.

type Command 

type Command interface {
	// CommandType returns the type identifier for this command (e.g., "CreateOrder").
	CommandType() string

	// Validate checks if the command is valid.
	// Returns nil if valid, or an error describing validation failures.
	Validate() error
}

Command represents an intent to change state in the system. Commands are the write side of CQRS and should be validated before execution.

type CommandBase 

type CommandBase struct {
	// CommandID is an optional unique identifier for this command instance.
	CommandID string `json:"commandId,omitempty"`

	// CorrelationID links related commands and events for distributed tracing.
	CorrelationID string `json:"correlationId,omitempty"`

	// CausationID identifies the event or command that caused this command.
	CausationID string `json:"causationId,omitempty"`

	// Metadata contains arbitrary key-value pairs for application-specific data.
	Metadata map[string]string `json:"metadata,omitempty"`
}

CommandBase provides a default partial implementation of Command. Embed this struct in your command types to get common functionality.

func (CommandBase) GetCausationID 

func (c CommandBase) GetCausationID() string

GetCausationID returns the causation ID.

func (CommandBase) GetCommandID 

func (c CommandBase) GetCommandID() string

GetCommandID returns the command ID.

func (CommandBase) GetCorrelationID 

func (c CommandBase) GetCorrelationID() string

GetCorrelationID returns the correlation ID.

func (CommandBase) GetMetadata 

func (c CommandBase) GetMetadata(key string) string

GetMetadata returns the value for a metadata key, or empty string if not found.

func (CommandBase) GetMetadataMap added in v1.1.0 

func (c CommandBase) GetMetadataMap() map[string]string

GetMetadataMap returns a defensive copy of the command's metadata map (or nil when empty), so callers cannot mutate CommandBase's internal state. It enables audit logging (via AuditConfig.IncludeMetadata) to capture command metadata.

func (CommandBase) WithCausationID 

func (c CommandBase) WithCausationID(id string) CommandBase

WithCausationID returns a copy of CommandBase with the causation ID set.

func (CommandBase) WithCommandID 

func (c CommandBase) WithCommandID(id string) CommandBase

WithCommandID returns a copy of CommandBase with the command ID set.

func (CommandBase) WithCorrelationID 

func (c CommandBase) WithCorrelationID(id string) CommandBase

WithCorrelationID returns a copy of CommandBase with the correlation ID set.

func (CommandBase) WithMetadata 

func (c CommandBase) WithMetadata(key, value string) CommandBase

WithMetadata returns a copy of CommandBase with a metadata key-value pair added.

type CommandBus 

type CommandBus struct {
	// contains filtered or unexported fields
}

CommandBus orchestrates command dispatching with middleware support. It routes commands to their handlers through a configurable middleware pipeline.

func NewCommandBus 

func NewCommandBus(opts ...CommandBusOption) *CommandBus

NewCommandBus creates a new CommandBus with the given options.

func (*CommandBus) Close 

func (b *CommandBus) Close() error

Close closes the command bus, preventing further dispatch operations and blocking until all in-flight dispatches (including those started via DispatchAsync) have completed.

Close must not be called from within a command handler: the calling dispatch is itself in-flight, so Close would wait on its own completion and deadlock. Trigger shutdown from outside the handler (e.g. signal a separate goroutine).

func (*CommandBus) Dispatch 

func (b *CommandBus) Dispatch(ctx context.Context, cmd Command) (CommandResult, error)

Dispatch sends a command through the middleware pipeline to its handler.

func (*CommandBus) DispatchAll 

func (b *CommandBus) DispatchAll(ctx context.Context, cmds ...Command) ([]DispatchResult, error)

DispatchAll dispatches multiple commands and returns all results. Commands are dispatched sequentially in order.

func (*CommandBus) DispatchAsync 

func (b *CommandBus) DispatchAsync(ctx context.Context, cmd Command) <-chan DispatchResult

DispatchAsync sends a command asynchronously and returns immediately. The result can be retrieved through the returned channel.

func (*CommandBus) HandlerCount 

func (b *CommandBus) HandlerCount() int

HandlerCount returns the number of registered handlers.

func (*CommandBus) HasHandler 

func (b *CommandBus) HasHandler(cmdType string) bool

HasHandler returns true if a handler is registered for the command type.

func (*CommandBus) IsClosed 

func (b *CommandBus) IsClosed() bool

IsClosed returns true if the command bus has been closed.

func (*CommandBus) MiddlewareCount 

func (b *CommandBus) MiddlewareCount() int

MiddlewareCount returns the number of registered middleware.

func (*CommandBus) Register 

func (b *CommandBus) Register(handler CommandHandler)

Register adds a handler to the command bus.

func (*CommandBus) RegisterFunc 

func (b *CommandBus) RegisterFunc(cmdType string, fn func(ctx context.Context, cmd Command) (CommandResult, error))

RegisterFunc registers a handler function for a command type.

func (*CommandBus) Use 

func (b *CommandBus) Use(middleware ...Middleware)

Use adds middleware to the command bus. Middleware is executed in the order it was added.

type CommandBusOption 

type CommandBusOption func(*CommandBus)

CommandBusOption configures a CommandBus.

func WithHandlerRegistry 

func WithHandlerRegistry(registry *HandlerRegistry) CommandBusOption

WithHandlerRegistry sets a custom handler registry.

func WithMiddleware 

func WithMiddleware(middleware ...Middleware) CommandBusOption

WithMiddleware adds middleware to the command bus.

type CommandContext 

type CommandContext struct {
	// Context is the standard Go context.
	Context context.Context

	// Command is the command being executed.
	Command Command

	// Result is the command execution result (set by handler).
	Result CommandResult

	// Metadata contains additional context data that can be set by middleware.
	Metadata map[string]interface{}
}

CommandContext carries command execution context through the middleware chain.

func NewCommandContext 

func NewCommandContext(ctx context.Context, cmd Command) *CommandContext

NewCommandContext creates a new CommandContext.

func (*CommandContext) Get 

func (c *CommandContext) Get(key string) (interface{}, bool)

Get retrieves a value from the context metadata.

func (*CommandContext) GetString 

func (c *CommandContext) GetString(key string) string

GetString retrieves a string value from the context metadata.

func (*CommandContext) Set 

func (c *CommandContext) Set(key string, value interface{})

Set stores a value in the context metadata.

func (*CommandContext) SetError 

func (c *CommandContext) SetError(err error)

SetError sets an error result.

func (*CommandContext) SetResult 

func (c *CommandContext) SetResult(result CommandResult)

SetResult sets the command execution result.

func (*CommandContext) SetSuccess 

func (c *CommandContext) SetSuccess(aggregateID string, version int64)

SetSuccess sets a successful result.

type CommandDispatcher 

type CommandDispatcher interface {
	// Dispatch sends a command to its handler and returns the result.
	Dispatch(ctx context.Context, cmd Command) (CommandResult, error)
}

CommandDispatcher can dispatch commands to handlers.

type CommandHandler 

type CommandHandler interface {
	// CommandType returns the type of command this handler processes.
	CommandType() string

	// Handle processes the command and returns a result.
	Handle(ctx context.Context, cmd Command) (CommandResult, error)
}

CommandHandler is the interface for handling a specific command type. Handlers contain the business logic for processing commands.

type CommandHandlerFunc 

type CommandHandlerFunc struct {
	// contains filtered or unexported fields
}

CommandHandlerFunc is a function type that implements CommandHandler.

func NewCommandHandlerFunc 

func NewCommandHandlerFunc(cmdType string, fn func(ctx context.Context, cmd Command) (CommandResult, error)) *CommandHandlerFunc

NewCommandHandlerFunc creates a new CommandHandlerFunc.

func (*CommandHandlerFunc) CommandType 

func (h *CommandHandlerFunc) CommandType() string

CommandType returns the command type this handler processes.

func (*CommandHandlerFunc) Handle 

func (h *CommandHandlerFunc) Handle(ctx context.Context, cmd Command) (CommandResult, error)

Handle processes the command.

type CommandResult 

type CommandResult struct {
	// Success indicates whether the command executed successfully.
	Success bool

	// AggregateID is the ID of the aggregate affected by the command.
	// For create commands, this is the ID of the newly created aggregate.
	AggregateID string

	// Version is the new version of the aggregate after command execution.
	Version int64

	// Data contains any additional result data.
	Data interface{}

	// Error contains the error if the command failed.
	Error error
}

CommandResult represents the result of command execution. It can contain either a successful result or an error.

func IdempotencyRecordToResult 

func IdempotencyRecordToResult(r *IdempotencyRecord) CommandResult

IdempotencyRecordToResult converts the record to a CommandResult.

func NewErrorResult 

func NewErrorResult(err error) CommandResult

NewErrorResult creates a failed CommandResult.

func NewSuccessResult 

func NewSuccessResult(aggregateID string, version int64) CommandResult

NewSuccessResult creates a successful CommandResult.

func NewSuccessResultWithData 

func NewSuccessResultWithData(aggregateID string, version int64, data interface{}) CommandResult

NewSuccessResultWithData creates a successful CommandResult with additional data.

func (CommandResult) IsError 

func (r CommandResult) IsError() bool

IsError returns true if the command failed.

func (CommandResult) IsSuccess 

func (r CommandResult) IsSuccess() bool

IsSuccess returns true if the command executed successfully.

type CompositeFilter 

type CompositeFilter struct {
	// contains filtered or unexported fields
}

CompositeFilter combines multiple filters with AND logic.

func NewCompositeFilter 

func NewCompositeFilter(filters ...EventFilter) *CompositeFilter

NewCompositeFilter creates a filter that matches only if all filters match.

func (*CompositeFilter) Matches 

func (f *CompositeFilter) Matches(event StoredEvent) bool

Matches returns true if all filters match.

type ConcurrencyError 

type ConcurrencyError struct {
	StreamID        string
	ExpectedVersion int64
	ActualVersion   int64
}

ConcurrencyError provides detailed information about a concurrency conflict.

func NewConcurrencyError 

func NewConcurrencyError(streamID string, expected, actual int64) *ConcurrencyError

NewConcurrencyError creates a new ConcurrencyError.

func (*ConcurrencyError) Error 

func (e *ConcurrencyError) Error() string

Error returns the error message.

func (*ConcurrencyError) Is 

func (e *ConcurrencyError) Is(target error) bool

Is reports whether this error matches the target error.

func (*ConcurrencyError) Unwrap 

func (e *ConcurrencyError) Unwrap() error

Unwrap returns the underlying error for errors.Unwrap().

type ContextValueMiddleware 

type ContextValueMiddleware struct {
	// contains filtered or unexported fields
}

ContextValueMiddleware adds values to the context.

func NewContextValueMiddleware 

func NewContextValueMiddleware(key, value interface{}) *ContextValueMiddleware

NewContextValueMiddleware creates middleware that adds a value to the context.

func (*ContextValueMiddleware) Middleware 

func (m *ContextValueMiddleware) Middleware() Middleware

Middleware returns the middleware function.

type DataEraser added in v1.1.3 

type DataEraser struct {
	// contains filtered or unexported fields
}

DataEraser performs GDPR right-to-erasure (Article 17) by crypto-shredding the encryption keys behind a data subject's events. It is the erasure counterpart to DataExporter and shares the same subject-targeting model (Streams / Filter).

Erasure revokes the master key(s) under which the subject's PII was encrypted — rendering those fields permanently unrecoverable — then optionally appends an append-only marker event recording that the erasure occurred. It never deletes or mutates historical event rows.

IMPORTANT: crypto-shredding erases ALL data encrypted under a revoked key. Subject-scoped erasure therefore requires per-subject keys; with per-tenant keys (WithTenantKeyResolver) revoking a key erases the whole tenant. ErasureResult reports exactly which keys were revoked so the caller can verify the blast radius.

func NewDataEraser added in v1.1.3 

func NewDataEraser(store *EventStore, opts ...DataEraserOption) *DataEraser

NewDataEraser creates a new DataEraser for the given event store.

func (*DataEraser) Erase added in v1.1.3 

func (e *DataEraser) Erase(ctx context.Context, req ErasureRequest) (*ErasureResult, error)

Erase crypto-shreds the data subject identified by req and returns a result. It is idempotent: re-erasing an already-erased subject revokes already-revoked keys as a no-op and succeeds. Partial failures (a key that fails to revoke, a marker that fails to append) are reported in ErasureResult.Errors, not returned fatally. A provider that does not support revocation, and a store without field encryption, are returned as typed (fatal) errors.

func (*DataEraser) Verify added in v1.1.3 

func (e *DataEraser) Verify(ctx context.Context, subjectID string) (*VerificationReport, error)

Verify checks that no recoverable PII remains for the subject across its event footprint. It is deterministic — an encrypted event is erased iff its key is revoked. Requires a configured SubjectResolver (WithEraseSubjectResolver).

type DataEraserOption added in v1.1.3 

type DataEraserOption func(*DataEraser)

DataEraserOption configures a DataEraser.

func AllowSharedKeyRevocation added in v1.1.3 

func AllowSharedKeyRevocation() DataEraserOption

AllowSharedKeyRevocation overrides WithSharedKeyGuard, permitting revocation of a shared (e.g. per-tenant) key with full knowledge of the blast radius. ErasureResult.KeysRevoked reports every key revoked.

func WithCertificateSink added in v1.1.3 

func WithCertificateSink(sink CertificateSink) DataEraserOption

WithCertificateSink makes Erase verify the erasure and emit a PII-free ErasureCertificate to the sink (e.g. an AuditStore) for Article 17 accountability. Sink failure is non-fatal (recorded in ErasureResult.Errors).

func WithEraseBatchSize added in v1.1.3 

func WithEraseBatchSize(size int) DataEraserOption

WithEraseBatchSize sets the number of events loaded per batch during scan-based key discovery. Default is 1000.

func WithEraseLogger added in v1.1.3 

func WithEraseLogger(l Logger) DataEraserOption

WithEraseLogger sets the logger for the data eraser.

func WithEraseSubjectResolver added in v1.1.3 

func WithEraseSubjectResolver(r *SubjectResolver) DataEraserOption

WithEraseSubjectResolver configures a SubjectResolver so a SubjectID-only ErasureRequest (no Streams/Filter/KeyIDs) is automatically resolved to the subject's complete footprint. A partial footprint propagates to ErasureResult.Partial — never a silent partial.

func WithErasureHook added in v1.1.3 

func WithErasureHook(hooks ...ErasureHook) DataEraserOption

WithErasureHook registers side-effect hooks run during Erase to clean PII the application owns outside the event store (blob storage, exports, caches, search indexes). Failures are reported on ErasureResult, not fatal.

func WithErasureMarker added in v1.1.3 

func WithErasureMarker(streamID string) DataEraserOption

WithErasureMarker makes Erase append an append-only ErasureMarker event to the given stream after a successful erasure, recording that the subject was erased (subject reference, timestamp, and key count — never the erased PII fields). Register ErasureMarker with the store's serializer to load markers back. Leave unset to skip.

func WithReadModelRebuilder added in v1.1.3 

func WithReadModelRebuilder(rebuilders ...ReadModelRebuilder) DataEraserOption

WithReadModelRebuilder registers projection rebuilders. After shredding, Erase triggers each one so crypto-shredded events re-materialize as redacted. Use this when a read model has no in-place SubjectRedactable hook.

func WithReadModelRedactor added in v1.1.3 

func WithReadModelRedactor(redactors ...SubjectRedactable) DataEraserOption

WithReadModelRedactor registers in-place read-model redactors. After shredding the subject's keys, Erase calls each redactor's RedactSubject so PII does not survive on the read side. Preferred over a rebuild when available.

func WithSharedKeyGuard added in v1.1.3 

func WithSharedKeyGuard() DataEraserOption

WithSharedKeyGuard makes Erase refuse — BEFORE any (irreversible) revocation — to revoke a key that also protects events tagged for a subject other than the target, or untagged events whose ownership cannot be proven. This catches the per-tenant-key blast radius (erasing one subject would crypto-shred the whole tenant). On a shared key it returns *SharedKeyError (errors.Is ErrSharedKeyRevocation). Pair with AllowSharedKeyRevocation to override. Zero overhead when unset. Requires a scannable adapter (SubscriptionAdapter).

func WithStrictAccountability added in v1.1.3 

func WithStrictAccountability() DataEraserOption

WithStrictAccountability makes a failure to append the erasure marker or emit the certificate FATAL (returned as an error) AFTER the idempotent key revocation, so a lost accountability record cannot be mistaken for a proven erasure. Re-run Erase to retry — revocation is idempotent. Default is best-effort (failures are recorded in ErasureResult.Errors, not returned).

func WithSubjectStore added in v1.1.3 

func WithSubjectStore(stores ...SubjectErasable) DataEraserOption

WithSubjectStore registers stores that hold PII derived from events (audit trail, saga state, snapshots, external sinks) so Erase reaches them too. Each is run after key revocation and read-model redaction; a per-store failure is non-fatal and reported in ErasureResult.SubjectStores (symmetric with WithErasureHook). Use the built-in NewAuditSubjectEraser / NewSagaSubjectEraser / NewSnapshotSubjectEraser or any custom SubjectErasable.

type DataExporter 

type DataExporter struct {
	// contains filtered or unexported fields
}

DataExporter handles GDPR data export (right to access / right to data portability). It collects events belonging to a data subject from the event store, decrypts encrypted fields, and returns them in a portable format.

When encrypted fields cannot be decrypted (e.g., key revoked via crypto-shredding), those events are included with Redacted=true and nil Data.

There are two enumeration strategies:

  • Stream-based: provide explicit stream IDs in ExportRequest.Streams.
  • Scan-based: provide an ExportFilter and the exporter scans all events. Requires the adapter to implement SubscriptionAdapter.

func NewDataExporter 

func NewDataExporter(store *EventStore, opts ...DataExporterOption) *DataExporter

NewDataExporter creates a new DataExporter for the given event store.

func (*DataExporter) Export 

func (e *DataExporter) Export(ctx context.Context, req ExportRequest) (*ExportResult, error)

Export collects all matching events for a data subject and returns them. Use ExportStream for large exports that should not be held in memory.

func (*DataExporter) ExportStream 

func (e *DataExporter) ExportStream(ctx context.Context, req ExportRequest, handler ExportHandler) error

ExportStream calls handler for each matching event, without holding all events in memory. This is suitable for large exports. Events are yielded in stream order for stream-based export, or global position order for scan-based export. Return a non-nil error from the handler to stop the export early. When a SubjectID-only request auto-resolves to a partial footprint, ExportStream returns ErrExportPartialFootprint rather than streaming incomplete data — use Export for the flag.

type DataExporterOption 

type DataExporterOption func(*DataExporter)

DataExporterOption configures a DataExporter.

func WithExportBatchSize 

func WithExportBatchSize(size int) DataExporterOption

WithExportBatchSize sets the number of events loaded per batch during scan-based export. Default is 1000.

func WithExportLogger 

func WithExportLogger(l Logger) DataExporterOption

WithExportLogger sets the logger for the data exporter.

func WithExportSubjectResolver added in v1.1.3 

func WithExportSubjectResolver(r *SubjectResolver) DataExporterOption

WithExportSubjectResolver configures a SubjectResolver so a SubjectID-only ExportRequest (no Streams/Filter) is automatically resolved to the subject's complete footprint — for both Export and ExportStream. A partial footprint is never silently exported: Export surfaces it as ExportResult.Partial, while ExportStream (which has no result object) fails with ErrExportPartialFootprint.

type DispatchResult 

type DispatchResult struct {
	CommandResult
	Error error
}

DispatchResult contains the result of an asynchronous dispatch operation.

func (DispatchResult) IsSuccess 

func (r DispatchResult) IsSuccess() bool

IsSuccess returns true if the dispatch was successful.

type EncryptionError 

type EncryptionError = encryption.EncryptionError

EncryptionError provides detailed information about an encryption or decryption failure.

type EncryptionOption 

type EncryptionOption func(*FieldEncryptionConfig)

EncryptionOption configures a FieldEncryptionConfig.

func WithDecryptionErrorHandler 

func WithDecryptionErrorHandler(handler func(err error, eventType string, metadata Metadata) error) EncryptionOption

WithDecryptionErrorHandler sets a handler for decryption errors. This is used for crypto-shredding: when a key has been deleted, the handler can return nil to skip the event or return a custom error. If the handler returns nil, the event data is returned as-is (still encrypted).

func WithDefaultKeyID 

func WithDefaultKeyID(keyID string) EncryptionOption

WithDefaultKeyID sets the default master key ID used when no tenant key resolver is configured or when the tenant ID is empty.

func WithEncryptedFields 

func WithEncryptedFields(eventType string, fields ...string) EncryptionOption

WithEncryptedFields registers field paths to encrypt for a given event type. Field paths are dot-separated JSON field names (e.g., "email", "address.street").

Which master key wraps each event is chosen by resolveKeyID. When a SubjectTagger (WithSubjectTagger) is configured, an event with an empty Metadata.TenantID — every event persisted via SaveAggregate — keys off its first $subjects tag instead of the default key, so the one tagger that defines a subject's erasure footprint also selects its shred key; the two can never drift. Pair with WithTenantKeyResolver or WithSubjectKeyResolver to map that id to a per-subject key. See WithSharedKeyGuard (blast-radius guard) for the complementary detection of keys shared across subjects.

func WithEncryptionProvider 

func WithEncryptionProvider(p encryption.Provider) EncryptionOption

WithEncryptionProvider sets the encryption provider.

func WithSubjectKeyResolver added in v1.1.5 

func WithSubjectKeyResolver(resolver func(subjectID string) string) EncryptionOption

WithSubjectKeyResolver sets a function that maps a data-subject id to a master key ID, enabling per-subject encryption keys so a subject's PII can be individually crypto-shredded (GDPR Article 17) by revoking its key.

It is a legibility alias of WithTenantKeyResolver: both configure the one resolver used by resolveKeyID, which selects the input as metadata.TenantID when present and otherwise the first $subjects tag (WithSubjectTagger) — so with a tagger configured, SaveAggregate events resolve to a per-subject key. Prefer this name for per-subject / GDPR setups and WithTenantKeyResolver for per-tenant ones. Because both set the same field, passing both is unambiguous: the last option applied wins.

func WithTenantKeyResolver 

func WithTenantKeyResolver(resolver func(tenantID string) string) EncryptionOption

WithTenantKeyResolver sets a function that maps tenant IDs to master key IDs. This enables per-tenant encryption keys for multi-tenant applications.

The resolver also drives the subject-tag fallback in resolveKeyID: when an event's Metadata.TenantID is empty (e.g. every SaveAggregate event) but a SubjectTagger has recorded a $subjects tag, the first tag is passed to this same resolver, giving aggregate PII a per-subject key. For per-subject setups prefer the WithSubjectKeyResolver alias, which reads by intent. Cross-link: WithSharedKeyGuard.

type ErasureCertificate added in v1.1.3 

type ErasureCertificate struct {
	SubjectID     string    `json:"subjectId"`
	VerifiedAt    time.Time `json:"verifiedAt"`
	Verified      bool      `json:"verified"`
	EventsChecked int       `json:"eventsChecked"`
	KeysRevoked   int       `json:"keysRevoked"`
	Partial       bool      `json:"partial"`

	// SideEffects names the external-PII domains erased alongside the event store
	// (no PII), e.g. "blob-storage", "search-index".
	SideEffects []string `json:"sideEffects,omitempty"`
}

ErasureCertificate is a PII-free record that an erasure was performed and verified, suitable for recording via an AuditStore for Article 17 accountability.

type ErasureContext added in v1.1.3 

type ErasureContext struct {
	SubjectID   string
	Streams     []string
	KeysRevoked []string
}

ErasureContext is passed to erasure side-effect hooks. It carries only references (subject id, affected streams, revoked keys) — never the erased PII itself.

type ErasureError added in v1.1.3 

type ErasureError struct {
	SubjectID string
	Cause     error
}

ErasureError provides detailed information about a data erasure failure.

func NewErasureError added in v1.1.3 

func NewErasureError(subjectID string, cause error) *ErasureError

NewErasureError creates a new ErasureError.

func (*ErasureError) Error added in v1.1.3 

func (e *ErasureError) Error() string

Error returns the error message.

func (*ErasureError) Is added in v1.1.3 

func (e *ErasureError) Is(target error) bool

Is reports whether this error matches the target error.

func (*ErasureError) Unwrap added in v1.1.3 

func (e *ErasureError) Unwrap() error

Unwrap returns the underlying cause for errors.Unwrap().

type ErasureHook added in v1.1.3 

type ErasureHook struct {
	Name string
	Run  func(ctx context.Context, ec ErasureContext) error
}

ErasureHook erases PII an application owns OUTSIDE the event store (blob storage, generated exports/PDFs, caches, search indexes, short URLs) as part of Erase. Run failures are reported on ErasureResult, never fatal. Register with WithErasureHook.

type ErasureMarker added in v1.1.3 

type ErasureMarker struct {
	SubjectID   string    `json:"subjectId"`
	ErasedAt    time.Time `json:"erasedAt"`
	KeysRevoked int       `json:"keysRevoked"`
}

ErasureMarker is the default payload appended when WithErasureMarker is set. It records that an erasure occurred WITHOUT carrying any erased PII.

type ErasureRequest added in v1.1.3 

type ErasureRequest struct {
	// SubjectID identifies the data subject (required). It is a reference/identifier
	// used for the audit marker, not the erased personal data itself.
	SubjectID string

	// Streams lists specific stream IDs whose events identify the keys to revoke.
	Streams []string

	// Filter selects which events identify the keys to revoke. When Streams is empty
	// and Filter is set, the eraser scans all events (requires the adapter to
	// implement SubscriptionAdapter).
	Filter ExportFilter

	// KeyIDs explicitly names master keys to revoke. They are revoked in addition to
	// any keys discovered from matched events; when set with no Streams/Filter, the
	// keys are revoked directly without scanning.
	KeyIDs []string

	// FromTime / ToTime bound the events considered during key discovery.
	FromTime *time.Time
	ToTime   *time.Time
}

ErasureRequest describes which data subject to erase. It mirrors ExportRequest.

For a race-free erasure, quiesce the subject's writes first (e.g. mark the subject non-writable) before calling Erase: an event appended between key discovery and revocation under a new key can otherwise survive. Erase mitigates this by re-resolving once and revoking newcomers, flagging ErasureResult.Partial if the footprint grew — but the guarantee only holds when writes are quiesced.

type ErasureResult added in v1.1.3 

type ErasureResult struct {
	// SubjectID is the data subject identifier from the request.
	SubjectID string

	// KeysRevoked lists the master keys that were successfully revoked
	// (sorted, de-duplicated).
	KeysRevoked []string

	// Streams lists the streams that contained matched events (sorted).
	Streams []string

	// EventsScanned is the number of events examined during key discovery.
	EventsScanned int

	// MarkerWritten reports whether an erasure-marker event was appended.
	MarkerWritten bool

	// Partial is true when the erasure was driven by a subject footprint that could
	// not be proven complete (e.g. legacy untagged events). The caller MUST treat
	// the erasure as incomplete and remediate (e.g. backfill subject tags).
	Partial bool

	// RedactedReadModels lists read models redacted in place or rebuilt to redacted.
	RedactedReadModels []string

	// ResidualReadModels lists read models that could not be redacted and may still
	// hold the subject's PII (residual risk; see Errors for the cause).
	ResidualReadModels []string

	// SideEffects lists external-PII erasure hooks that ran successfully.
	SideEffects []string

	// SubjectStores reports the outcome of each registered SubjectErasable (sibling
	// stores holding derived PII: audit trail, saga state, snapshots). See
	// WithSubjectStore.
	SubjectStores []SubjectErasureOutcome

	// ErasedAt is when the erasure was performed.
	ErasedAt time.Time

	// Errors holds non-fatal per-key / marker failures (partial-failure contract).
	Errors []error
}

ErasureResult reports the outcome of an erasure.

func (*ErasureResult) Failed added in v1.1.3 

func (r *ErasureResult) Failed() bool

Failed reports whether the erasure did not fully succeed: any non-nil Errors (e.g. a key that failed to revoke — a Vault key without deletion_allowed, a partial provider outage), a partial footprint, a residual read model, or a registered subject store that was skipped (its optional purger interface is unimplemented, so the subject's PII survives there). Because partial failures are non-fatal by contract (Erase returns a nil error), a caller SHOULD check Failed() — or inspect Errors / the gap between requested keys and KeysRevoked — rather than only the returned error.

type ErrorClass added in v1.1.13 

type ErrorClass int

ErrorClass classifies a projection processing error for retry accounting. Its zero value is ErrorClassPoison, so an unclassified error behaves exactly as before this type existed. See AsyncOptions.ErrorClassifier and DefaultErrorClassifier. 

const (
	// ErrorClassPoison is the zero value: the error counts against the poison retry
	// budget (AsyncOptions.MaxRetries / RetryPolicy) and, once the budget is exhausted,
	// reaches OnPoisonEvent or faults the worker — the classic poison-event path.
	ErrorClassPoison ErrorClass = iota

	// ErrorClassTransient marks a retryable infrastructure error: it is retried with
	// backoff but does NOT consume the poison budget, so it never trips OnPoisonEvent and
	// never faults the worker. Use it for errors that a later attempt can succeed on
	// (dropped connections, failovers), letting the worker self-heal when infra returns.
	ErrorClassTransient
)

func DefaultErrorClassifier added in v1.1.13 

func DefaultErrorClassifier(err error) ErrorClass

DefaultErrorClassifier classifies err as ErrorClassTransient when err, or any error in its Unwrap chain, (a) satisfies errors.Is(err, ErrTransient), (b) implements Retryable() bool returning true, or (c) implements interface{ Temporary() bool } returning true (the net.Error idiom); otherwise it returns ErrorClassPoison. It deliberately does NOT treat context.DeadlineExceeded as transient (even though that error reports Temporary() == true), so a genuinely hung poison event is not retried forever behind a batch timeout; a caller who wants deadlines treated as transient must classify them explicitly. Exported so a custom classifier can wrap or compose it, e.g. to add driver-specific transient codes.

type Event 

type Event struct {
	// ID is the globally unique event identifier.
	ID string

	// StreamID identifies the stream this event belongs to.
	StreamID string

	// Type is the event type identifier.
	Type string

	// Data is the deserialized event payload.
	Data interface{}

	// Metadata contains contextual information.
	Metadata Metadata

	// Version is the position within the stream (1-based).
	Version int64

	// GlobalPosition is the position across all streams.
	GlobalPosition uint64

	// Timestamp is when the event was stored.
	Timestamp time.Time
}

Event represents a deserialized event with its data as a Go type. This is the high-level representation used by applications.

func DeserializeEvent 

func DeserializeEvent(serializer Serializer, stored StoredEvent) (Event, error)

DeserializeEvent is a convenience function that deserializes a StoredEvent to an Event.

func EventFromStored 

func EventFromStored(stored StoredEvent, data interface{}) Event

EventFromStored creates an Event from a StoredEvent with deserialized data.

type EventApplier 

type EventApplier func(aggregate interface{}, event interface{}) error

EventApplier is a function type for applying events to aggregates.

type EventData 

type EventData struct {
	// Type is the event type identifier (e.g., "OrderCreated").
	Type string

	// Data is the serialized event payload.
	Data []byte

	// Metadata contains optional contextual information.
	Metadata Metadata
}

EventData represents an event to be stored. It contains the event type, serialized payload, and optional metadata.

func NewEventData 

func NewEventData(eventType string, data []byte) EventData

NewEventData creates a new EventData with the given type and data.

func SerializeEvent 

func SerializeEvent(serializer Serializer, event interface{}, metadata Metadata) (EventData, error)

SerializeEvent is a convenience function that serializes an event and returns EventData.

func SerializeEventWithVersion 

func SerializeEventWithVersion(serializer Serializer, event interface{}, metadata Metadata, schemaVersion int) (EventData, error)

SerializeEventWithVersion is a convenience function that serializes an event and stamps the metadata with the given schema version.

func (EventData) Validate 

func (e EventData) Validate() error

Validate checks if the EventData is valid.

func (EventData) WithMetadata 

func (e EventData) WithMetadata(m Metadata) EventData

WithMetadata returns a copy of EventData with the metadata set.

type EventFilter 

type EventFilter interface {
	// Matches returns true if the event should be delivered.
	Matches(event StoredEvent) bool
}

EventFilter determines which events should be delivered.

type EventRegistry 

type EventRegistry struct {
	// contains filtered or unexported fields
}

EventRegistry maps event type names to Go types. It is used by the JSONSerializer to deserialize events to the correct type.

func NewEventRegistry 

func NewEventRegistry() *EventRegistry

NewEventRegistry creates a new empty EventRegistry.

func (*EventRegistry) Count 

func (r *EventRegistry) Count() int

Count returns the number of registered event types.

func (*EventRegistry) Lookup 

func (r *EventRegistry) Lookup(eventType string) (reflect.Type, bool)

Lookup returns the Go type for the given event type name. Returns nil and false if the type is not registered.

func (*EventRegistry) Register 

func (r *EventRegistry) Register(eventType string, example interface{})

Register adds a mapping from eventType to the Go type of the example. The example should be a value (not a pointer) of the event type.

func (*EventRegistry) RegisterAll 

func (r *EventRegistry) RegisterAll(examples ...interface{})

RegisterAll registers multiple events using their struct names as type names. Each example should be a value (not a pointer) of the event type.

func (*EventRegistry) RegisteredTypes 

func (r *EventRegistry) RegisteredTypes() []string

RegisteredTypes returns a slice of all registered event type names.

type EventStore 

type EventStore struct {
	// contains filtered or unexported fields
}

EventStore is the main entry point for event sourcing operations. It provides methods for appending events, loading aggregates, and managing streams.

func New 

func New(adapter adapters.EventStoreAdapter, opts ...Option) *EventStore

New creates a new EventStore with the given adapter and options.

If a binary serializer (one whose BinaryFormat reports true, such as serializer/msgpack or serializer/protobuf) is paired with an adapter that stores event data as JSON text (one whose RequiresJSONData reports true, such as the PostgreSQL JSONB adapter), the pairing can never succeed — the binary payload is rejected by the JSON column on every write. New does not panic on this: it records the incompatibility so the first Append/SaveAggregate returns a clear ErrBinarySerializerUnsupported instead of a cryptic driver error. Use the default JSON serializer with JSON-backed adapters.

func (*EventStore) Adapter 

func (s *EventStore) Adapter() adapters.EventStoreAdapter

Adapter returns the underlying adapter.

func (*EventStore) Append 

func (s *EventStore) Append(ctx context.Context, streamID string, events []interface{}, opts ...AppendOption) error

Append stores events to the specified stream. Events can be Go structs which will be serialized using the configured serializer.

func (*EventStore) Close 

func (s *EventStore) Close() error

Close releases resources held by the event store.

func (*EventStore) DecryptStoredEvent added in v1.1.6 

func (s *EventStore) DecryptStoredEvent(ctx context.Context, stored StoredEvent) (StoredEvent, error)

DecryptStoredEvent returns stored with its field-encrypted Data decrypted, reusing the same decrypt + WithDecryptionErrorHandler path as the deserializing read paths (Load/LoadAggregate/DataExporter). It is the StoredEvent-preserving counterpart to ProcessStoredEvent (which returns a deserialized Event): the value stays a StoredEvent and only its Data changes. Upcasting is intentionally NOT applied here — raw fields are preserved for Type-based consumers (projections, subscriptions).

It is the single primitive every raw-StoredEvent delivery surface (the projection engine, event subscriptions, live catch-up) routes through, so decryption transparency is defined in exactly one place. Passthrough with zero overhead when no encryption is configured or the event is not encrypted. A crypto-shredded subject is handled exactly as elsewhere: decryptFields consults WithDecryptionErrorHandler; a handler that returns nil yields the event with its fields left as stored and no error. A hard, unhandled decryption error is returned for the caller to decide (retry / poison / fail).

func (*EventStore) EncryptStoredEvent added in v1.1.8 

func (s *EventStore) EncryptStoredEvent(ctx context.Context, stored StoredEvent) (StoredEvent, error)

EncryptStoredEvent is the encode counterpart to DecryptStoredEvent: it seals a stored event's configured fields under the current FieldEncryptionConfig and stamps the current envelope in its metadata, exactly as the append path (prepareEventData) would — the configured SubjectTagger runs first so the wrapping key is resolved identically. It is a passthrough (input returned unchanged) when no encryption is configured, the event type has no configured fields, or the event is already encrypted. Zero overhead when unused.

func (*EventStore) EncryptionConfig added in v1.1.3 

func (s *EventStore) EncryptionConfig() *FieldEncryptionConfig

EncryptionConfig returns the field-encryption configuration, or nil if the store was not configured with WithFieldEncryption. The erasure machinery (DataEraser) uses it to crypto-shred a subject's keys.

func (*EventStore) GetLastPosition 

func (s *EventStore) GetLastPosition(ctx context.Context) (uint64, error)

GetLastPosition returns the global position of the last stored event.

func (*EventStore) GetStreamInfo 

func (s *EventStore) GetStreamInfo(ctx context.Context, streamID string) (*StreamInfo, error)

GetStreamInfo returns metadata about a stream.

func (*EventStore) Initialize 

func (s *EventStore) Initialize(ctx context.Context) error

Initialize sets up the required storage schema.

func (*EventStore) Load 

func (s *EventStore) Load(ctx context.Context, streamID string) ([]Event, error)

Load retrieves all events from a stream.

func (*EventStore) LoadAggregate 

func (s *EventStore) LoadAggregate(ctx context.Context, agg Aggregate) error

LoadAggregate loads an aggregate's state by replaying its events. The aggregate should be a new instance with its ID and type already set.

If the aggregate implements VersionSetter, the version will be set to the number of events loaded. This is required for proper optimistic concurrency control when saving the aggregate later.

Note: AggregateBase implements VersionSetter, so aggregates embedding AggregateBase will automatically have their version set correctly.

func (*EventStore) LoadEventsFromPosition 

func (s *EventStore) LoadEventsFromPosition(ctx context.Context, fromPosition uint64, limit int) ([]StoredEvent, error)

LoadEventsFromPosition loads events starting from a global position. Returns ErrSubscriptionNotSupported if the adapter does not implement SubscriptionAdapter. This is a helper method used by ProjectionEngine and ProjectionRebuilder.

func (*EventStore) LoadEventsFromPositionFiltered added in v1.1.8 

func (s *EventStore) LoadEventsFromPositionFiltered(ctx context.Context, fromPosition uint64, limit int, filter FeedFilter) ([]StoredEvent, error)

LoadEventsFromPositionFiltered loads events from a global position that match an indexed FeedFilter (event type, stream id, and/or category), pushing the predicate down to storage instead of scanning the whole feed. It returns ErrFilteredFeedNotSupported if the adapter does not implement FilteredFeedAdapter. Results keep the LoadEventsFromPosition contract: ascending global position, exclusive of fromPosition, bounded by limit, and (on the PostgreSQL adapter) subject to the same gapless safe watermark. An empty filter is equivalent to LoadEventsFromPosition.

This is an introspection / ops / migration primitive for reading the raw event log by indexed axis (audit browsers, backfill scanners, diagnostics) — NOT an application read path. For queries over unindexed data (payload fields, a tenant in metadata) or for application reads, project a read model instead.

func (*EventStore) LoadFrom 

func (s *EventStore) LoadFrom(ctx context.Context, streamID string, fromVersion int64) ([]Event, error)

LoadFrom retrieves events from a stream starting from the specified version.

func (*EventStore) LoadRaw 

func (s *EventStore) LoadRaw(ctx context.Context, streamID string, fromVersion int64) ([]StoredEvent, error)

LoadRaw retrieves raw (non-deserialized) events from a stream.

func (*EventStore) ProcessStoredEvent 

func (s *EventStore) ProcessStoredEvent(ctx context.Context, stored StoredEvent) (Event, error)

ProcessStoredEvent applies decryption, upcasting, and deserialization to a stored event. This is used by components that work with raw StoredEvents (e.g., DataExporter) and need the full processing pipeline.

func (*EventStore) ReEncryptStreamInPlace added in v1.1.8 

func (s *EventStore) ReEncryptStreamInPlace(ctx context.Context, streamID string) (int, []string, error)

ReEncryptStreamInPlace brings a stream's existing events under the current field-encryption scheme in place: it seals each not-yet-encrypted event's configured fields under the current key and rewrites the same row, preserving the event's id, type, stream, version, global position, and timestamp (only the at-rest encoding of data/metadata changes — history is not rewritten). It is the opt-in, operator-triggered historical backfill for a consumer that enabled encryption after it already had data, making that PII crypto-shreddable; contrast with ReEncryptStream, which copies to a NEW stream (for rotation).

It is idempotent — an event already encrypted, of a type with no configured fields, or whose configured field is absent from this particular event, is skipped — so a re-run or a resume after a mid-stream failure is safe. Returns the number of events re-encrypted and the distinct wrapping key ids used. Requires an adapter implementing adapters.EventRewriteAdapter (else ErrRewriteNotSupported); a no-op with zero overhead when no encryption is configured.

func (*EventStore) RegisterAggregateEvents added in v1.1.6 

func (s *EventStore) RegisterAggregateEvents(events ...interface{})

RegisterAggregateEvents registers a set of event types in one call. It is an alias of RegisterEvents named for the common, important case: register every event type an aggregate applies so LoadAggregate never silently drops one. Pre-flight it in a test with RegisteredEventTypes.

func (*EventStore) RegisterEvents 

func (s *EventStore) RegisterEvents(events ...interface{})

RegisterEvents registers event types with the serializer. This is required for deserializing events back to their original types.

func (*EventStore) RegisterUpcasters 

func (s *EventStore) RegisterUpcasters(upcasters ...Upcaster) error

RegisterUpcasters is a convenience method that registers upcasters with the event store. If no UpcasterChain has been configured, a new one is created automatically. It is safe to call concurrently with Load/Append (the upcaster pointer is atomic and UpcasterChain.Register is internally synchronized).

func (*EventStore) RegisteredEventTypes added in v1.1.6 

func (s *EventStore) RegisteredEventTypes() []string

RegisteredEventTypes returns the event type names registered with the serializer, or nil if the serializer does not support introspection. Use it to assert, in a test, that every event type your aggregates apply is registered — a pre-flight against silent replay drops.

func (*EventStore) SaveAggregate 

func (s *EventStore) SaveAggregate(ctx context.Context, agg Aggregate) error

SaveAggregate persists uncommitted events from an aggregate. The aggregate's current version (agg.Version()) is used as the expected stream version for optimistic concurrency control.

Invariant: creating uncommitted events (via AggregateBase.Apply) must not change Version(). Version is only advanced by ApplyEvent while replaying stored events during LoadAggregate. If a command mutator also increments the version before save, the expected version will no longer match the stored version and the save will fail with ErrConcurrencyConflict.

After a successful save, if the aggregate implements VersionSetter, the version is updated to reflect the new stream version (and OriginalVersion is advanced), allowing subsequent modifications without reloading.

func (*EventStore) Serializer 

func (s *EventStore) Serializer() Serializer

Serializer returns the event store's serializer.

func (*EventStore) SubscribeAll added in v1.0.25 

func (s *EventStore) SubscribeAll(ctx context.Context, fromPosition uint64, opts ...SubscriptionOptions) (Subscription, error)

SubscribeAll subscribes to all events starting from the given global position. It returns a started catch-up subscription backed by the store's adapter.

func (*EventStore) SubscribeCategory added in v1.0.25 

func (s *EventStore) SubscribeCategory(ctx context.Context, category string, fromPosition uint64, opts ...SubscriptionOptions) (Subscription, error)

SubscribeCategory subscribes to events from all streams in a category, starting from the given global position.

func (*EventStore) SubscribeStream added in v1.0.25 

func (s *EventStore) SubscribeStream(ctx context.Context, streamID string, fromVersion int64, opts ...SubscriptionOptions) (Subscription, error)

SubscribeStream subscribes to events from a single stream with version greater than fromVersion (exclusive, consistent with Load). When the adapter implements SubscriptionAdapter it delegates to the adapter-native per-stream subscription; otherwise it falls back to a catch-up subscription filtered by stream.

func (*EventStore) UnregisteredEventTypes added in v1.1.6 

func (s *EventStore) UnregisteredEventTypes(ctx context.Context) ([]string, error)

UnregisteredEventTypes scans the entire event log (in batches) and returns the distinct event type names present in storage that are not registered — the all-streams counterpart to UnregisteredStreamTypes. Read-only; requires a SubscriptionAdapter for the global scan (returns ErrSubscriptionNotSupported otherwise), and nil when the serializer lacks introspection.

Completeness caveat: the scan uses the same position-load path as delivery, which on the PostgreSQL adapter is capped at the gapless safe watermark. Events from still-in-flight transactions are therefore excluded, so run this against a quiescent store for a complete pre-deploy/migration audit — under concurrent writers it may miss the most recent events.

func (*EventStore) UnregisteredStreamTypes added in v1.1.6 

func (s *EventStore) UnregisteredStreamTypes(ctx context.Context, streamID string) ([]string, error)

UnregisteredStreamTypes scans a single stream and returns the distinct event type names present in storage that are not registered with the serializer — a read-only pre-deploy / migration safety check that never mutates the append-only log. It is empty when every type in the stream is registered, and nil when the serializer does not support introspection.

type EventStoreWithOutbox 

type EventStoreWithOutbox struct {
	// contains filtered or unexported fields
}

EventStoreWithOutbox wraps an EventStore to automatically schedule outbox messages when events are appended. If the adapter implements OutboxAppender, events and outbox messages are written atomically in the same transaction.

func NewEventStoreWithOutbox 

func NewEventStoreWithOutbox(store *EventStore, outboxStore OutboxStore, routes []OutboxRoute, opts ...OutboxOption) *EventStoreWithOutbox

NewEventStoreWithOutbox creates a new EventStoreWithOutbox wrapper.

func (*EventStoreWithOutbox) Append 

func (es *EventStoreWithOutbox) Append(ctx context.Context, streamID string, events []interface{}, opts ...AppendOption) error

Append stores events and schedules outbox messages.

func (*EventStoreWithOutbox) OutboxStore 

func (es *EventStoreWithOutbox) OutboxStore() OutboxStore

OutboxStore returns the underlying OutboxStore.

func (*EventStoreWithOutbox) SaveAggregate 

func (es *EventStoreWithOutbox) SaveAggregate(ctx context.Context, agg Aggregate) error

SaveAggregate persists uncommitted events and schedules outbox messages.

func (*EventStoreWithOutbox) Store 

func (es *EventStoreWithOutbox) Store() *EventStore

Store returns the underlying EventStore.

type EventSubscriber 

type EventSubscriber interface {
	// SubscribeAll subscribes to all events starting from the given position.
	SubscribeAll(ctx context.Context, fromPosition uint64, opts ...SubscriptionOptions) (Subscription, error)

	// SubscribeStream subscribes to events from a specific stream.
	SubscribeStream(ctx context.Context, streamID string, fromVersion int64, opts ...SubscriptionOptions) (Subscription, error)

	// SubscribeCategory subscribes to events from all streams in a category.
	SubscribeCategory(ctx context.Context, category string, fromPosition uint64, opts ...SubscriptionOptions) (Subscription, error)
}

EventSubscriber provides event subscription capabilities.

type EventTypeFilter 

type EventTypeFilter struct {
	// contains filtered or unexported fields
}

EventTypeFilter filters events by type.

func NewEventTypeFilter 

func NewEventTypeFilter(eventTypes ...string) *EventTypeFilter

NewEventTypeFilter creates a filter that only matches the specified event types.

func (*EventTypeFilter) Matches 

func (f *EventTypeFilter) Matches(event StoredEvent) bool

Matches returns true if the event type is in the filter.

type EventTypeNotRegisteredError 

type EventTypeNotRegisteredError struct {
	EventType string
}

EventTypeNotRegisteredError provides detailed information about an unregistered event type.

func NewEventTypeNotRegisteredError 

func NewEventTypeNotRegisteredError(eventType string) *EventTypeNotRegisteredError

NewEventTypeNotRegisteredError creates a new EventTypeNotRegisteredError.

func (*EventTypeNotRegisteredError) Error 

func (e *EventTypeNotRegisteredError) Error() string

Error returns the error message.

func (*EventTypeNotRegisteredError) Is 

func (e *EventTypeNotRegisteredError) Is(target error) bool

Is reports whether this error matches the target error.

func (*EventTypeNotRegisteredError) Unwrap 

func (e *EventTypeNotRegisteredError) Unwrap() error

Unwrap returns the underlying error for errors.Unwrap().

type EventTypeRegistrar added in v1.1.6 

type EventTypeRegistrar interface {
	// IsRegistered reports whether eventType resolves to a concrete registered Go type
	// (rather than the untyped map fallback used for unknown types).
	IsRegistered(eventType string) bool
	// RegisteredEventTypes returns the names of all registered event types.
	RegisteredEventTypes() []string
}

EventTypeRegistrar is an optional serializer capability: reporting which event types are registered. The default JSONSerializer implements it. The aggregate replay-safety checks (unregistered-type detection, WithStrictReplay, EventStore.RegisteredEventTypes, and the stream audit) use it when the configured serializer provides it; a serializer that does not implement it simply disables those checks (they no-op) rather than failing.

type ExportError 

type ExportError struct {
	SubjectID string
	Cause     error
}

ExportError provides detailed information about a data export failure.

func NewExportError 

func NewExportError(subjectID string, cause error) *ExportError

NewExportError creates a new ExportError.

func (*ExportError) Error 

func (e *ExportError) Error() string

Error returns the error message.

func (*ExportError) Is 

func (e *ExportError) Is(target error) bool

Is reports whether this error matches the target error.

func (*ExportError) Unwrap 

func (e *ExportError) Unwrap() error

Unwrap returns the underlying cause for errors.Unwrap().

type ExportFilter 

type ExportFilter func(event StoredEvent) bool

ExportFilter determines whether a stored event should be included in a data export.

func CombineFilters 

func CombineFilters(filters ...ExportFilter) ExportFilter

CombineFilters returns a filter that matches events passing ALL provided filters (AND logic).

func FilterByEventTypes 

func FilterByEventTypes(types ...string) ExportFilter

FilterByEventTypes returns a filter that matches events of any of the given types.

func FilterByMetadata 

func FilterByMetadata(key, value string) ExportFilter

FilterByMetadata returns a filter that matches events with a specific custom metadata key-value pair.

func FilterByStreamPrefix 

func FilterByStreamPrefix(prefix string) ExportFilter

FilterByStreamPrefix returns a filter that matches events from streams whose ID starts with the given prefix.

func FilterByTenantID 

func FilterByTenantID(tenantID string) ExportFilter

FilterByTenantID returns a filter that matches events with the given tenant ID.

func FilterByUserID 

func FilterByUserID(userID string) ExportFilter

FilterByUserID returns a filter that matches events with the given user ID.

func SubjectFilter added in v1.1.3 

func SubjectFilter(subjectID string) ExportFilter

SubjectFilter returns an ExportFilter matching events tagged with subjectID. It bridges subject tagging into the export/erasure scan model.

type ExportHandler 

type ExportHandler func(ctx context.Context, event ExportedEvent) error

ExportHandler is called for each exported event during streaming export. Return a non-nil error to stop the export.

type ExportRequest 

type ExportRequest struct {
	// SubjectID identifies the data subject (required).
	SubjectID string

	// Streams lists specific stream IDs to export.
	// When provided, only these streams are loaded (efficient, no full scan).
	Streams []string

	// Filter selects which events to include.
	// When Streams is empty, the exporter scans all events and applies this filter
	// (requires the adapter to implement SubscriptionAdapter).
	// When Streams is provided, the filter is applied within each stream.
	Filter ExportFilter

	// FromTime limits export to events stored at or after this time.
	FromTime *time.Time

	// ToTime limits export to events stored at or before this time.
	ToTime *time.Time
}

ExportRequest describes what data to export for a data subject.

type ExportResult 

type ExportResult struct {
	// SubjectID is the data subject identifier from the request.
	SubjectID string

	// Events contains all exported events, ordered by stream then version.
	Events []ExportedEvent

	// Streams lists all unique stream IDs that contained matching events.
	Streams []string

	// TotalEvents is the total number of events included (including redacted).
	TotalEvents int

	// RedactedCount is the number of events whose PII could not be decrypted
	// (e.g., due to crypto-shredding / key revocation).
	RedactedCount int

	// ExportedAt is the timestamp when the export was generated.
	ExportedAt time.Time

	// Partial is true when the export was driven by a subject footprint that could
	// not be proven complete (e.g. legacy untagged events); the caller MUST treat
	// the export as potentially incomplete.
	Partial bool
}

ExportResult contains all exported data for a data subject.

type ExportedEvent 

type ExportedEvent struct {
	// StreamID identifies the stream this event belongs to.
	StreamID string

	// EventType is the event type identifier.
	EventType string

	// Data is the deserialized event payload.
	// When Redacted is true, Data is nil.
	Data interface{}

	// RawData is the serialized event payload after decryption and upcasting
	// (plaintext JSON bytes). When Redacted is true, RawData instead contains the
	// original (still-encrypted) on-disk bytes.
	RawData []byte

	// Metadata contains non-PII contextual information about the event.
	Metadata ExportedMetadata

	// Version is the position within the stream (1-based).
	Version int64

	// GlobalPosition is the position across all streams.
	GlobalPosition uint64

	// Timestamp is when the event was stored.
	Timestamp time.Time

	// Redacted indicates the event's encrypted fields could not be decrypted
	// (e.g., because the encryption key was revoked via crypto-shredding).
	Redacted bool
}

ExportedEvent represents a single event in the data export.

type ExportedMetadata 

type ExportedMetadata struct {
	CorrelationID string
	CausationID   string
	TenantID      string
	SchemaVersion int
}

ExportedMetadata contains non-PII metadata included in the export.

type FeedFilter added in v1.1.8 

type FeedFilter = adapters.FeedFilter

FeedFilter selects events for a filtered load-from-position read by indexed axis (event type, stream id, or category). It is an input DTO with no conversion step, so it is aliased to the adapter type rather than duplicated. See adapters.FeedFilter.

type FieldDefinition 

type FieldDefinition struct {
	// Name is the field name.
	Name string

	// Type is the field type (e.g., "string", "int", "bool").
	Type string

	// Required indicates whether the field must be present.
	Required bool
}

FieldDefinition describes a single field in an event schema.

type FieldEncryptionConfig 

type FieldEncryptionConfig struct {
	// contains filtered or unexported fields
}

FieldEncryptionConfig configures per-event-type field encryption. It uses envelope encryption: a data encryption key (DEK) is generated per event, encrypted with the master key, and stored in metadata. Individual fields are encrypted locally with the DEK for performance.

Integrity model: each field's ciphertext is bound (via AES-GCM AAD) to its stream ID and field name, so it cannot be relocated to a different stream/field and still decrypt. Event metadata itself (correlation/causation/tenant IDs and the $encryption_* markers) is NOT authenticated by the library — tampering with it causes decryption to fail closed (never a plaintext leak), but guaranteeing metadata integrity is the storage layer's responsibility (the event store is append-only and access-controlled).

func NewFieldEncryptionConfig 

func NewFieldEncryptionConfig(opts ...EncryptionOption) *FieldEncryptionConfig

NewFieldEncryptionConfig creates a new FieldEncryptionConfig with the given options.

func (*FieldEncryptionConfig) HasEncryptedFields 

func (c *FieldEncryptionConfig) HasEncryptedFields(eventType string) bool

HasEncryptedFields reports whether any fields are configured for encryption for the given event type.

func (*FieldEncryptionConfig) IsRevoked added in v1.1.3 

func (c *FieldEncryptionConfig) IsRevoked(keyID string) (bool, error)

IsRevoked reports whether keyID is revoked via the configured provider, returning encryption.ErrRevocationUnsupported when revocation is unavailable.

func (*FieldEncryptionConfig) Provider added in v1.1.3 

func (c *FieldEncryptionConfig) Provider() encryption.Provider

Provider returns the configured encryption provider (may be nil).

func (*FieldEncryptionConfig) RevocationState added in v1.1.3 

func (c *FieldEncryptionConfig) RevocationState(keyID string) (encryption.RevocationState, error)

RevocationState reports keyID's fine-grained revocation state (NotRevoked / SoftRevoked / Revoked) via the configured provider. It lets erasure verification tell a still-recoverable soft-revocation from a permanent shred; providers without StatefulRevocable fall back to IsRevoked (so they never report SoftRevoked).

func (*FieldEncryptionConfig) RevokeKey added in v1.1.3 

func (c *FieldEncryptionConfig) RevokeKey(keyID string) error

RevokeKey crypto-shreds keyID via the configured provider, implementing the erasure side of GDPR (right to be forgotten). It returns encryption.ErrRevocationUnsupported when the provider does not implement encryption.Revocable, and is idempotent for providers that do.

type Filter 

type Filter struct {
	// Field is the field name to filter on.
	Field string

	// Op is the comparison operator.
	Op FilterOp

	// Value is the value to compare against.
	Value interface{}
}

Filter represents a query filter condition.

type FilterOp 

type FilterOp string

FilterOp represents a filter operation. 

const (
	// FilterOpEq matches equal values.
	FilterOpEq FilterOp = "="

	// FilterOpNe matches not equal values.
	FilterOpNe FilterOp = "!="

	// FilterOpGt matches greater than values.
	FilterOpGt FilterOp = ">"

	// FilterOpGte matches greater than or equal values.
	FilterOpGte FilterOp = ">="

	// FilterOpLt matches less than values.
	FilterOpLt FilterOp = "<"

	// FilterOpLte matches less than or equal values.
	FilterOpLte FilterOp = "<="

	// FilterOpIn matches any value in a list.
	FilterOpIn FilterOp = "IN"

	// FilterOpNotIn matches no value in a list.
	FilterOpNotIn FilterOp = "NOT IN"

	// FilterOpLike matches using SQL LIKE pattern.
	FilterOpLike FilterOp = "LIKE"

	// FilterOpIsNull matches null values.
	FilterOpIsNull FilterOp = "IS NULL"

	// FilterOpIsNotNull matches non-null values.
	FilterOpIsNotNull FilterOp = "IS NOT NULL"

	// FilterOpContains matches values that contain the given value.
	// On JSONB columns it tests containment (the stored array or object
	// contains the value); on text columns it performs a case-sensitive
	// substring match with the value treated literally.
	FilterOpContains FilterOp = "CONTAINS"

	// FilterOpBetween matches values between two bounds.
	FilterOpBetween FilterOp = "BETWEEN"
)

type GenericHandler 

type GenericHandler[C Command] struct {
	// contains filtered or unexported fields
}

GenericHandler is a type-safe command handler for a specific command type. Use this to create handlers with compile-time type checking.

func NewGenericHandler 

func NewGenericHandler[C Command](handler func(ctx context.Context, cmd C) (CommandResult, error)) *GenericHandler[C]

NewGenericHandler creates a new GenericHandler for the specified command type.

func (*GenericHandler[C]) CommandType 

func (h *GenericHandler[C]) CommandType() string

CommandType returns the command type this handler processes.

func (*GenericHandler[C]) Handle 

func (h *GenericHandler[C]) Handle(ctx context.Context, cmd Command) (CommandResult, error)

Handle processes the command with type checking.

type HandlerNotFoundError 

type HandlerNotFoundError struct {
	CommandType string
}

HandlerNotFoundError provides detailed information about a missing handler.

func NewHandlerNotFoundError 

func NewHandlerNotFoundError(cmdType string) *HandlerNotFoundError

NewHandlerNotFoundError creates a new HandlerNotFoundError.

func (*HandlerNotFoundError) Error 

func (e *HandlerNotFoundError) Error() string

Error returns the error message.

func (*HandlerNotFoundError) Is 

func (e *HandlerNotFoundError) Is(target error) bool

Is reports whether this error matches the target error.

func (*HandlerNotFoundError) Unwrap 

func (e *HandlerNotFoundError) Unwrap() error

Unwrap returns the underlying error for errors.Unwrap().

type HandlerRegistry 

type HandlerRegistry struct {
	// contains filtered or unexported fields
}

HandlerRegistry manages command handler registration and lookup.

func NewHandlerRegistry 

func NewHandlerRegistry() *HandlerRegistry

NewHandlerRegistry creates a new HandlerRegistry.

func (*HandlerRegistry) Clear 

func (r *HandlerRegistry) Clear()

Clear removes all handlers.

func (*HandlerRegistry) CommandTypes 

func (r *HandlerRegistry) CommandTypes() []string

CommandTypes returns all registered command types.

func (*HandlerRegistry) Count 

func (r *HandlerRegistry) Count() int

Count returns the number of registered handlers.

func (*HandlerRegistry) Get 

func (r *HandlerRegistry) Get(cmdType string) CommandHandler

Get returns the handler for a command type. Returns nil if no handler is registered.

func (*HandlerRegistry) Has 

func (r *HandlerRegistry) Has(cmdType string) bool

Has returns true if a handler is registered for the command type.

func (*HandlerRegistry) Register 

func (r *HandlerRegistry) Register(handler CommandHandler)

Register adds a handler for a command type. If a handler is already registered for this type, it will be replaced.

func (*HandlerRegistry) RegisterFunc 

func (r *HandlerRegistry) RegisterFunc(cmdType string, fn func(ctx context.Context, cmd Command) (CommandResult, error))

RegisterFunc registers a handler function for a command type.

func (*HandlerRegistry) Remove 

func (r *HandlerRegistry) Remove(cmdType string)

Remove removes a handler for a command type.

type IdempotencyConfig 

type IdempotencyConfig struct {
	// Store is the idempotency store to use.
	Store IdempotencyStore

	// TTL is how long to keep idempotency records.
	// Default is 24 hours.
	TTL time.Duration

	// KeyGenerator generates idempotency keys from commands.
	// If nil, GetIdempotencyKey is used.
	KeyGenerator func(Command) string

	// StoreErrors determines if failed commands should be stored.
	// If true, replaying a failed command returns the same error.
	// If false, failed commands can be retried.
	// Default is false.
	StoreErrors bool

	// FailClosed determines behavior when the idempotency store is unavailable.
	// If true, a store error fails the command (so a store outage cannot allow
	// duplicate processing). If false (the default), the command proceeds without
	// the idempotency guarantee (fail-open).
	FailClosed bool

	// ReservationTTL bounds how long an in-flight reservation blocks a key before
	// it self-expires. It should be longer than the slowest handler but much
	// shorter than TTL, so a process that crashes mid-handler does not block a
	// legitimate retry of the command for the full result TTL. Default: 5 minutes.
	ReservationTTL time.Duration

	// SkipCommands is a list of command types to skip idempotency checking.
	SkipCommands []string
}

IdempotencyConfig configures the idempotency middleware.

func DefaultIdempotencyConfig 

func DefaultIdempotencyConfig(store IdempotencyStore) IdempotencyConfig

DefaultIdempotencyConfig returns a default idempotency configuration.

type IdempotencyRecord 

type IdempotencyRecord = adapters.IdempotencyRecord

IdempotencyRecord stores information about a processed command.

func NewIdempotencyRecord 

func NewIdempotencyRecord(key, cmdType string, result CommandResult, ttl time.Duration) *IdempotencyRecord

NewIdempotencyRecord creates a new IdempotencyRecord from a CommandResult.

type IdempotencyReplayError 

type IdempotencyReplayError struct {
	Key     string
	Message string
}

IdempotencyReplayError indicates a command was already processed.

func (*IdempotencyReplayError) Error 

func (e *IdempotencyReplayError) Error() string

func (*IdempotencyReplayError) Is 

func (e *IdempotencyReplayError) Is(target error) bool

func (*IdempotencyReplayError) Unwrap 

func (e *IdempotencyReplayError) Unwrap() error

type IdempotencyStore 

type IdempotencyStore = adapters.IdempotencyStore

IdempotencyStore tracks processed commands to prevent duplicate processing.

type IdempotentCommand 

type IdempotentCommand interface {
	Command

	// IdempotencyKey returns a unique key for deduplication.
	// Commands with the same key will only be processed once.
	IdempotencyKey() string
}

IdempotentCommand is a command that supports idempotency.

type InMemoryRepository 

type InMemoryRepository[T any] struct {
	// contains filtered or unexported fields
}

InMemoryRepository provides an in-memory implementation of ReadModelRepository. Useful for testing and prototyping.

func NewInMemoryRepository 

func NewInMemoryRepository[T any](getID func(*T) string) *InMemoryRepository[T]

NewInMemoryRepository creates a new in-memory repository. The getID function extracts the ID from a read model.

func NewInMemoryRepositoryFor added in v1.0.25 

func NewInMemoryRepositoryFor[T any]() *InMemoryRepository[T]

NewInMemoryRepositoryFor creates a new in-memory repository for a read model whose pointer type implements ReadModelID. The ID is derived from the model's GetID method, so no explicit getID function is required. 

repo := NewInMemoryRepositoryFor[Account]() // *Account must implement ReadModelID

func (*InMemoryRepository[T]) Clear 

func (r *InMemoryRepository[T]) Clear(ctx context.Context) error

Clear removes all read models.

func (*InMemoryRepository[T]) Count 

func (r *InMemoryRepository[T]) Count(ctx context.Context, query Query) (int64, error)

Count returns the number of read models matching the query's filters.

func (*InMemoryRepository[T]) Delete 

func (r *InMemoryRepository[T]) Delete(ctx context.Context, id string) error

Delete removes a read model by ID.

func (*InMemoryRepository[T]) DeleteMany 

func (r *InMemoryRepository[T]) DeleteMany(ctx context.Context, query Query) (int64, error)

DeleteMany removes all read models matching the query's filters and returns the number deleted. With no filters it removes everything (equivalent to Clear).

func (*InMemoryRepository[T]) Exists 

func (r *InMemoryRepository[T]) Exists(ctx context.Context, id string) (bool, error)

Exists checks if a read model with the given ID exists.

func (*InMemoryRepository[T]) Find 

func (r *InMemoryRepository[T]) Find(ctx context.Context, query Query) ([]*T, error)

Find queries read models with the given criteria. Filters (AND-combined), ordering, offset, and limit are all applied. When no OrderBy is supplied the results are ordered by ID for deterministic pagination. Returned pointers are shallow copies of the stored models.

func (*InMemoryRepository[T]) FindOne 

func (r *InMemoryRepository[T]) FindOne(ctx context.Context, query Query) (*T, error)

FindOne returns the first read model matching the query.

func (*InMemoryRepository[T]) Get 

func (r *InMemoryRepository[T]) Get(ctx context.Context, id string) (*T, error)

Get retrieves a read model by ID. The returned pointer is a shallow copy of the stored model, so mutating it does not affect repository state (use Update for that).

func (*InMemoryRepository[T]) GetAll 

func (r *InMemoryRepository[T]) GetAll(ctx context.Context) ([]*T, error)

GetAll returns all read models in the repository. Returned pointers are shallow copies of the stored models.

func (*InMemoryRepository[T]) GetMany 

func (r *InMemoryRepository[T]) GetMany(ctx context.Context, ids []string) ([]*T, error)

GetMany retrieves multiple read models by their IDs. Returned pointers are shallow copies of the stored models.

func (*InMemoryRepository[T]) Insert 

func (r *InMemoryRepository[T]) Insert(ctx context.Context, model *T) error

Insert creates a new read model.

func (*InMemoryRepository[T]) Len 

func (r *InMemoryRepository[T]) Len() int

Len returns the number of items in the repository.

func (*InMemoryRepository[T]) Update 

func (r *InMemoryRepository[T]) Update(ctx context.Context, id string, updateFn func(*T)) error

Update modifies an existing read model.

func (*InMemoryRepository[T]) Upsert 

func (r *InMemoryRepository[T]) Upsert(ctx context.Context, model *T) error

Upsert creates or updates a read model.

type IncompatibleSchemaError 

type IncompatibleSchemaError struct {
	EventType     string
	OldVersion    int
	NewVersion    int
	Compatibility SchemaCompatibility
	Reason        string
}

IncompatibleSchemaError provides detailed information about a schema incompatibility.

func NewIncompatibleSchemaError 

func NewIncompatibleSchemaError(eventType string, oldVersion, newVersion int, compatibility SchemaCompatibility, reason string) *IncompatibleSchemaError

NewIncompatibleSchemaError creates a new IncompatibleSchemaError.

func (*IncompatibleSchemaError) Error 

func (e *IncompatibleSchemaError) Error() string

Error returns the error message.

func (*IncompatibleSchemaError) Is 

func (e *IncompatibleSchemaError) Is(target error) bool

Is reports whether this error matches the target error.

func (*IncompatibleSchemaError) Unwrap 

func (e *IncompatibleSchemaError) Unwrap() error

Unwrap returns the underlying error for errors.Unwrap().

type InlineProjection 

type InlineProjection interface {
	Projection

	// Apply processes a single event within the event store transaction.
	// The projection should update its read model based on the event.
	Apply(ctx context.Context, event StoredEvent) error
}

InlineProjection processes events in the same transaction as the event store append. This provides strong consistency but may impact write performance.

type JSONSerializer 

type JSONSerializer struct {
	// contains filtered or unexported fields
}

JSONSerializer is the default Serializer implementation using JSON encoding.

func NewJSONSerializer 

func NewJSONSerializer() *JSONSerializer

NewJSONSerializer creates a new JSONSerializer with an empty registry.

func NewJSONSerializerWithRegistry 

func NewJSONSerializerWithRegistry(registry *EventRegistry) *JSONSerializer

NewJSONSerializerWithRegistry creates a new JSONSerializer with the given registry.

func (*JSONSerializer) BinaryFormat added in v1.1.6 

func (s *JSONSerializer) BinaryFormat() bool

BinaryFormat reports that JSON output is textual, not binary. It satisfies BinaryFormatReporter so JSON-backed adapters accept this serializer.

func (*JSONSerializer) Deserialize 

func (s *JSONSerializer) Deserialize(data []byte, eventType string) (interface{}, error)

Deserialize converts JSON bytes back to an event. If the event type is registered, returns a value of that type. Otherwise, returns a map[string]interface{}.

func (*JSONSerializer) IsRegistered added in v1.1.6 

func (s *JSONSerializer) IsRegistered(eventType string) bool

IsRegistered reports whether eventType resolves to a concrete registered Go type (rather than the map[string]interface{} fallback Deserialize returns for unknown types).

func (*JSONSerializer) Register 

func (s *JSONSerializer) Register(eventType string, example interface{})

Register adds an event type to the serializer's registry.

func (*JSONSerializer) RegisterAll 

func (s *JSONSerializer) RegisterAll(examples ...interface{})

RegisterAll registers multiple events using their struct names as type names.

func (*JSONSerializer) RegisteredEventTypes added in v1.1.6 

func (s *JSONSerializer) RegisteredEventTypes() []string

RegisteredEventTypes returns the names of all registered event types.

func (*JSONSerializer) Registry 

func (s *JSONSerializer) Registry() *EventRegistry

Registry returns the underlying EventRegistry.

func (*JSONSerializer) Serialize 

func (s *JSONSerializer) Serialize(event interface{}) ([]byte, error)

Serialize converts an event to JSON bytes.

type KeyNotFoundError 

type KeyNotFoundError = encryption.KeyNotFoundError

KeyNotFoundError provides detailed information about a missing encryption key.

type KeyRevokedError 

type KeyRevokedError = encryption.KeyRevokedError

KeyRevokedError provides detailed information about a revoked encryption key.

type LiveOptions 

type LiveOptions struct {
	// BufferSize is the size of the event channel buffer.
	// Default: 1000
	BufferSize int
}

LiveOptions configures live projection behavior.

func DefaultLiveOptions 

func DefaultLiveOptions() LiveOptions

DefaultLiveOptions returns the default live projection options.

type LiveProjection 

type LiveProjection interface {
	Projection

	// OnEvent is called for each event in real-time.
	// This method should not block for long periods.
	OnEvent(ctx context.Context, event StoredEvent)

	// IsTransient returns true if this projection doesn't persist state.
	// Transient projections are not checkpointed.
	IsTransient() bool
}

LiveProjection receives events in real-time for dashboards and notifications. These projections are transient and don't persist state.

type LiveProjectionBase 

type LiveProjectionBase struct {
	ProjectionBase
	// contains filtered or unexported fields
}

LiveProjectionBase provides a default implementation of LiveProjection. Embed this struct and override OnEvent to create live projections.

func NewLiveProjectionBase 

func NewLiveProjectionBase(name string, transient bool, handledEvents ...string) LiveProjectionBase

NewLiveProjectionBase creates a new LiveProjectionBase.

func (*LiveProjectionBase) IsTransient 

func (p *LiveProjectionBase) IsTransient() bool

IsTransient returns whether this projection is transient.

type Logger 

type Logger interface {
	Debug(msg string, args ...interface{})
	Info(msg string, args ...interface{})
	Warn(msg string, args ...interface{})
	Error(msg string, args ...interface{})
}

Logger defines the logging interface for the event store.

type LoggingMiddleware 

type LoggingMiddleware struct {
	// contains filtered or unexported fields
}

LoggingMiddleware logs command execution.

func NewLoggingMiddleware 

func NewLoggingMiddleware(logger Logger) *LoggingMiddleware

NewLoggingMiddleware creates a new LoggingMiddleware.

func (*LoggingMiddleware) Middleware 

func (m *LoggingMiddleware) Middleware() Middleware

Middleware returns the middleware function.

type MemorySubjectIndex added in v1.1.3 

type MemorySubjectIndex struct {
	// contains filtered or unexported fields
}

MemorySubjectIndex is a thread-safe in-memory subject index implementing both SubjectIndexAdapter (read) and SubjectIndexWriter (write). It maps each subject to the set of streams that contain its events, so a resolver can find a subject's footprint without scanning the whole store. Populate it at append time (WithSubjectIndexWriter) and/or via BackfillSubjectIndex; it is NOT persistent — use a durable index (e.g. a table-backed one) in production.

func NewMemorySubjectIndex added in v1.1.3 

func NewMemorySubjectIndex() *MemorySubjectIndex

NewMemorySubjectIndex creates an empty in-memory subject index.

func (*MemorySubjectIndex) IndexSubjects added in v1.1.3 

func (m *MemorySubjectIndex) IndexSubjects(ctx context.Context, streamID string, subjectIDs []string) error

IndexSubjects records that streamID contains events for each subject. Idempotent.

func (*MemorySubjectIndex) StreamsBySubject added in v1.1.3 

func (m *MemorySubjectIndex) StreamsBySubject(ctx context.Context, subjectID string) ([]string, error)

StreamsBySubject returns the sorted streams recorded for subjectID (empty if none).

type Metadata 

type Metadata struct {
	// CorrelationID links related events across services for distributed tracing.
	CorrelationID string `json:"correlationId,omitempty"`

	// CausationID identifies the event or command that caused this event.
	CausationID string `json:"causationId,omitempty"`

	// UserID identifies the user who triggered this event.
	UserID string `json:"userId,omitempty"`

	// TenantID identifies the tenant for multi-tenant applications.
	TenantID string `json:"tenantId,omitempty"`

	// Custom contains arbitrary key-value pairs for application-specific metadata.
	Custom map[string]string `json:"custom,omitempty"`
}

Metadata contains contextual information about an event. It supports distributed tracing, multi-tenancy, and custom key-value pairs.

func SetSchemaVersion 

func SetSchemaVersion(m Metadata, version int) Metadata

SetSchemaVersion returns a copy of Metadata with the schema version set.

func (Metadata) IsEmpty 

func (m Metadata) IsEmpty() bool

IsEmpty reports whether the Metadata has no values set.

func (Metadata) WithCausationID 

func (m Metadata) WithCausationID(id string) Metadata

WithCausationID returns a copy of Metadata with the causation ID set.

func (Metadata) WithCorrelationID 

func (m Metadata) WithCorrelationID(id string) Metadata

WithCorrelationID returns a copy of Metadata with the correlation ID set.

func (Metadata) WithCustom 

func (m Metadata) WithCustom(key, value string) Metadata

WithCustom returns a copy of Metadata with a custom key-value pair added.

func (Metadata) WithTenantID 

func (m Metadata) WithTenantID(id string) Metadata

WithTenantID returns a copy of Metadata with the tenant ID set.

func (Metadata) WithUserID 

func (m Metadata) WithUserID(id string) Metadata

WithUserID returns a copy of Metadata with the user ID set.

type MetricsCollector 

type MetricsCollector interface {
	// RecordCommand records a command execution.
	RecordCommand(cmdType string, duration time.Duration, success bool, err error)
}

MetricsMiddleware collects metrics about command execution.

type Middleware 

type Middleware func(next MiddlewareFunc) MiddlewareFunc

Middleware wraps a handler function with additional functionality.

func AuditMiddleware added in v1.1.0 

func AuditMiddleware(config AuditConfig) Middleware

AuditMiddleware creates middleware that writes an immutable audit entry for every dispatched command. Both successful and failed executions are audited.

The audit write happens after the command runs. By default the middleware is fail-open: if the store write fails, the failure is ignored and the original command result is returned. Set FailClosed to surface the audit write failure instead — note that the command's side effect has already happened, so this is not a transactional guarantee.

This middleware does not recover panics itself. To audit a handler that panics, place RecoveryMiddleware *inside* this one (i.e. closer to the handler) so the panic is converted to an error result before the audit entry is written, e.g.: 

mink.ChainMiddleware(mink.AuditMiddleware(cfg), mink.RecoveryMiddleware())

func CausationIDMiddleware 

func CausationIDMiddleware() Middleware

CausationIDMiddleware creates middleware that propagates causation IDs. The causation ID links events/commands to the command that caused them. This is essential for tracking the chain of causality in event sourcing.

func ChainMiddleware 

func ChainMiddleware(middleware ...Middleware) Middleware

ChainMiddleware creates a single middleware from multiple middleware.

func CommandTypeMiddleware 

func CommandTypeMiddleware(types []string, middleware Middleware) Middleware

CommandTypeMiddleware applies middleware only for specific command types.

func ConditionalMiddleware 

func ConditionalMiddleware(condition func(Command) bool, middleware Middleware) Middleware

ConditionalMiddleware applies middleware only if the condition is true.

func CorrelationIDMiddleware 

func CorrelationIDMiddleware(generator func() string) Middleware

CorrelationIDMiddleware creates middleware that propagates correlation IDs.

func IdempotencyMiddleware 

func IdempotencyMiddleware(config IdempotencyConfig) Middleware

IdempotencyMiddleware creates middleware that prevents duplicate command processing.

func MetricsMiddleware 

func MetricsMiddleware(collector MetricsCollector) Middleware

MetricsMiddleware creates middleware that records metrics.

func RecoveryMiddleware 

func RecoveryMiddleware() Middleware

RecoveryMiddleware recovers from panics in handlers and returns them as errors. It captures a sanitized representation of the command data for debugging.

func RetryMiddleware 

func RetryMiddleware(config RetryConfig) Middleware

RetryMiddleware creates middleware that retries failed commands.

func TenantMiddleware 

func TenantMiddleware(extractor func(Command) string, required bool) Middleware

TenantMiddleware extracts and validates tenant ID.

func TimeoutMiddleware 

func TimeoutMiddleware(timeout time.Duration) Middleware

TimeoutMiddleware adds a timeout to command execution.

func ValidationMiddleware 

func ValidationMiddleware() Middleware

ValidationMiddleware validates commands before they reach the handler. If validation fails, the command is not dispatched.

type MiddlewareFunc 

type MiddlewareFunc func(ctx context.Context, cmd Command) (CommandResult, error)

MiddlewareFunc is the function signature for command middleware.

type MultiValidationError 

type MultiValidationError struct {
	// CommandType is the type of command that failed validation.
	CommandType string

	// Errors contains all validation errors.
	Errors []*ValidationError
}

MultiValidationError contains multiple validation errors.

func NewMultiValidationError 

func NewMultiValidationError(cmdType string) *MultiValidationError

NewMultiValidationError creates a new MultiValidationError.

func (*MultiValidationError) Add 

func (e *MultiValidationError) Add(err *ValidationError)

Add adds a validation error.

func (*MultiValidationError) AddField 

func (e *MultiValidationError) AddField(field, message string)

AddField adds a validation error for a specific field.

func (*MultiValidationError) Error 

func (e *MultiValidationError) Error() string

Error returns the error message.

func (*MultiValidationError) HasErrors 

func (e *MultiValidationError) HasErrors() bool

HasErrors returns true if there are any validation errors.

func (*MultiValidationError) Is 

func (e *MultiValidationError) Is(target error) bool

Is reports whether this error matches the target error.

func (*MultiValidationError) Unwrap 

func (e *MultiValidationError) Unwrap() error

Unwrap returns the first error for errors.Unwrap().

type NullColumnError added in v1.1.11 

type NullColumnError struct {
	Column string // SQL column that held NULL
	Field  string // destination struct field (empty if unresolved)
	GoType string // Go type of the field, e.g. "string" (empty if unresolved)
	Err    error  // underlying driver error
}

NullColumnError reports that a stored SQL NULL was read into a non-pointer scalar struct field whose column is NOT declared `nullable`. A column tagged `mink:"...,nullable"` reads a NULL as the field's Go zero value; this error is the actionable replacement for the driver's opaque "converting NULL to <type> is unsupported" when the tag is missing. Resolve it by tagging the field `nullable` (NULL -> zero value) or making it a pointer (NULL -> nil). errors.Is(err, ErrNullColumn) matches it; the underlying driver error is retrievable via errors.Unwrap.

func (*NullColumnError) Error added in v1.1.11 

func (e *NullColumnError) Error() string

func (*NullColumnError) Is added in v1.1.11 

func (e *NullColumnError) Is(target error) bool

Is reports a match against the ErrNullColumn sentinel, so callers can write errors.Is(err, mink.ErrNullColumn) as well as errors.As for the details.

func (*NullColumnError) Unwrap added in v1.1.11 

func (e *NullColumnError) Unwrap() error

Unwrap returns the underlying driver error so errors.Is/As can reach it.

type Option 

type Option func(*EventStore)

Option configures an EventStore.

func WithAutoRegisterOnAppend added in v1.1.6 

func WithAutoRegisterOnAppend() Option

WithAutoRegisterOnAppend registers each event's concrete type with the serializer the first time it is appended/saved in this process, so a save-then-load round-trip in the same process does not require a separate RegisterEvents call. In-process only: a cold process that only loads historical events still needs explicit registration. Off by default to keep registration explicit and discoverable.

func WithFieldEncryption 

func WithFieldEncryption(config *FieldEncryptionConfig) Option

WithFieldEncryption configures the event store with field-level encryption. When set, configured fields are automatically encrypted during appending and decrypted during loading. Uses envelope encryption for performance.

func WithLogger 

func WithLogger(l Logger) Option

WithLogger sets a custom logger.

func WithMaxEventSize added in v1.0.25 

func WithMaxEventSize(maxBytes int) Option

WithMaxEventSize sets the maximum serialized size, in bytes, of a single event's data. Appending an event whose serialized (and, if configured, encrypted) data exceeds this returns ErrEventTooLarge. The default (0) imposes no limit; set it to bound memory/IO from oversized or malicious payloads.

func WithSerializer 

func WithSerializer(s Serializer) Option

WithSerializer sets a custom serializer.

func WithStrictReplay added in v1.1.6 

func WithStrictReplay() Option

WithStrictReplay makes LoadAggregate return an *UnregisteredEventTypeError when it encounters an event whose type is not registered (which would otherwise deserialize to the map fallback and be silently dropped from the rebuilt aggregate state). The default is lenient — such an event is logged (WARN, once per stream/type/version) and skipped, preserving existing v1.x behavior; use this to fail fast in dev/CI. Zero overhead unset.

func WithSubjectIndexWriter added in v1.1.3 

func WithSubjectIndexWriter(w SubjectIndexWriter) Option

WithSubjectIndexWriter records, at append time, which subject(s) each stream touches into the given index (derived from the events' subject tags — pair with WithSubjectTagger). Writes are best-effort: an index-write failure is logged, never failing the append. Combine with a SubjectResolver using WithResolverIndex to make discovery and erasure O(a subject's events) instead of a full scan. Zero overhead when unset.

func WithSubjectTagger added in v1.1.3 

func WithSubjectTagger(tagger SubjectTagger) Option

WithSubjectTagger configures a tagger that records, at append time, which data subject(s) each event concerns (in Metadata.Custom). It enables a SubjectResolver to later enumerate a subject's complete cross-stream footprint for GDPR export and erasure. Zero overhead when unset.

func WithUpcasters 

func WithUpcasters(chain *UpcasterChain) Option

WithUpcasters configures the event store with an upcaster chain for transparent schema evolution. When set, events are automatically upcasted to the latest schema version during loading and stamped with the latest version during appending.

type OrderBy 

type OrderBy struct {
	// Field is the field name to sort by.
	Field string

	// Desc specifies descending order.
	Desc bool
}

OrderBy represents a sort order.

type OutboxMessage 

type OutboxMessage = adapters.OutboxMessage

OutboxMessage represents a message in the transactional outbox.

type OutboxMetrics 

type OutboxMetrics interface {
	RecordMessageProcessed(destination string, success bool)
	RecordMessageFailed(destination string)
	RecordMessageDeadLettered()
	RecordBatchDuration(duration time.Duration)
	RecordPendingMessages(count int64)
}

OutboxMetrics collects metrics about outbox processing.

type OutboxOption 

type OutboxOption func(*EventStoreWithOutbox)

OutboxOption configures an EventStoreWithOutbox.

func WithOutboxLogger 

func WithOutboxLogger(l Logger) OutboxOption

WithOutboxLogger sets a logger for the outbox wrapper.

func WithOutboxMaxAttempts 

func WithOutboxMaxAttempts(n int) OutboxOption

WithOutboxMaxAttempts sets the default max attempts for outbox messages.

type OutboxProcessor 

type OutboxProcessor struct {
	// contains filtered or unexported fields
}

OutboxProcessor polls the outbox store for pending messages and publishes them via registered publishers. It handles retries, dead-lettering, and cleanup.

func NewOutboxProcessor 

func NewOutboxProcessor(store OutboxStore, opts ...ProcessorOption) *OutboxProcessor

NewOutboxProcessor creates a new OutboxProcessor.

func (*OutboxProcessor) IsRunning 

func (p *OutboxProcessor) IsRunning() bool

IsRunning returns true if the processor is running.

func (*OutboxProcessor) Start 

func (p *OutboxProcessor) Start(ctx context.Context) error

Start begins the background processing loop.

func (*OutboxProcessor) Stop 

func (p *OutboxProcessor) Stop(ctx context.Context) error

Stop gracefully stops the processor, draining in-flight work.

type OutboxRoute 

type OutboxRoute struct {
	// EventTypes is the list of event types this route matches. Empty matches all.
	EventTypes []string

	// Destination is the target (e.g., "webhook:https://example.com/events", "kafka:orders").
	Destination string

	// Transform optionally transforms the event payload before outbox scheduling.
	// Note: the event parameter is always nil because the outbox operates on raw serialized data.
	// Use the stored parameter to access stream ID, event type, raw data, and metadata.
	Transform func(event interface{}, stored StoredEvent) ([]byte, error)

	// Filter optionally filters events. Return true to include the event.
	// Note: the event parameter is always nil because the outbox operates on raw serialized data.
	// Use the stored parameter to access stream ID, event type, raw data, and metadata.
	Filter func(event interface{}, stored StoredEvent) bool
}

OutboxRoute defines routing rules for outbox messages.

type OutboxStatus 

type OutboxStatus = adapters.OutboxStatus

OutboxStatus represents the current status of an outbox message.

type OutboxStore 

type OutboxStore = adapters.OutboxStore

OutboxStore defines the interface for outbox message persistence.

type PanicError 

type PanicError struct {
	CommandType string
	Value       interface{}
	Stack       string
	// CommandData contains a sanitized JSON representation of the command for debugging.
	// Sensitive fields should be masked by the caller before setting this field.
	CommandData string
}

PanicError provides detailed information about a handler panic.

func NewPanicError 

func NewPanicError(cmdType string, value interface{}, stack string) *PanicError

NewPanicError creates a new PanicError.

func NewPanicErrorWithCommand 

func NewPanicErrorWithCommand(cmdType string, value interface{}, stack string, commandData string) *PanicError

NewPanicErrorWithCommand creates a new PanicError with command data for debugging. The commandData should be a sanitized representation of the command (sensitive fields masked).

func (*PanicError) Error 

func (e *PanicError) Error() string

Error returns the error message.

func (*PanicError) Is 

func (e *PanicError) Is(target error) bool

Is reports whether this error matches the target error.

func (*PanicError) Unwrap 

func (e *PanicError) Unwrap() error

Unwrap returns the underlying error for errors.Unwrap().

type ParallelRebuilder 

type ParallelRebuilder struct {
	// contains filtered or unexported fields
}

ParallelRebuilder rebuilds multiple projections in parallel.

func NewParallelRebuilder 

func NewParallelRebuilder(rebuilder *ProjectionRebuilder, concurrency int) *ParallelRebuilder

NewParallelRebuilder creates a new parallel rebuilder.

func (*ParallelRebuilder) RebuildAll 

func (pr *ParallelRebuilder) RebuildAll(ctx context.Context, projections []AsyncProjection, opts ...RebuildOptions) error

RebuildAll rebuilds multiple async projections in parallel.

type PollingSubscription 

type PollingSubscription struct {
	// contains filtered or unexported fields
}

PollingSubscription polls the event store for new events. This is a fallback when push-based subscriptions aren't available.

func NewPollingSubscription 

func NewPollingSubscription(
	store *EventStore,
	fromPosition uint64,
	opts ...SubscriptionOptions,
) *PollingSubscription

NewPollingSubscription creates a new polling subscription.

func (*PollingSubscription) Close 

func (s *PollingSubscription) Close() error

Close stops the subscription.

func (*PollingSubscription) Err 

func (s *PollingSubscription) Err() error

Err returns any error that caused the subscription to close.

func (*PollingSubscription) Events 

func (s *PollingSubscription) Events() <-chan StoredEvent

Events returns the channel for receiving events.

func (*PollingSubscription) Start 

func (s *PollingSubscription) Start(ctx context.Context, pollInterval time.Duration)

Start begins polling for events.

type ProcessorOption 

type ProcessorOption func(*OutboxProcessor)

ProcessorOption configures an OutboxProcessor.

func WithBatchSize 

func WithBatchSize(n int) ProcessorOption

WithBatchSize sets the maximum number of messages to process in a single batch.

func WithCleanupAge 

func WithCleanupAge(d time.Duration) ProcessorOption

WithCleanupAge sets the age threshold for cleaning up completed messages.

func WithCleanupInterval 

func WithCleanupInterval(d time.Duration) ProcessorOption

WithCleanupInterval sets how often completed messages are cleaned up.

func WithMaxRetries 

func WithMaxRetries(n int) ProcessorOption

WithMaxRetries sets the maximum number of delivery attempts.

func WithOutboxMetrics 

func WithOutboxMetrics(metrics OutboxMetrics) ProcessorOption

WithOutboxMetrics sets the metrics collector for the processor.

func WithPollInterval 

func WithPollInterval(d time.Duration) ProcessorOption

WithPollInterval sets how often the processor polls for pending messages.

func WithProcessingTimeout added in v1.0.25 

func WithProcessingTimeout(d time.Duration) ProcessorOption

WithProcessingTimeout sets how long a message may remain claimed (in OutboxProcessing) before the maintenance sweep reclaims it back to pending. This recovers messages orphaned by a processor crash. Default: 5 minutes.

func WithProcessorLogger 

func WithProcessorLogger(logger Logger) ProcessorOption

WithProcessorLogger sets the logger for the processor.

func WithPublisher 

func WithPublisher(publisher Publisher) ProcessorOption

WithPublisher registers a publisher for a given destination prefix.

func WithRetryBackoff 

func WithRetryBackoff(d time.Duration) ProcessorOption

WithRetryBackoff sets the duration between retry cycles.

type ProgressCallback 

type ProgressCallback func(progress RebuildProgress)

ProgressCallback is called periodically during rebuild with progress updates.

type Projection 

type Projection interface {
	// Name returns the unique identifier for this projection.
	// This name is used for checkpointing and management.
	Name() string

	// HandledEvents returns the list of event types this projection handles.
	// An empty list means the projection handles all event types.
	HandledEvents() []string
}

Projection is the base interface for all projection types. Projections transform events into optimized read models.

type ProjectionBase 

type ProjectionBase struct {
	// contains filtered or unexported fields
}

ProjectionBase provides a default partial implementation of Projection. Embed this struct in your projection types to get common functionality.

func NewProjectionBase 

func NewProjectionBase(name string, handledEvents ...string) ProjectionBase

NewProjectionBase creates a new ProjectionBase.

func (*ProjectionBase) HandledEvents 

func (p *ProjectionBase) HandledEvents() []string

HandledEvents returns the list of event types this projection handles.

func (*ProjectionBase) HandlesEvent 

func (p *ProjectionBase) HandlesEvent(eventType string) bool

HandlesEvent returns true if this projection handles the given event type.

func (*ProjectionBase) Name 

func (p *ProjectionBase) Name() string

Name returns the projection name.

type ProjectionEngine 

type ProjectionEngine struct {
	// contains filtered or unexported fields
}

ProjectionEngine manages the lifecycle of projections. It handles registration, starting, stopping, and monitoring projections.

func NewProjectionEngine 

func NewProjectionEngine(store *EventStore, opts ...ProjectionEngineOption) *ProjectionEngine

NewProjectionEngine creates a new ProjectionEngine.

func (*ProjectionEngine) GetAllStatuses 

func (e *ProjectionEngine) GetAllStatuses() []*ProjectionStatus

GetAllStatuses returns the status of all registered projections.

func (*ProjectionEngine) GetStatus 

func (e *ProjectionEngine) GetStatus(name string) (*ProjectionStatus, error)

GetStatus returns the status of a projection by name.

func (*ProjectionEngine) IsRunning 

func (e *ProjectionEngine) IsRunning() bool

IsRunning returns true if the engine is running.

func (*ProjectionEngine) NotifyLiveProjections 

func (e *ProjectionEngine) NotifyLiveProjections(ctx context.Context, events []StoredEvent)

NotifyLiveProjections notifies all live projections of new events.

func (*ProjectionEngine) Pause added in v1.0.25 

func (e *ProjectionEngine) Pause(name string) error

Pause pauses a registered async or live projection. A paused async worker stops processing new events (entering ProjectionStatePaused) but stays alive; a paused live projection stops receiving notifications. Returns ErrProjectionNotFound if no projection with the given name is registered.

func (*ProjectionEngine) ProcessInlineProjections 

func (e *ProjectionEngine) ProcessInlineProjections(ctx context.Context, events []StoredEvent) error

ProcessInlineProjections processes all inline projections for the given events. This is called by the event store after appending events.

func (*ProjectionEngine) Rebuild added in v1.0.25 

func (e *ProjectionEngine) Rebuild(ctx context.Context, name string, opts ...RebuildOptions) error

Rebuild rebuilds a registered async projection from scratch. The worker is paused and marked ProjectionStateRebuilding for the duration; on completion it resumes from the rebuilt checkpoint. For a consistent rebuild the projection should be quiescent (the engine stopped, or the projection already paused). Returns ErrProjectionNotFound if the async projection is not registered.

func (*ProjectionEngine) RegisterAsync 

func (e *ProjectionEngine) RegisterAsync(projection AsyncProjection, opts ...AsyncOptions) error

RegisterAsync registers an async projection with the given options. Async projections are processed in background workers.

func (*ProjectionEngine) RegisterInline 

func (e *ProjectionEngine) RegisterInline(projection InlineProjection) error

RegisterInline registers an inline projection. Inline projections are processed synchronously with event appends.

func (*ProjectionEngine) RegisterLive 

func (e *ProjectionEngine) RegisterLive(projection LiveProjection, opts ...LiveOptions) error

RegisterLive registers a live projection with optional configuration. Live projections receive events in real-time.

func (*ProjectionEngine) Restart added in v1.1.13 

func (e *ProjectionEngine) Restart(ctx context.Context, name string) error

Restart relaunches a Faulted async projection worker, resuming strictly from its persisted checkpoint (never reprocessing from position 0, regardless of StartFromBeginning). It is the operator counterpart to a configured RestartPolicy and is symmetric with Pause/Resume/Rebuild. It is idempotent: a no-op returning nil when the named worker is not Faulted (or when the engine is not running), and it returns ErrProjectionNotFound for an unknown name. Concurrent Restart calls — and a race with an automatic policy-driven restart — relaunch at most one worker goroutine.

func (*ProjectionEngine) Resume added in v1.0.25 

func (e *ProjectionEngine) Resume(name string) error

Resume resumes a previously paused async or live projection. Returns ErrProjectionNotFound if no projection with the given name is registered.

func (*ProjectionEngine) Start 

func (e *ProjectionEngine) Start(ctx context.Context) error

Start starts the projection engine and all registered projections.

func (*ProjectionEngine) Stop 

func (e *ProjectionEngine) Stop(ctx context.Context) error

Stop gracefully stops the projection engine.

func (*ProjectionEngine) Unregister 

func (e *ProjectionEngine) Unregister(name string) error

Unregister removes a projection by name.

type ProjectionEngineOption 

type ProjectionEngineOption func(*ProjectionEngine)

ProjectionEngineOption configures a ProjectionEngine.

func WithCheckpointStore 

func WithCheckpointStore(store CheckpointStore) ProjectionEngineOption

WithCheckpointStore sets the checkpoint store for the engine.

func WithProjectionLogger 

func WithProjectionLogger(logger Logger) ProjectionEngineOption

WithProjectionLogger sets the logger for the engine.

func WithProjectionMetrics 

func WithProjectionMetrics(metrics ProjectionMetrics) ProjectionEngineOption

WithProjectionMetrics sets the metrics collector for the engine.

func WithProjectionStateObserver added in v1.1.13 

func WithProjectionStateObserver(fn func(name string, oldState, newState ProjectionState, err error)) ProjectionEngineOption

WithProjectionStateObserver registers a callback invoked on every async/live projection state transition (oldState → newState), carrying the fault error when a worker enters ProjectionStateFaulted (nil otherwise). Use it to push an alert on Faulted / Restarting and on recovery instead of polling GetStatus. The callback runs outside the worker's state lock, so it may safely call back into the engine (e.g. GetStatus) without deadlock or reentrancy; it should not block for long, as it runs on the worker's goroutine. nil (the default) means no callback and zero overhead. This does not change or extend the ProjectionMetrics interface.

type ProjectionError 

type ProjectionError struct {
	ProjectionName string
	EventType      string
	Position       uint64
	Cause          error
}

ProjectionError provides detailed information about a projection failure.

func NewProjectionError 

func NewProjectionError(projectionName, eventType string, position uint64, cause error) *ProjectionError

NewProjectionError creates a new ProjectionError.

func (*ProjectionError) Error 

func (e *ProjectionError) Error() string

Error returns the error message.

func (*ProjectionError) Is 

func (e *ProjectionError) Is(target error) bool

Is reports whether this error matches the target error.

func (*ProjectionError) Unwrap 

func (e *ProjectionError) Unwrap() error

Unwrap returns the underlying cause for errors.Unwrap().

type ProjectionMetrics 

type ProjectionMetrics interface {
	// RecordEventProcessed records that an event was processed.
	RecordEventProcessed(projectionName, eventType string, duration time.Duration, success bool)

	// RecordBatchProcessed records that a batch of events was processed.
	RecordBatchProcessed(projectionName string, count int, duration time.Duration, success bool)

	// RecordCheckpoint records a checkpoint update.
	RecordCheckpoint(projectionName string, position uint64)

	// RecordError records a projection error.
	RecordError(projectionName string, err error)
}

ProjectionMetrics collects metrics about projection processing.

type ProjectionRebuilder 

type ProjectionRebuilder struct {
	// contains filtered or unexported fields
}

ProjectionRebuilder rebuilds projections from scratch. It replays all events through a projection to reconstruct its read model.

func NewProjectionRebuilder 

func NewProjectionRebuilder(store *EventStore, checkpointStore CheckpointStore, opts ...ProjectionRebuilderOption) *ProjectionRebuilder

NewProjectionRebuilder creates a new projection rebuilder.

func (*ProjectionRebuilder) RebuildAsync 

func (r *ProjectionRebuilder) RebuildAsync(ctx context.Context, projection AsyncProjection, opts ...RebuildOptions) error

RebuildAsync rebuilds an async projection from scratch.

func (*ProjectionRebuilder) RebuildInline 

func (r *ProjectionRebuilder) RebuildInline(ctx context.Context, projection InlineProjection, opts ...RebuildOptions) error

RebuildInline rebuilds an inline projection from scratch.

type ProjectionRebuilderOption 

type ProjectionRebuilderOption func(*ProjectionRebuilder)

ProjectionRebuilderOption configures a ProjectionRebuilder.

func WithRebuilderBatchSize 

func WithRebuilderBatchSize(size int) ProjectionRebuilderOption

WithRebuilderBatchSize sets the batch size for rebuilding.

func WithRebuilderLogger 

func WithRebuilderLogger(logger Logger) ProjectionRebuilderOption

WithRebuilderLogger sets the logger for the rebuilder.

func WithRebuilderMetrics 

func WithRebuilderMetrics(metrics ProjectionMetrics) ProjectionRebuilderOption

WithRebuilderMetrics sets the metrics collector for the rebuilder.

type ProjectionState 

type ProjectionState string

ProjectionState represents the current state of a projection. 

const (
	// ProjectionStateStopped indicates the projection is not running.
	ProjectionStateStopped ProjectionState = "stopped"

	// ProjectionStateRunning indicates the projection is actively processing events.
	ProjectionStateRunning ProjectionState = "running"

	// ProjectionStatePaused indicates the projection is paused.
	ProjectionStatePaused ProjectionState = "paused"

	// ProjectionStateFaulted indicates the projection has encountered an error.
	ProjectionStateFaulted ProjectionState = "faulted"

	// ProjectionStateRebuilding indicates the projection is being rebuilt.
	ProjectionStateRebuilding ProjectionState = "rebuilding"

	// ProjectionStateCatchingUp indicates the projection is catching up to current events.
	ProjectionStateCatchingUp ProjectionState = "catching_up"

	// ProjectionStateRestarting indicates a Faulted async worker is waiting for its
	// configured RestartPolicy to relaunch it (see AsyncOptions.RestartPolicy). It marks
	// the backoff window between a fault and a policy-driven restart. Additive; the
	// existing states are unchanged, and a projection without a RestartPolicy never
	// enters this state.
	ProjectionStateRestarting ProjectionState = "restarting"
)

type ProjectionStatus 

type ProjectionStatus struct {
	// Name is the projection name.
	Name string

	// State is the current state of the projection.
	State ProjectionState

	// LastPosition is the global position of the last processed event.
	LastPosition uint64

	// EventsProcessed is the total number of events processed.
	EventsProcessed uint64

	// LastProcessedAt is when the last event was processed.
	LastProcessedAt time.Time

	// Error contains the error message if the projection is faulted.
	Error string

	// Lag is the number of events behind the head of the event store.
	Lag uint64

	// AverageLatency is the average time to process an event.
	AverageLatency time.Duration
}

ProjectionStatus provides detailed information about a projection's current state.

type Publisher 

type Publisher interface {
	// Publish sends one or more messages to the external system.
	Publish(ctx context.Context, messages []*OutboxMessage) error

	// Destination returns the destination prefix this publisher handles (e.g., "webhook", "kafka", "sns").
	Destination() string
}

Publisher publishes outbox messages to an external system.

type Query 

type Query struct {
	// Filters to apply.
	Filters []Filter

	// Ordering criteria.
	OrderBy []OrderBy

	// Maximum number of results to return.
	// 0 means no limit.
	Limit int

	// Number of results to skip.
	Offset int

	// IncludeCount includes the total count in paginated results.
	IncludeCount bool
}

Query represents a query for read models.

func NewQuery 

func NewQuery() *Query

NewQuery creates a new empty Query.

func (*Query) And 

func (q *Query) And(field string, op FilterOp, value interface{}) *Query

And is an alias for Where for readability.

func (*Query) Build 

func (q *Query) Build() Query

Build returns a copy of the query (useful for chaining).

func (*Query) OrderByAsc 

func (q *Query) OrderByAsc(field string) *Query

OrderByAsc adds ascending order.

func (*Query) OrderByDesc 

func (q *Query) OrderByDesc(field string) *Query

OrderByDesc adds descending order.

func (*Query) Where 

func (q *Query) Where(field string, op FilterOp, value interface{}) *Query

Where adds a filter condition.

func (*Query) WithCount 

func (q *Query) WithCount() *Query

WithCount includes total count in results.

func (*Query) WithLimit 

func (q *Query) WithLimit(limit int) *Query

WithLimit sets the maximum number of results.

func (*Query) WithOffset 

func (q *Query) WithOffset(offset int) *Query

WithOffset sets the number of results to skip.

func (*Query) WithPagination 

func (q *Query) WithPagination(page, pageSize int) *Query

WithPagination sets limit and offset for pagination.

type QueryResult 

type QueryResult[T any] struct {
	// Items contains the matching read models.
	Items []*T

	// TotalCount is the total number of matching items (before pagination).
	// Only populated if IncludeCount was true in the query.
	TotalCount int64

	// HasMore indicates if there are more results beyond the limit.
	HasMore bool
}

QueryResult contains query results with optional count.

type ReadModelID 

type ReadModelID interface {
	// GetID returns the read model's unique identifier.
	GetID() string
}

ReadModelID is an interface for read models that can return their ID.

type ReadModelRebuilder added in v1.1.3 

type ReadModelRebuilder struct {
	Name    string
	Rebuild func(ctx context.Context) error
}

ReadModelRebuilder pairs a projection name with the function that rebuilds it over the (now crypto-shredded) event stream. DataEraser triggers it when no in-place SubjectRedactable hook exists, so revoked-key events re-materialize as redacted (the existing WithDecryptionErrorHandler yields redacted payloads on ErrKeyRevoked). Register one with WithReadModelRebuilder.

type ReadModelRepository 

type ReadModelRepository[T any] interface {
	// Get retrieves a read model by ID.
	// Returns ErrNotFound if not found.
	Get(ctx context.Context, id string) (*T, error)

	// GetMany retrieves multiple read models by their IDs.
	// Missing IDs are silently skipped.
	GetMany(ctx context.Context, ids []string) ([]*T, error)

	// Find queries read models with the given criteria.
	Find(ctx context.Context, query Query) ([]*T, error)

	// FindOne returns the first read model matching the query.
	// Returns ErrNotFound if no match.
	FindOne(ctx context.Context, query Query) (*T, error)

	// Count returns the number of read models matching the query.
	Count(ctx context.Context, query Query) (int64, error)

	// Insert creates a new read model.
	// Returns ErrAlreadyExists if ID already exists.
	Insert(ctx context.Context, model *T) error

	// Update modifies an existing read model.
	// Returns ErrNotFound if not found.
	Update(ctx context.Context, id string, updateFn func(*T)) error

	// Upsert creates or updates a read model.
	Upsert(ctx context.Context, model *T) error

	// Delete removes a read model by ID.
	// Returns ErrNotFound if not found.
	Delete(ctx context.Context, id string) error

	// DeleteMany removes all read models matching the query.
	// Returns the number of deleted models.
	DeleteMany(ctx context.Context, query Query) (int64, error)

	// Clear removes all read models.
	Clear(ctx context.Context) error
}

ReadModelRepository provides generic CRUD operations for read models. T is the read model type.

type RebuildOptions 

type RebuildOptions struct {
	// DeleteCheckpoint deletes the existing checkpoint before rebuilding.
	// Default: true
	DeleteCheckpoint bool

	// ClearReadModel calls the projection's Clear method before rebuilding.
	// Only applicable for projections that implement Clearable.
	// Default: true
	ClearReadModel bool

	// ProgressCallback is called periodically with progress updates.
	ProgressCallback ProgressCallback

	// ProgressInterval is how often to call the progress callback.
	// Default: 1 second
	ProgressInterval time.Duration

	// FromPosition starts rebuilding from a specific position.
	// Default: 0 (from beginning)
	FromPosition uint64

	// ToPosition stops rebuilding at a specific position.
	// Default: 0 (to end)
	ToPosition uint64
}

RebuildOptions configures a projection rebuild.

func DefaultRebuildOptions 

func DefaultRebuildOptions() RebuildOptions

DefaultRebuildOptions returns the default rebuild options.

type RebuildProgress 

type RebuildProgress struct {
	// ProjectionName is the name of the projection being rebuilt.
	ProjectionName string

	// TotalEvents is the total number of events to process.
	TotalEvents uint64

	// ProcessedEvents is the number of events processed so far.
	ProcessedEvents uint64

	// CurrentPosition is the current global position.
	CurrentPosition uint64

	// StartedAt is when the rebuild started.
	StartedAt time.Time

	// Duration is the elapsed time.
	Duration time.Duration

	// EventsPerSecond is the processing rate.
	EventsPerSecond float64

	// EstimatedRemaining is the estimated time remaining.
	EstimatedRemaining time.Duration

	// Completed indicates if the rebuild is complete.
	Completed bool

	// Error contains any error that occurred.
	Error error
}

RebuildProgress tracks the progress of a projection rebuild.

type RestartPolicy added in v1.1.13 

type RestartPolicy interface {
	// ShouldRestart reports whether a worker that has now faulted restartCount times
	// (1 for the first restart) should be restarted again. lastErr is the fault cause.
	ShouldRestart(restartCount int, lastErr error) bool

	// Delay returns how long to wait before the given restart attempt (restartCount
	// starts at 1). It should grow with restartCount and be capped, to avoid hot-looping.
	Delay(restartCount int) time.Duration
}

RestartPolicy decides whether and how long to wait before restarting a Faulted async projection worker. The engine's supervising loop consults it after a worker exits into Faulted; see AsyncOptions.RestartPolicy. A restart always resumes from the worker's persisted checkpoint (never from position 0).

func ExponentialBackoffRestart added in v1.1.13 

func ExponentialBackoffRestart(maxRestarts int, baseDelay, maxDelay time.Duration) RestartPolicy

ExponentialBackoffRestart returns a RestartPolicy that restarts a Faulted worker with exponential backoff. A positive maxRestarts caps the number of restarts (after which the worker stays Faulted); a non-positive maxRestarts (0 or negative) restarts indefinitely — the same "non-positive = unlimited" convention as ExponentialBackoffRetry and RetryForever. baseDelay is the first backoff and maxDelay caps it.

func RestartForever added in v1.1.13 

func RestartForever(baseDelay, maxDelay time.Duration) RestartPolicy

RestartForever returns a RestartPolicy that restarts a Faulted worker without limit, with exponential backoff capped at maxDelay. It is the self-documenting spelling of ExponentialBackoffRestart(0, baseDelay, maxDelay).

type RetentionAction added in v1.1.3 

type RetentionAction int

RetentionAction is the action a RetentionPolicy applies to matched events. 

const (
	// ActionShred crypto-shreds matched events by revoking their encryption key.
	// Append-only-safe and unrecoverable. The primary, GDPR-meaningful action.
	ActionShred RetentionAction = iota

	// ActionRedactFields redacts the policy's Fields on matched events. Because the
	// event store is append-only, the redaction is applied through the policy's Apply
	// hook (typically to read models / external stores), not by mutating event rows.
	ActionRedactFields

	// ActionAnonymize pseudonymizes the policy's Fields on matched events, also via
	// the policy's Apply hook (see anonymizer in pii-anonymization).
	ActionAnonymize
)

func (RetentionAction) String added in v1.1.3 

func (a RetentionAction) String() string

String returns the action name.

type RetentionManager added in v1.1.3 

type RetentionManager struct {
	// contains filtered or unexported fields
}

RetentionManager applies retention policies over the event store. It NEVER deletes or mutates event rows — Shred revokes keys, Redact/Anonymize delegate to the policy Apply hook — preserving the append-only log.

Scheduling is the caller's responsibility: Apply performs a single sweep and returns. go-mink does NOT run it on a timer — wire Apply to your own scheduler (cron, gocron, a ticker) at whatever cadence your retention SLA requires. "Sweep" here means one pass, not a self-scheduling loop.

func NewRetentionManager added in v1.1.3 

func NewRetentionManager(store *EventStore, policies []RetentionPolicy, opts ...RetentionManagerOption) *RetentionManager

NewRetentionManager creates a manager for the given store and policies.

func (*RetentionManager) Apply added in v1.1.3 

func (m *RetentionManager) Apply(ctx context.Context) (*RetentionReport, error)

Apply enforces the configured policies and returns a report.

func (*RetentionManager) DryRun added in v1.1.3 

func (m *RetentionManager) DryRun(ctx context.Context) (*RetentionReport, error)

DryRun reports what Apply would do without making any change.

func (*RetentionManager) Validate added in v1.1.3 

func (m *RetentionManager) Validate() []error

Validate returns any policy misconfigurations (e.g. a RedactFields/Anonymize policy with no Apply hook) so a caller can fail fast at startup instead of discovering it in a report. Apply and DryRun also surface these on every run.

type RetentionManagerOption added in v1.1.3 

type RetentionManagerOption func(*RetentionManager)

RetentionManagerOption configures a RetentionManager.

func WithRetentionBatchSize added in v1.1.3 

func WithRetentionBatchSize(size int) RetentionManagerOption

WithRetentionBatchSize sets the scan batch size (default 1000).

func WithRetentionCheckpoint added in v1.1.8 

func WithRetentionCheckpoint(store CheckpointStore, name string) RetentionManagerOption

WithRetentionCheckpoint makes sweeps resumable (mirroring the projection engine's WithCheckpointStore). When configured, Apply starts its scan from the position persisted under name — via the same CheckpointStore projections use — instead of position 0, and after acting persists a safe-resume frontier: the highest global position below which no event can newly match a policy on a future run. This bounds a scheduled sweep's steady-state cost to the events within the retention window rather than the whole, ever-growing store. Absent this option, Apply scans from 0 and persists nothing — exactly as before.

name shares the CheckpointStore keyspace with projection checkpoints, so it MUST NOT collide with a projection name; use a reserved sentinel such as "__mink_retention__".

The persisted frontier is valid only for the current policy set's STATIC matchers (category / stream prefix / event type / tenant). Changing a policy's MaxAge is safe, but broadening a static matcher so it now covers older events already scanned past requires resetting the checkpoint (CheckpointStore.DeleteCheckpoint) or a fresh name — the same rule as rebuilding a projection after changing its logic.

A nil store or empty name is ignored (leaves the manager in its default full-scan mode).

func WithRetentionClock added in v1.1.3 

func WithRetentionClock(now func() time.Time) RetentionManagerOption

WithRetentionClock overrides the clock used for MaxAge evaluation (for testing).

func WithRetentionMaxScan added in v1.1.8 

func WithRetentionMaxScan(n int) RetentionManagerOption

WithRetentionMaxScan bounds a single sweep to at most n scanned events (n <= 0 ⇒ unbounded, the default). It exists to bound the FIRST sweep after enabling retention on an already-large store — which WithRetentionCheckpoint alone cannot, since the first run has no prior frontier and must reach the aged tail once. The remainder is resumed on the next run via the checkpoint, so the cap is only meaningful together with WithRetentionCheckpoint; configured without one it is reported (non-fatally) as ErrRetentionMaxScanNeedsCheckpoint in RetentionReport.Errors and the sweep runs unbounded rather than silently capping and never reaching the tail. A capped run sets RetentionReport.Truncated.

type RetentionPolicy added in v1.1.3 

type RetentionPolicy struct {
	Name string

	// Matchers — any left zero is ignored.
	Category     string        // stream category (text before the first "-")
	StreamPrefix string        // stream-id prefix
	EventTypes   []string      // any-of event types
	TenantID     string        // metadata tenant id
	MaxAge       time.Duration // matches events older than MaxAge (0 = no age bound)

	Action RetentionAction
	Fields []string // fields for RedactFields/Anonymize (applied by Apply)

	// Apply performs RedactFields/Anonymize for a matched event. go-mink cannot mutate
	// append-only rows, so the caller applies the transform to its read models /
	// external stores. Ignored for ActionShred; a Redact/Anonymize policy without
	// Apply leaves matched events unhandled (reported as Skipped, never silent).
	Apply func(ctx context.Context, e StoredEvent) error
}

RetentionPolicy describes a retention rule: a matcher (all set fields must match, AND) and an action. Policies are composable.

func (RetentionPolicy) Validate added in v1.1.3 

func (p RetentionPolicy) Validate() error

Validate reports a configuration error that would make the policy silently do nothing: a RedactFields or Anonymize policy with no Apply hook. go-mink cannot mutate append-only event rows, so those actions MUST be carried out against read models / external stores via Apply — without it, every match is skipped and no anonymization happens even though the sweep "succeeds". RetentionManager surfaces this on every Apply/DryRun so it can never pass unnoticed.

type RetentionReport added in v1.1.3 

type RetentionReport struct {
	DryRun  bool
	Scanned int // events examined
	// Truncated is true when WithRetentionMaxScan stopped this sweep at the per-run scan cap.
	// Not an error: the checkpoint has advanced and the next run resumes from there — if the
	// cap happened to land exactly on the head of the store, that next run simply finds
	// nothing new. Only ever true when a checkpoint is configured.
	Truncated   bool
	Matched     int      // (policy, event) matches
	Acted       int      // matches acted on (shred-with-key or applied hook)
	Skipped     int      // matches with no applicable handler (residual; e.g. Redact w/o Apply)
	KeysRevoked []string // distinct keys crypto-shredded (sorted)
	Errors      []error  // non-fatal per-action errors (incl. loud policy-misconfig errors)
}

RetentionReport summarizes a retention sweep.

func (*RetentionReport) Failed added in v1.1.3 

func (r *RetentionReport) Failed() bool

Failed reports whether the sweep had any error — a per-action failure or a misconfigured policy (e.g. RedactFields/Anonymize with no Apply hook). A caller SHOULD check it: a "successful" (nil-error) Apply can still have skipped everything.

type RetryConfig 

type RetryConfig struct {
	// MaxAttempts is the maximum number of attempts (including the first one).
	MaxAttempts int

	// InitialDelay is the initial delay between retries.
	InitialDelay time.Duration

	// MaxDelay is the maximum delay between retries.
	MaxDelay time.Duration

	// Multiplier is the factor by which the delay increases on each retry.
	Multiplier float64

	// ShouldRetry determines if an error should be retried.
	// If nil, all errors are retried.
	ShouldRetry func(err error) bool
}

RetryMiddleware retries failed commands.

func DefaultRetryConfig 

func DefaultRetryConfig() RetryConfig

DefaultRetryConfig returns a default retry configuration.

type RetryEvent added in v1.1.10 

type RetryEvent struct {
	// SagaID is the id of the re-driven saga.
	SagaID string

	// SagaType is the saga's type.
	SagaType string

	// FromStatus is the status the saga was in when the re-drive was initiated.
	FromStatus SagaStatus

	// ResultStatus is the status the saga reached after the re-drive (meaningful
	// when Err is nil): SagaStatusCompleted on a clean recovery, or a settled
	// unsuccessful status (Failed / Compensated / CompensationFailed) if it failed
	// again. Zero-valued when the attempt did not run to a persisted outcome.
	ResultStatus SagaStatus

	// At is when the attempt was made.
	At time.Time

	// Err is the attempt's operational error, or nil (see the type doc).
	Err error
}

RetryEvent describes a single operator-initiated saga re-drive attempt reported to a WithSagaRetryObserver.

Err reports an OPERATIONAL failure of the attempt — the re-drive could not be applied — e.g. ErrConcurrencyConflict, a store error, or a failure to decode the captured event. It is nil when the re-drive ran to a persisted outcome. A re-drive that RAN but whose saga failed again (its command re-failed, driving compensation) is NOT reported via Err — the whole saga machinery treats compensation as handled; inspect ResultStatus (or the saga's persisted status) to see the business outcome.

type RetryOutcome added in v1.1.10 

type RetryOutcome int

RetryOutcome classifies the result of re-driving one saga within a batch (RetrySagasByType). 

const (
	// RetrySucceeded means the re-drive was applied and the saga reached Completed.
	RetrySucceeded RetryOutcome = iota

	// RetryFailedAgain means the re-drive ran but the saga did not complete — it hit
	// a fresh failure and followed the compensation path again, or the attempt
	// returned a non-conflict operational error (see SagaRetryResult.Err).
	RetryFailedAgain

	// RetryConflicted means an optimistic-concurrency conflict prevented the
	// re-drive; the saga is unchanged and the operation can be re-issued.
	RetryConflicted

	// RetrySkipped means the saga was not re-driven because it was not retryable
	// (e.g. its status changed under the lock, or it has no captured trigger event).
	RetrySkipped
)

func (RetryOutcome) String added in v1.1.10 

func (o RetryOutcome) String() string

String returns a lower-case label for the outcome.

type RetryPolicy 

type RetryPolicy interface {
	// ShouldRetry returns true if the operation should be retried.
	ShouldRetry(attempt int, err error) bool

	// Delay returns the duration to wait before the next retry.
	Delay(attempt int) time.Duration
}

RetryPolicy defines how to handle retries for failed operations.

func ExponentialBackoffRetry 

func ExponentialBackoffRetry(maxRetries int, baseDelay, maxDelay time.Duration) RetryPolicy

ExponentialBackoffRetry creates a RetryPolicy with exponential backoff. A positive maxRetries caps the attempts (retry while attempt < maxRetries); a non-positive maxRetries (0 or negative) means retry indefinitely — the same "non-positive = retry forever" convention AsyncOptions.MaxRetries uses, and the basis of RetryForever. To never retry, use NoRetry rather than a zero count. baseDelay is the first backoff and maxDelay caps it.

func NoRetry 

func NoRetry() RetryPolicy

NoRetry returns a retry policy that never retries. It is the single canonical way to stop on the first error; prefer it over ExponentialBackoffRetry(0, …), which now means "retry forever."

func RetryForever added in v1.1.13 

func RetryForever(baseDelay, maxDelay time.Duration) RetryPolicy

RetryForever returns a RetryPolicy that retries on every non-nil error, without limit, using exponential backoff between attempts capped at maxDelay. It is the self-documenting spelling of ExponentialBackoffRetry(0, baseDelay, maxDelay) — prefer it when the intent is "retry forever with backoff." Use NoRetry to never retry.

type RetryReport added in v1.1.10 

type RetryReport struct {
	// Results holds the per-saga outcomes, in the order the sagas were processed.
	Results []SagaRetryResult
}

RetryReport is the aggregate result of RetrySagasByType, one entry per saga.

func (RetryReport) Count added in v1.1.10 

func (r RetryReport) Count(o RetryOutcome) int

Count returns how many sagas ended with the given outcome.

type Retryable added in v1.1.13 

type Retryable interface {
	Retryable() bool
}

Retryable is an error that can declare itself retryable. DefaultErrorClassifier classifies an error whose Unwrap chain implements Retryable and returns true as ErrorClassTransient. It mirrors the net.Error "Temporary() bool" idiom, but is exported so callers can mark their own errors without depending on the net package.

type Saga 

type Saga interface {
	// SagaID returns the unique identifier for this saga instance.
	SagaID() string

	// SagaType returns the type of this saga (e.g., "OrderFulfillment").
	SagaType() string

	// Status returns the current status of the saga.
	Status() SagaStatus

	// SetStatus sets the saga status.
	SetStatus(status SagaStatus)

	// CurrentStep returns the current step number (0-based).
	CurrentStep() int

	// SetCurrentStep sets the current step number.
	SetCurrentStep(step int)

	// CorrelationID returns the correlation ID for this saga.
	// Used to correlate events to this saga instance.
	CorrelationID() string

	// SetCorrelationID sets the correlation ID.
	SetCorrelationID(id string)

	// HandledEvents returns the list of event types this saga handles.
	HandledEvents() []string

	// HandleEvent processes an event and returns commands to dispatch.
	// The returned commands will be executed by the saga manager.
	HandleEvent(ctx context.Context, event StoredEvent) ([]Command, error)

	// Compensate is called when the saga needs to rollback.
	// It returns compensating commands to undo previous steps.
	Compensate(ctx context.Context, failedStep int, failureReason error) ([]Command, error)

	// IsComplete returns true if the saga has completed successfully.
	IsComplete() bool

	// StartedAt returns when the saga started.
	StartedAt() time.Time

	// SetStartedAt sets when the saga started.
	SetStartedAt(t time.Time)

	// CompletedAt returns when the saga completed (nil if not completed).
	CompletedAt() *time.Time

	// SetCompletedAt sets when the saga completed.
	SetCompletedAt(t *time.Time)

	// Data returns the saga's internal state as a map.
	// This is serialized and stored in the saga store.
	Data() map[string]interface{}

	// SetData restores the saga's internal state from a map.
	SetData(data map[string]interface{})

	// Version returns the saga version for optimistic concurrency.
	Version() int64

	// SetVersion sets the saga version.
	SetVersion(v int64)

	// IncrementVersion increments the saga version.
	IncrementVersion()
}

Saga defines the interface for saga implementations. A saga coordinates long-running business processes across multiple aggregates.

type SagaBase 

type SagaBase struct {
	// contains filtered or unexported fields
}

SagaBase provides a default partial implementation of the Saga interface. Embed this struct in your saga types to get default behavior.

func NewSagaBase 

func NewSagaBase(id, sagaType string) SagaBase

NewSagaBase creates a new SagaBase with the given ID and type.

func (*SagaBase) Complete 

func (s *SagaBase) Complete()

Complete marks the saga as completed.

func (*SagaBase) CompletedAt 

func (s *SagaBase) CompletedAt() *time.Time

CompletedAt returns when the saga completed.

func (*SagaBase) CorrelationID 

func (s *SagaBase) CorrelationID() string

CorrelationID returns the correlation ID.

func (*SagaBase) CurrentStep 

func (s *SagaBase) CurrentStep() int

CurrentStep returns the current step number.

func (*SagaBase) Fail 

func (s *SagaBase) Fail()

Fail marks the saga as failed.

func (*SagaBase) IncrementVersion 

func (s *SagaBase) IncrementVersion()

IncrementVersion increments the saga version.

func (*SagaBase) MarkCompensated 

func (s *SagaBase) MarkCompensated()

MarkCompensated marks the saga as compensated.

func (*SagaBase) RecordStep added in v1.1.10 

func (s *SagaBase) RecordStep(step SagaStep)

RecordStep appends a step to the saga's history.

func (*SagaBase) SagaID 

func (s *SagaBase) SagaID() string

SagaID returns the saga's unique identifier.

func (*SagaBase) SagaType 

func (s *SagaBase) SagaType() string

SagaType returns the saga type.

func (*SagaBase) SetCompletedAt 

func (s *SagaBase) SetCompletedAt(t *time.Time)

SetCompletedAt sets when the saga completed.

func (*SagaBase) SetCorrelationID 

func (s *SagaBase) SetCorrelationID(id string)

SetCorrelationID sets the correlation ID.

func (*SagaBase) SetCurrentStep 

func (s *SagaBase) SetCurrentStep(step int)

SetCurrentStep sets the current step number.

func (*SagaBase) SetID 

func (s *SagaBase) SetID(id string)

SetID sets the saga's ID.

func (*SagaBase) SetStartedAt 

func (s *SagaBase) SetStartedAt(t time.Time)

SetStartedAt sets when the saga started.

func (*SagaBase) SetStatus 

func (s *SagaBase) SetStatus(status SagaStatus)

SetStatus sets the saga status.

func (*SagaBase) SetSteps added in v1.1.10 

func (s *SagaBase) SetSteps(steps []SagaStep)

SetSteps replaces the saga's recorded step history (used by the manager during hydration).

func (*SagaBase) SetType 

func (s *SagaBase) SetType(t string)

SetType sets the saga type.

func (*SagaBase) SetVersion 

func (s *SagaBase) SetVersion(v int64)

SetVersion sets the saga version.

func (*SagaBase) StartCompensation 

func (s *SagaBase) StartCompensation()

StartCompensation marks the saga as compensating.

func (*SagaBase) StartedAt 

func (s *SagaBase) StartedAt() time.Time

StartedAt returns when the saga started.

func (*SagaBase) Status 

func (s *SagaBase) Status() SagaStatus

Status returns the current status.

func (*SagaBase) Steps added in v1.1.10 

func (s *SagaBase) Steps() []SagaStep

Steps returns the saga's recorded step history. The SagaManager persists this to SagaState.Steps on save and restores it on hydrate, so the history survives across events; it is currently used to record operator re-drives (see SagaManager.RetrySaga). A saga type that does not embed SagaBase simply has no persisted step history.

func (*SagaBase) Version 

func (s *SagaBase) Version() int64

Version returns the saga version.

type SagaCorrelation 

type SagaCorrelation struct {
	// SagaType is the type of saga this correlation applies to.
	SagaType string

	// EventTypes are the event types that can start this saga.
	StartingEvents []string

	// CorrelationIDFunc extracts the correlation ID from an event.
	// This is used to find existing sagas or create new ones.
	CorrelationIDFunc func(event StoredEvent) string
}

SagaCorrelation provides strategies for correlating events to sagas.

type SagaFactory 

type SagaFactory func(id string) Saga

SagaFactory creates new saga instances.

type SagaFailedError 

type SagaFailedError struct {
	SagaID      string
	SagaType    string
	FailedStep  int
	Reason      string
	Recoverable bool
}

SagaFailedError provides detailed information about a saga failure.

func NewSagaFailedError 

func NewSagaFailedError(sagaID, sagaType string, failedStep int, reason string, recoverable bool) *SagaFailedError

NewSagaFailedError creates a new SagaFailedError.

func (*SagaFailedError) Error 

func (e *SagaFailedError) Error() string

Error returns the error message.

func (*SagaFailedError) Is 

func (e *SagaFailedError) Is(target error) bool

Is reports whether this error matches the target error.

type SagaManager 

type SagaManager struct {
	// contains filtered or unexported fields
}

SagaManager orchestrates saga lifecycle and event processing. It subscribes to events, routes them to appropriate sagas, and dispatches resulting commands.

Concurrency and Idempotency

SagaManager provides several mechanisms to ensure correct saga processing under concurrent access:

  1. Per-Saga Locking: Each saga ID has an associated mutex that serializes access. This prevents race conditions when the same event is delivered from multiple sources (e.g., pg_notify + polling) or when multiple events for the same saga arrive simultaneously.

  2. Fresh State Loading: Before processing each event, the saga state is loaded fresh from the store. This ensures terminal status checks and idempotency checks see the latest state.

  3. Event Idempotency: Processed events are tracked in the SagaState.ProcessedEvents field (not in the saga's Data map) to prevent duplicate processing on retries. This is handled transparently by the SagaManager - saga implementations don't need to preserve these internal tracking fields.

  4. Optimistic Concurrency: The saga store uses version-based optimistic concurrency control. On conflict, the event is retried with fresh state.

func NewSagaManager 

func NewSagaManager(eventStore *EventStore, opts ...SagaManagerOption) *SagaManager

NewSagaManager creates a new SagaManager.

func (*SagaManager) FindSagaByCorrelationID 

func (m *SagaManager) FindSagaByCorrelationID(ctx context.Context, correlationID string) (*SagaState, error)

FindSagaByCorrelationID finds a saga by its correlation ID.

func (*SagaManager) FindSagasByType added in v1.0.25 

func (m *SagaManager) FindSagasByType(ctx context.Context, sagaType string, statuses ...SagaStatus) ([]*SagaState, error)

FindSagasByType returns saga states of the given type, optionally filtered by status. This is useful for recovery tooling and dashboards (e.g. listing Running or CompensationFailed sagas that need operator attention).

func (*SagaManager) GetSaga 

func (m *SagaManager) GetSaga(ctx context.Context, sagaID string) (*SagaState, error)

GetSaga retrieves a saga by its ID.

func (*SagaManager) IsRunning 

func (m *SagaManager) IsRunning() bool

IsRunning returns true if the saga manager is running.

func (*SagaManager) Position 

func (m *SagaManager) Position() uint64

Position returns the current event position.

func (*SagaManager) ProcessEvent 

func (m *SagaManager) ProcessEvent(ctx context.Context, event StoredEvent) error

ProcessEvent manually processes a single event (for testing or manual replay).

func (*SagaManager) Register 

func (m *SagaManager) Register(sagaType string, factory SagaFactory, correlation SagaCorrelation)

Register registers a saga type with its factory and correlation configuration.

func (*SagaManager) RegisterSimple 

func (m *SagaManager) RegisterSimple(sagaType string, factory SagaFactory, startingEvents ...string)

RegisterSimple registers a saga with a simple correlation based on event stream ID.

func (*SagaManager) ResumeStalled added in v1.1.10 

func (m *SagaManager) ResumeStalled(ctx context.Context, sagaID string) error

ResumeStalled re-drives a Running saga whose worker died mid-dispatch, now, instead of waiting for the WithSagaTimeout sweep to compensate it. It accepts only a Running saga (any other status is rejected with *SagaNotRetryableError / ErrSagaNotRetryable) and, to avoid fighting a live worker, requires the saga's UpdatedAt to be older than the configured saga timeout (WithSagaTimeout); a saga updated more recently, or a manager with no timeout configured, is refused.

It reuses RetrySaga's machinery exactly: the same per-saga lock, optimistic concurrency, single-event idempotency reset, SagaStep record, and WithSagaRetryObserver reporting.

func (*SagaManager) RetrySaga added in v1.1.10 

func (m *SagaManager) RetrySaga(ctx context.Context, sagaID string) error

RetrySaga re-drives a settled-but-unsuccessful saga once its underlying cause is fixed — the operator-initiated recovery counterpart to the automatic timeout sweep. It re-delivers the saga's last trigger event through the normal processing path, so a retry has identical semantics to a fresh delivery: on success the saga reaches Completed; on a fresh failure it follows the same compensation path as a first-time failure.

Which statuses are retryable

RetrySaga accepts exactly the settled-but-unsuccessful statuses (SagaState.IsRetryable): 

Failed              re-drive the forward path (no compensation ran)
Compensated         rolled back to a clean state — safe to re-drive forward
CompensationFailed  stuck after a partial rollback — re-drive is the recovery

It rejects, with a typed *SagaNotRetryableError (matching errors.Is(err, ErrSagaNotRetryable)): 

Completed                       terminal success — never re-run
Started / Running / Compensating in-flight — the event loop / timeout sweep owns it

A missing saga returns the existing ErrSagaNotFound. To re-drive a stalled Running saga whose worker died mid-dispatch, use ResumeStalled instead.

Idempotency

Re-drive re-delivers a single event: only that event's idempotency key is cleared before re-delivery, so already-succeeded earlier steps (recorded in ProcessedEvents) are NOT re-dispatched. Beyond that, re-drive relies on the same contract as the store's at-least-once delivery — saga command handlers MUST be idempotent. A retry that fails again lands in the normal failure/compensation path and may be retried again once the cause is truly fixed.

Concurrency

The load -> re-drive -> save runs under the same per-saga lock the event loop uses (so it cannot race an in-flight event for that saga), and the save uses optimistic concurrency on the loaded version. A concurrent modification surfaces as ErrConcurrencyConflict and is NOT silently swallowed or retried internally — an operator action fails loudly and can be re-issued.

Mechanism

The last trigger event is recovered from the manager-owned reserved slot of the saga's stored state, captured during normal processing when WithSagaRetryCapture is enabled (see reservedLastEventKey) — no event-store re-read and no schema change. Capture is opt-in and zero-overhead when off; a retryable-status saga with no captured event (capture disabled, or the saga last ran before it was enabled) is rejected with a clear reason rather than guessing.

Observed outcome

The returned error reports an OPERATIONAL failure (ErrConcurrencyConflict, a store error, or an undecodable captured event). A re-drive that RAN but whose saga failed again (driving compensation) returns nil — inspect the saga's status, or a WithSagaRetryObserver's RetryEvent.ResultStatus, for the business outcome (RetrySagasByType classifies this for you). The re-drive is auditable: it is recorded as a SagaStep on the saga's history and reported to any observer.

func (*SagaManager) RetrySagasByType added in v1.1.10 

func (m *SagaManager) RetrySagasByType(ctx context.Context, sagaType string, statuses ...SagaStatus) (RetryReport, error)

RetrySagasByType finds sagas of the given type in the given statuses (via FindByType) and applies RetrySaga to each, sequentially, returning a per-saga RetryReport. It is a pure composition of the single-saga primitive: each saga is re-driven under its own per-saga lock, and a failure of one does NOT abort the batch — every matched saga is attempted and its individual outcome recorded.

Pass the retryable statuses to target (e.g. SagaStatusFailed, SagaStatusCompensationFailed); a matched saga that is not retryable when its turn comes is reported as RetrySkipped rather than failing the batch. The returned error is non-nil only if the initial FindByType lookup fails.

func (*SagaManager) SetPosition 

func (m *SagaManager) SetPosition(pos uint64)

SetPosition sets the starting position for event processing.

func (*SagaManager) Start 

func (m *SagaManager) Start(ctx context.Context) error

Start begins processing events and routing them to sagas. This method blocks until the context is cancelled.

func (*SagaManager) StartAsync 

func (m *SagaManager) StartAsync(ctx context.Context) *AsyncResult

StartAsync begins processing events and routing them to sagas in a background goroutine. It returns immediately with an AsyncResult that can be used to:

  • Wait for the saga manager to stop: result.Wait()
  • Wait with timeout: result.WaitWithTimeout(5 * time.Second)
  • Check if stopped: result.IsComplete()
  • Cancel the manager: result.Cancel()
  • Get the error: result.Err()

The saga manager will continue processing until:

  • The provided context is cancelled
  • result.Cancel() is called
  • An unrecoverable error occurs

Example: 

result := manager.StartAsync(ctx)

// Do other work while saga manager runs in background...

// Later, when shutting down:
result.Cancel()
if err := result.WaitWithTimeout(10 * time.Second); err != nil {
    log.Printf("Saga manager shutdown: %v", err)
}

func (*SagaManager) StartSaga 

func (m *SagaManager) StartSaga(ctx context.Context, sagaType string, triggerEvent StoredEvent) error

StartSaga manually triggers a new saga instance synchronously. The saga will be created and the trigger event will be processed immediately.

This is useful when you want to start a saga based on an external trigger rather than waiting for an event to flow through the event store subscription.

func (*SagaManager) StartSagaAsync 

func (m *SagaManager) StartSagaAsync(ctx context.Context, sagaType string, triggerEvent StoredEvent) *AsyncResult

StartSagaAsync manually triggers a new saga instance asynchronously. The saga will be started and the first event will be processed in a background goroutine. Returns an AsyncResult that can be used to wait for the initial processing to complete.

This is useful when you want to start a saga based on an external trigger rather than waiting for an event to flow through the event store subscription.

Example: 

result := manager.StartSagaAsync(ctx, "OrderFulfillment", initialEvent)
if err := result.WaitWithTimeout(5 * time.Second); err != nil {
    log.Printf("Failed to start saga: %v", err)
}

func (*SagaManager) Stop 

func (m *SagaManager) Stop()

Stop gracefully stops the saga manager and waits for the processing loop to exit, so no event is being processed once Stop returns.

type SagaManagerOption 

type SagaManagerOption func(*SagaManager)

SagaManagerOption configures a SagaManager.

func WithCommandBus 

func WithCommandBus(bus *CommandBus) SagaManagerOption

WithCommandBus sets the command bus for dispatching commands.

func WithSagaLogger 

func WithSagaLogger(logger Logger) SagaManagerOption

WithSagaLogger sets the logger.

func WithSagaPollInterval 

func WithSagaPollInterval(d time.Duration) SagaManagerOption

WithSagaPollInterval sets the cadence of the background timeout sweep that detects abandoned sagas (see WithSagaTimeout). It does not affect the primary event subscription, which is push-based.

func WithSagaRetryAttempts 

func WithSagaRetryAttempts(attempts int) SagaManagerOption

WithSagaRetryAttempts sets the number of retry attempts for failed commands. The value is clamped to a minimum of 1: a saga event is always attempted at least once. Passing 0 (or a negative value) would otherwise make the retry loops skip their body entirely, silently dropping the event while advancing the manager's position past it.

func WithSagaRetryCapture added in v1.1.10 

func WithSagaRetryCapture() SagaManagerOption

WithSagaRetryCapture enables capturing each saga's last trigger event so a settled saga can later be re-driven by RetrySaga / ResumeStalled. It is opt-in and OFF by default: when unset the manager captures nothing and adds zero overhead (honoring the library's zero-overhead-when-unconfigured invariant), and RetrySaga on a saga with no captured event returns ErrSagaNotRetryable. Enable it on managers whose sagas you may need to recover operationally.

The captured event is stored in the manager-owned reserved slot of the saga's persisted state (SagaState.Data under reservedLastEventKey). The manager stamps and reads it itself and strips it before the saga's own SetData ever sees it, so it never leaks into saga-author code and it works regardless of how a saga implements Data()/SetData() — including projection-style sagas that rebuild Data from typed fields (which would otherwise silently drop a key written into Data).

func WithSagaRetryDelay 

func WithSagaRetryDelay(d time.Duration) SagaManagerOption

WithSagaRetryDelay sets the delay between retry attempts.

func WithSagaRetryObserver added in v1.1.10 

func WithSagaRetryObserver(fn func(RetryEvent)) SagaManagerOption

WithSagaRetryObserver registers a hook invoked once per operator-initiated re-drive attempt (RetrySaga / ResumeStalled / RetrySagasByType), with the saga id, saga type, the status it was re-driven from, the status it reached, the attempt time, and the attempt's error. It makes a re-drive auditable — route it to an audit log or metrics. It is additive and opt-in: with no observer configured a re-drive is still recorded as a SagaStep on the saga's history, so it is never a silent state change.

The observer is called synchronously on the caller's goroutine while the per-saga lock is NOT held; keep it fast and non-blocking, and do not call back into the SagaManager from it.

func WithSagaSerializer 

func WithSagaSerializer(serializer Serializer) SagaManagerOption

WithSagaSerializer sets the serializer for saga data.

func WithSagaStore 

func WithSagaStore(store SagaStore) SagaManagerOption

WithSagaStore sets the saga store.

func WithSagaSweepInterval added in v1.0.25 

func WithSagaSweepInterval(d time.Duration) SagaManagerOption

WithSagaSweepInterval sets how often the abandoned-saga timeout sweep runs (see WithSagaTimeout). Timeout detection tolerates coarse cadence, so this should be well above the per-event poll interval to avoid constant store queries. When unset, the sweep falls back to the poll interval, floored at 1 second.

func WithSagaTimeout added in v1.0.25 

func WithSagaTimeout(d time.Duration) SagaManagerOption

WithSagaTimeout enables timeout handling for long-running sagas. When set to a positive duration, the manager periodically sweeps for Running sagas whose last update is older than the timeout and drives them into compensation. The default (0) disables the sweep. The sweep cadence is controlled by WithSagaSweepInterval.

type SagaNotFoundError 

type SagaNotFoundError = adapters.SagaNotFoundError

SagaNotFoundError provides detailed information about a missing saga. This is a type alias to adapters.SagaNotFoundError for consistency.

type SagaNotRetryableError added in v1.1.10 

type SagaNotRetryableError struct {
	SagaID string
	Status SagaStatus
	Reason string
}

SagaNotRetryableError is the typed form of ErrSagaNotRetryable, carrying the saga's id, the status it was in when the re-drive was refused, and a human-readable reason. Match with errors.Is(err, ErrSagaNotRetryable); type-assert to *SagaNotRetryableError for the details.

func (*SagaNotRetryableError) Error added in v1.1.10 

func (e *SagaNotRetryableError) Error() string

Error returns the error message.

func (*SagaNotRetryableError) Is added in v1.1.10 

func (e *SagaNotRetryableError) Is(target error) bool

Is reports whether this error matches the target error.

func (*SagaNotRetryableError) Unwrap added in v1.1.10 

func (e *SagaNotRetryableError) Unwrap() error

Unwrap returns the underlying sentinel for errors.Is / errors.Unwrap.

type SagaRetryResult added in v1.1.10 

type SagaRetryResult struct {
	// SagaID is the saga this result is for.
	SagaID string

	// Outcome classifies what happened.
	Outcome RetryOutcome

	// Err is the attempt's error, when any (nil for a clean RetrySucceeded).
	Err error
}

SagaRetryResult is the outcome of re-driving a single saga in a batch.

type SagaState 

type SagaState = adapters.SagaState

SagaState represents the persisted state of a saga.

func SagaStateFromJSON 

func SagaStateFromJSON(data []byte) (*SagaState, error)

SagaStateFromJSON parses saga state from JSON.

type SagaStatus 

type SagaStatus = adapters.SagaStatus

SagaStatus represents the current status of a saga.

type SagaStep 

type SagaStep = adapters.SagaStep

SagaStep represents a single step in a saga.

type SagaStepStatus 

type SagaStepStatus = adapters.SagaStepStatus

SagaStepStatus represents the status of a saga step.

type SagaStore 

type SagaStore = adapters.SagaStore

SagaStore defines the interface for saga persistence.

type SchemaCompatibility 

type SchemaCompatibility int

SchemaCompatibility represents the level of backward compatibility between schema versions. 

const (
	// SchemaFullyCompatible indicates the schema change is fully backward and forward compatible.
	// No fields were added, removed, or changed — only documentation or ordering changes.
	SchemaFullyCompatible SchemaCompatibility = iota

	// SchemaBackwardCompatible indicates old data can be read by new code.
	// Fields may have been added (with defaults) but none removed or changed.
	SchemaBackwardCompatible

	// SchemaForwardCompatible indicates new data can be read by old code.
	// Fields may have been removed but none added or changed.
	SchemaForwardCompatible

	// SchemaBreaking indicates the schema change breaks compatibility.
	// Fields were changed, renamed, or removed without migration support.
	SchemaBreaking
)

func (SchemaCompatibility) String 

func (c SchemaCompatibility) String() string

String returns a human-readable name for the compatibility level.

type SchemaDefinition 

type SchemaDefinition struct {
	// Version is the schema version number.
	Version int

	// Fields describes the fields in this schema version.
	Fields []FieldDefinition

	// JSONSchema is an optional JSON Schema document for validation.
	JSONSchema json.RawMessage

	// RegisteredAt is when this schema version was registered.
	RegisteredAt time.Time
}

SchemaDefinition describes the schema for a specific version of an event type.

type SchemaRegistry 

type SchemaRegistry struct {
	// contains filtered or unexported fields
}

SchemaRegistry is an in-memory registry that tracks event schemas and their versions. It provides compatibility checking between schema versions.

func NewSchemaRegistry 

func NewSchemaRegistry() *SchemaRegistry

NewSchemaRegistry creates a new empty SchemaRegistry.

func (*SchemaRegistry) CheckCompatibility 

func (r *SchemaRegistry) CheckCompatibility(eventType string, oldVersion, newVersion int) (SchemaCompatibility, error)

CheckCompatibility determines the compatibility level between two schema versions. It compares the field definitions — name, type, and the Required flag — to classify the change. newVersion must be greater than oldVersion.

A field added as Required, or an existing field whose Required flag flips from false to true, breaks backward compatibility because data written under the old schema may omit it.

func (*SchemaRegistry) GetLatestVersion 

func (r *SchemaRegistry) GetLatestVersion(eventType string) (int, error)

GetLatestVersion returns the highest registered schema version for an event type.

func (*SchemaRegistry) GetSchema 

func (r *SchemaRegistry) GetSchema(eventType string, version int) (*SchemaDefinition, error)

GetSchema retrieves a specific schema version for an event type.

func (*SchemaRegistry) Register 

func (r *SchemaRegistry) Register(eventType string, schema SchemaDefinition) error

Register adds a schema definition for an event type. Returns an error if the version is < 1 or already registered.

func (*SchemaRegistry) RegisteredEventTypes 

func (r *SchemaRegistry) RegisteredEventTypes() []string

RegisteredEventTypes returns a sorted list of event types that have schemas registered.

func (*SchemaRegistry) RequireBackwardCompatible added in v1.0.25 

func (r *SchemaRegistry) RequireBackwardCompatible(eventType string, oldVersion, newVersion int) error

RequireBackwardCompatible returns an *IncompatibleSchemaError if evolving eventType from oldVersion to newVersion is not at least backward compatible (i.e. new code cannot safely read old data). It returns nil when the change is SchemaFullyCompatible or SchemaBackwardCompatible.

type SchemaVersionGapError 

type SchemaVersionGapError struct {
	EventType       string
	MissingVersion  int
	ExpectedVersion int
}

SchemaVersionGapError provides detailed information about a gap in the upcaster chain.

func NewSchemaVersionGapError 

func NewSchemaVersionGapError(eventType string, missingVersion, expectedVersion int) *SchemaVersionGapError

NewSchemaVersionGapError creates a new SchemaVersionGapError.

func (*SchemaVersionGapError) Error 

func (e *SchemaVersionGapError) Error() string

Error returns the error message.

func (*SchemaVersionGapError) Is 

func (e *SchemaVersionGapError) Is(target error) bool

Is reports whether this error matches the target error.

func (*SchemaVersionGapError) Unwrap 

func (e *SchemaVersionGapError) Unwrap() error

Unwrap returns the underlying error for errors.Unwrap().

type SerializationError 

type SerializationError struct {
	EventType string
	Operation string // "serialize" or "deserialize"
	Cause     error
}

SerializationError provides detailed information about a serialization failure.

func NewSerializationError 

func NewSerializationError(eventType, operation string, cause error) *SerializationError

NewSerializationError creates a new SerializationError.

func (*SerializationError) Error 

func (e *SerializationError) Error() string

Error returns the error message.

func (*SerializationError) Is 

func (e *SerializationError) Is(target error) bool

Is reports whether this error matches the target error.

func (*SerializationError) Unwrap 

func (e *SerializationError) Unwrap() error

Unwrap returns the underlying cause for errors.Unwrap().

type Serializer 

type Serializer interface {
	// Serialize converts an event to bytes.
	Serialize(event interface{}) ([]byte, error)

	// Deserialize converts bytes back to an event.
	// The eventType is used to determine the target type.
	Deserialize(data []byte, eventType string) (interface{}, error)
}

Serializer handles event payload serialization and deserialization.

type SharedKeyError added in v1.1.3 

type SharedKeyError struct {
	SubjectID string
	// SharedKeys are the keys that also protect other (or untagged) subjects' events.
	SharedKeys []string
	// OtherSubjects samples subject ids (other than the target) found under those keys.
	OtherSubjects []string
}

SharedKeyError reports that erasing the target subject would revoke one or more keys that also protect other subjects' events — the per-tenant-key blast radius. It is returned by Erase (before any revocation) when WithSharedKeyGuard is set and AllowSharedKeyRevocation is not.

func (*SharedKeyError) Error added in v1.1.3 

func (e *SharedKeyError) Error() string

Error returns the error message.

func (*SharedKeyError) Is added in v1.1.3 

func (e *SharedKeyError) Is(target error) bool

Is reports whether this error matches the target error.

type SimpleDispatcher 

type SimpleDispatcher struct {
	// contains filtered or unexported fields
}

SimpleDispatcher is a basic dispatcher that forwards commands to handlers.

func NewSimpleDispatcher 

func NewSimpleDispatcher(registry *HandlerRegistry) *SimpleDispatcher

NewSimpleDispatcher creates a new SimpleDispatcher.

func (*SimpleDispatcher) Dispatch 

func (d *SimpleDispatcher) Dispatch(ctx context.Context, cmd Command) (CommandResult, error)

Dispatch sends a command to its handler.

type StoredEvent 

type StoredEvent struct {
	// ID is the globally unique event identifier.
	ID string

	// StreamID identifies the stream this event belongs to.
	StreamID string

	// Type is the event type identifier.
	Type string

	// Data is the serialized event payload.
	Data []byte

	// Metadata contains contextual information.
	Metadata Metadata

	// Version is the position within the stream (1-based).
	Version int64

	// GlobalPosition is the position across all streams.
	GlobalPosition uint64

	// Timestamp is when the event was stored.
	Timestamp time.Time
}

StoredEvent represents a persisted event with all storage metadata.

type StreamID 

type StreamID struct {
	// Category represents the aggregate type (e.g., "Order", "Customer").
	Category string

	// ID is the unique identifier within the category (e.g., "order-123").
	ID string
}

StreamID uniquely identifies an event stream. It consists of a category (aggregate type) and an instance ID.

func NewStreamID 

func NewStreamID(category, id string) StreamID

NewStreamID creates a new StreamID from category and ID.

func ParseStreamID 

func ParseStreamID(s string) (StreamID, error)

ParseStreamID parses a stream ID string in the format "Category-ID". Returns an error if the format is invalid.

func (StreamID) IsZero 

func (s StreamID) IsZero() bool

IsZero reports whether the StreamID is empty.

func (StreamID) String 

func (s StreamID) String() string

String returns the stream ID as "Category-ID".

func (StreamID) Validate 

func (s StreamID) Validate() error

Validate checks if the StreamID is valid.

type StreamInfo 

type StreamInfo struct {
	// StreamID is the stream identifier.
	StreamID string

	// Category is the stream category (aggregate type).
	Category string

	// Version is the current stream version.
	Version int64

	// EventCount is the number of events in the stream.
	EventCount int64

	// CreatedAt is when the stream was created.
	CreatedAt time.Time

	// UpdatedAt is when the stream was last modified.
	UpdatedAt time.Time
}

StreamInfo contains metadata about an event stream.

type StreamNotFoundError 

type StreamNotFoundError struct {
	StreamID string
}

StreamNotFoundError provides detailed information about a missing stream.

func NewStreamNotFoundError 

func NewStreamNotFoundError(streamID string) *StreamNotFoundError

NewStreamNotFoundError creates a new StreamNotFoundError.

func (*StreamNotFoundError) Error 

func (e *StreamNotFoundError) Error() string

Error returns the error message.

func (*StreamNotFoundError) Is 

func (e *StreamNotFoundError) Is(target error) bool

Is reports whether this error matches the target error.

func (*StreamNotFoundError) Unwrap 

func (e *StreamNotFoundError) Unwrap() error

Unwrap returns the underlying error for errors.Unwrap().

type SubjectAuditPurger added in v1.1.3 

type SubjectAuditPurger = adapters.SubjectAuditPurger

SubjectAuditPurger is the optional AuditStore extension for GDPR erasure of a subject's audit trail (see NewAuditSubjectEraser).

type SubjectErasable added in v1.1.3 

type SubjectErasable interface {
	// EraseSubject removes the subject's data from the store. footprint carries the
	// subject's resolved streams and revoked keys (e.g. a snapshot eraser keys off
	// footprint.Streams). It returns an outcome (count erased, or Skipped when the
	// store cannot target the subject) and a non-nil error only on a real failure.
	EraseSubject(ctx context.Context, subjectID string, footprint *SubjectFootprint) (SubjectErasureOutcome, error)

	// ErasableName identifies the store in ErasureResult (no PII), e.g. "audit".
	ErasableName() string
}

SubjectErasable is an OPTIONAL seam letting stores that hold PII *derived* from events — the audit trail, saga state, snapshots, or an external sink — be erased alongside crypto-shredding. Register implementations via the WithSubjectStore option of NewDataEraser; Erase invokes each after key revocation and read-model redaction, records a SubjectErasureOutcome, and treats a per-store failure as non-fatal (symmetric with WithErasureHook).

Implementations MUST target READ-SIDE stores only, so subject-scoped deletion never violates the append-only event log. Use the built-in NewAuditSubjectEraser / NewSagaSubjectEraser / NewSnapshotSubjectEraser, or supply your own.

func NewAuditSubjectEraser added in v1.1.3 

func NewAuditSubjectEraser(store AuditStore) SubjectErasable

NewAuditSubjectEraser wraps an AuditStore as a SubjectErasable so DataEraser reaches the subject's audit trail (Article 17). The audit log records who/what/when in plaintext (actor, tenant, arbitrary metadata, raw error strings that can carry PII), which crypto-shredding the events does NOT touch. If the store implements the optional adapters.SubjectAuditPurger, EraseSubject deletes the subject's rows; otherwise it reports Skipped (never fails). Register with the WithSubjectStore option of NewDataEraser.

func NewIdempotencySubjectEraser added in v1.1.3 

func NewIdempotencySubjectEraser(store IdempotencyStore) SubjectErasable

NewIdempotencySubjectEraser wraps an IdempotencyStore as a SubjectErasable so DataEraser reaches the subject's idempotency records. Records are TTL-bounded and keyed by a command hash, but the optional Response payload can hold PII. If the store implements the optional adapters.SubjectIdempotencyPurger, EraseSubject deletes records whose AggregateID equals the subject id; otherwise it reports Skipped. Register with the WithSubjectStore option of NewDataEraser.

func NewOutboxSubjectEraser added in v1.1.3 

func NewOutboxSubjectEraser(store OutboxStore) SubjectErasable

NewOutboxSubjectEraser wraps an OutboxStore as a SubjectErasable so DataEraser reaches the subject's outbox rows. The default outbox path stores the ENCRYPTED payload (which crypto-shredding erases), but a route Transform that emits a decrypted/reshaped payload leaves an independent plaintext copy, and dead-lettered rows persist. If the store implements the optional adapters.SubjectOutboxPurger, EraseSubject deletes rows whose AggregateID equals the subject id; otherwise it reports Skipped. Register with the WithSubjectStore option of NewDataEraser.

func NewSagaSubjectEraser added in v1.1.3 

func NewSagaSubjectEraser(store SagaStore) SubjectErasable

NewSagaSubjectEraser wraps a SagaStore as a SubjectErasable so DataEraser reaches the subject's saga state. Sagas copy correlation/business data out of events into their own plaintext state, which crypto-shredding does NOT touch. If the store implements the optional adapters.SubjectSagaPurger, EraseSubject deletes sagas whose CorrelationID equals the subject id; otherwise it reports Skipped. Register with the WithSubjectStore option of NewDataEraser.

func NewSnapshotSubjectEraser added in v1.1.3 

func NewSnapshotSubjectEraser(adapter adapters.SnapshotAdapter) SubjectErasable

NewSnapshotSubjectEraser wraps a SnapshotAdapter as a SubjectErasable that deletes the snapshot of each stream in the subject's resolved footprint. Snapshots serialize decrypted aggregate STATE in plaintext, which crypto-shredding does NOT touch, so an un-deleted snapshot leaves the subject's PII recoverable. DeleteSnapshot is idempotent, so Erased counts the footprint streams whose snapshot was cleared. Register with the WithSubjectStore option of NewDataEraser.

type SubjectErasureOutcome added in v1.1.3 

type SubjectErasureOutcome struct {
	// Name identifies the store (e.g. "audit", "saga", "snapshot").
	Name string `json:"name"`

	// Erased is the number of rows/records removed for the subject. It is int64 to match
	// the purger APIs' row counts (e.g. sql.Result.RowsAffected), so large purges are
	// reported without truncation or overflow.
	Erased int64 `json:"erased"`

	// Skipped is true when the store could not target the subject (e.g. the
	// underlying adapter does not implement the optional purger sub-interface).
	Skipped bool `json:"skipped,omitempty"`

	// Err holds a non-fatal failure message, if any.
	Err string `json:"error,omitempty"`
}

SubjectErasureOutcome reports what one SubjectErasable did for a subject. It carries no PII and is safe to record on an ErasureResult / certificate.

type SubjectFootprint added in v1.1.3 

type SubjectFootprint struct {
	SubjectID         string
	Streams           []string       // sorted, de-duplicated
	StreamEventCounts map[string]int // tagged events per stream
	EventCount        int            // total tagged events
	KeyIDs            []string       // distinct encryption key ids on tagged events (sorted)

	// Partial is true when completeness cannot be proven — e.g. the store contains
	// untagged (legacy) events that could belong to the subject. Callers MUST treat
	// a partial footprint as incomplete (never a silent partial).
	Partial bool
}

SubjectFootprint describes the complete extent of a data subject's events. It drives complete-by-default export and erasure and doubles as an erasure preview.

type SubjectIdempotencyPurger added in v1.1.3 

type SubjectIdempotencyPurger = adapters.SubjectIdempotencyPurger

SubjectIdempotencyPurger is the optional IdempotencyStore extension for GDPR erasure of a subject's idempotency records (see NewIdempotencySubjectEraser).

type SubjectIndexAdapter added in v1.1.3 

type SubjectIndexAdapter interface {
	StreamsBySubject(ctx context.Context, subjectID string) ([]string, error)
}

SubjectIndexAdapter is an OPTIONAL extension that resolves a subject's streams from an index, avoiding a full scan. The event-store adapter MAY implement it, or an index can be injected into the resolver with WithResolverIndex; the resolver falls back to a scan otherwise.

type SubjectIndexWriter added in v1.1.3 

type SubjectIndexWriter interface {
	// IndexSubjects records that streamID contains events for each of subjectIDs.
	// It MUST be idempotent (indexing the same (subject, stream) twice is a no-op).
	IndexSubjects(ctx context.Context, streamID string, subjectIDs []string) error
}

SubjectIndexWriter is the OPTIONAL write side of a subject index: it records which streams touch a subject so SubjectIndexAdapter can later resolve them without a scan. Wire one into the store with WithSubjectIndexWriter (populated at append time) and/or populate history with BackfillSubjectIndex. A type that implements both interfaces is a complete, keep-in-sync subject index (see MemorySubjectIndex).

type SubjectOutboxPurger added in v1.1.3 

type SubjectOutboxPurger = adapters.SubjectOutboxPurger

SubjectOutboxPurger is the optional OutboxStore extension for GDPR erasure of a subject's outbox rows (see NewOutboxSubjectEraser).

type SubjectRedactable added in v1.1.3 

type SubjectRedactable interface {
	// RedactSubject removes or masks the subject's personal data from the read model.
	RedactSubject(ctx context.Context, subjectID string) error
	// ReadModelName identifies the read model for reporting on ErasureResult.
	ReadModelName() string
}

SubjectRedactable is implemented by a read model / projection that can redact a single data subject's PII in place — far cheaper than a full rebuild. DataEraser prefers it when available; register one with WithReadModelRedactor.

type SubjectResolver added in v1.1.3 

type SubjectResolver struct {
	// contains filtered or unexported fields
}

SubjectResolver resolves a subject id to its complete footprint across all streams, using a subject index when available or a scan otherwise.

func NewSubjectResolver added in v1.1.3 

func NewSubjectResolver(store *EventStore, opts ...SubjectResolverOption) *SubjectResolver

NewSubjectResolver creates a resolver for the given store.

func (*SubjectResolver) Resolve added in v1.1.3 

func (r *SubjectResolver) Resolve(ctx context.Context, subjectID string) (*SubjectFootprint, error)

Resolve returns the subject's footprint. It is read-only and therefore doubles as an erasure preview.

type SubjectResolverOption added in v1.1.3 

type SubjectResolverOption func(*SubjectResolver)

SubjectResolverOption configures a SubjectResolver.

func WithAuthoritativeIndex added in v1.1.3 

func WithAuthoritativeIndex() SubjectResolverOption

WithAuthoritativeIndex asserts that the injected index (WithResolverIndex) is complete — every stream touching a subject is recorded — so an index-backed resolve may report a non-partial footprint. Use it only when you can guarantee completeness: after BackfillSubjectIndex with no concurrent writes, or with a transactionally-consistent index. Without it, an index-backed resolve is honestly marked Partial.

func WithResolverBatchSize added in v1.1.3 

func WithResolverBatchSize(size int) SubjectResolverOption

WithResolverBatchSize sets the scan batch size (default 1000).

func WithResolverIndex added in v1.1.3 

func WithResolverIndex(idx SubjectIndexAdapter) SubjectResolverOption

WithResolverIndex injects a subject index (read side) the resolver prefers over both the adapter's own index and a full scan — turning resolution into O(subject's events).

By itself the index is treated as POSSIBLY INCOMPLETE: append-time index writes are best-effort (a failed write is logged, not fatal), so an index can silently drift behind the event log. A resolve that used the index therefore reports Partial=true unless you ALSO pass WithAuthoritativeIndex to assert the index is complete. This prevents an out-of-sync index from producing a falsely-complete footprint that would make Erase miss streams while certifying success.

type SubjectSagaPurger added in v1.1.3 

type SubjectSagaPurger = adapters.SubjectSagaPurger

SubjectSagaPurger is the optional SagaStore extension for GDPR erasure of a subject's saga state (see NewSagaSubjectEraser).

type SubjectTagger added in v1.1.3 

type SubjectTagger func(eventType string, data []byte, md Metadata) []string

SubjectTagger derives the data-subject identifier(s) a freshly-appended event concerns, from its serialized data and metadata. The returned ids are recorded in Metadata.Custom so the subject's complete footprint can later be resolved for GDPR export/erasure. Returning nil tags nothing (zero overhead). Configure via WithSubjectTagger.

data is the serialized event payload (the bytes being appended); applied at the single shared prepare-event hook, so it covers Append, SaveAggregate, and the outbox uniformly. Most taggers derive the subject from md (UserID/TenantID) or a known field within data.

type Subscription 

type Subscription interface {
	// Events returns the channel for receiving events.
	Events() <-chan StoredEvent

	// Close stops the subscription.
	Close() error

	// Err returns any error that caused the subscription to close.
	Err() error
}

Subscription represents an active event subscription.

type SubscriptionAdapter 

type SubscriptionAdapter interface {
	// LoadFromPosition loads events starting from a global position.
	LoadFromPosition(ctx context.Context, fromPosition uint64, limit int) ([]StoredEvent, error)

	// SubscribeAll subscribes to all events across all streams.
	SubscribeAll(ctx context.Context, fromPosition uint64) (<-chan StoredEvent, error)

	// SubscribeStream subscribes to events from a specific stream.
	SubscribeStream(ctx context.Context, streamID string, fromVersion int64) (<-chan StoredEvent, error)

	// SubscribeCategory subscribes to all events from streams in a category.
	SubscribeCategory(ctx context.Context, category string, fromPosition uint64) (<-chan StoredEvent, error)
}

SubscriptionAdapter provides methods for subscribing to event streams. This interface extends the basic EventStoreAdapter for subscription capabilities.

type SubscriptionOptions 

type SubscriptionOptions struct {
	// BufferSize is the size of the event channel buffer.
	// Default: 256
	BufferSize int

	// Filter optionally filters which events are delivered.
	Filter EventFilter

	// RetryOnError determines whether to retry on transient errors.
	// Default: true
	RetryOnError bool

	// RetryInterval is the time to wait between retries.
	// Default: 1 second
	RetryInterval time.Duration

	// MaxRetries is the maximum number of retry attempts.
	// Default: 5
	MaxRetries int
}

SubscriptionOptions configures a subscription.

func DefaultSubscriptionOptions 

func DefaultSubscriptionOptions() SubscriptionOptions

DefaultSubscriptionOptions returns the default subscription options.

type UnknownFilterFieldError added in v1.1.6 

type UnknownFilterFieldError struct{ Field string }

UnknownFilterFieldError names the unresolved filter field. Both errors.Is(err, ErrUnknownFilterField) and errors.Is(err, ErrInvalidQuery) hold.

func (*UnknownFilterFieldError) Error added in v1.1.6 

func (e *UnknownFilterFieldError) Error() string

func (*UnknownFilterFieldError) Is added in v1.1.6 

func (e *UnknownFilterFieldError) Is(target error) bool

func (*UnknownFilterFieldError) Unwrap added in v1.1.6 

func (e *UnknownFilterFieldError) Unwrap() error

Unwrap returns the wrapped sentinel (itself wrapping ErrInvalidQuery), so errors.Is and errors.As traverse the chain to both ErrUnknownFilterField and ErrInvalidQuery.

type UnregisteredEventTypeError added in v1.1.6 

type UnregisteredEventTypeError struct {
	StreamID  string
	EventType string
	Version   int64
}

UnregisteredEventTypeError reports an event type encountered during aggregate replay (LoadAggregate) that is not registered with the serializer, so it deserialized to the map fallback and the aggregate's ApplyEvent could not apply it — the event would be silently dropped from the rebuilt state. Register the type (RegisterEvents / RegisterAggregateEvents) or use WithStrictReplay to fail fast. errors.Is(err, ErrUnregisteredEventType) holds.

func (*UnregisteredEventTypeError) Error added in v1.1.6 

func (e *UnregisteredEventTypeError) Error() string

Error returns the error message.

func (*UnregisteredEventTypeError) Is added in v1.1.6 

func (e *UnregisteredEventTypeError) Is(target error) bool

Is reports whether this error matches the target error.

func (*UnregisteredEventTypeError) Unwrap added in v1.1.6 

func (e *UnregisteredEventTypeError) Unwrap() error

Unwrap returns the sentinel this error wraps, so errors.Is/errors.As traverse the chain.

type UpcastError 

type UpcastError struct {
	EventType   string
	FromVersion int
	ToVersion   int
	Cause       error
}

UpcastError provides detailed information about an upcasting failure.

func NewUpcastError 

func NewUpcastError(eventType string, fromVersion, toVersion int, cause error) *UpcastError

NewUpcastError creates a new UpcastError.

func (*UpcastError) Error 

func (e *UpcastError) Error() string

Error returns the error message.

func (*UpcastError) Is 

func (e *UpcastError) Is(target error) bool

Is reports whether this error matches the target error.

func (*UpcastError) Unwrap 

func (e *UpcastError) Unwrap() error

Unwrap returns the underlying cause for errors.Unwrap().

type Upcaster 

type Upcaster interface {
	// EventType returns the event type this upcaster handles.
	EventType() string

	// FromVersion returns the source schema version.
	FromVersion() int

	// ToVersion returns the target schema version. Must equal FromVersion() + 1.
	ToVersion() int

	// Upcast transforms event data from FromVersion to ToVersion.
	// Metadata is provided as read-only context (e.g., for tenant-specific defaults).
	Upcast(data []byte, metadata Metadata) ([]byte, error)
}

Upcaster transforms event data from one schema version to the next. Upcasters operate on raw bytes and are serializer-agnostic. Each upcaster handles exactly one version transition (FromVersion → ToVersion).

type UpcasterChain 

type UpcasterChain struct {
	// contains filtered or unexported fields
}

UpcasterChain is a thread-safe registry of upcasters that applies them in sequence. It validates that there are no gaps or duplicates in the version chain.

func NewUpcasterChain 

func NewUpcasterChain() *UpcasterChain

NewUpcasterChain creates a new empty UpcasterChain.

func (*UpcasterChain) HasUpcasters 

func (c *UpcasterChain) HasUpcasters(eventType string) bool

HasUpcasters reports whether any upcasters are registered for the given event type.

func (*UpcasterChain) LatestVersion 

func (c *UpcasterChain) LatestVersion(eventType string) int

LatestVersion returns the latest schema version for the given event type. Returns DefaultSchemaVersion if no upcasters are registered for the type.

func (*UpcasterChain) Register 

func (c *UpcasterChain) Register(u Upcaster) error

Register adds an upcaster to the chain. Returns an error if the upcaster's ToVersion != FromVersion + 1 or if an upcaster for the same event type and version transition already exists.

func (*UpcasterChain) RegisteredEventTypes 

func (c *UpcasterChain) RegisteredEventTypes() []string

RegisteredEventTypes returns a sorted list of event types that have upcasters registered.

func (*UpcasterChain) Upcast 

func (c *UpcasterChain) Upcast(eventType string, fromVersion int, data []byte, metadata Metadata) ([]byte, int, error)

Upcast transforms event data from fromVersion to the latest version. Returns the transformed data, the final version, and any error. If no upcasters exist for the event type or the data is already at the latest version, the original data is returned unchanged.

func (*UpcasterChain) Validate 

func (c *UpcasterChain) Validate() error

Validate checks the entire chain for gaps. For each event type, the upcasters must form a contiguous chain from the lowest FromVersion to the highest ToVersion.

type UpcastingSerializer 

type UpcastingSerializer struct {
	// contains filtered or unexported fields
}

UpcastingSerializer is a decorator that wraps any Serializer and applies upcasting during deserialization. Serialization passes through unchanged.

func NewUpcastingSerializer 

func NewUpcastingSerializer(inner Serializer, chain *UpcasterChain) *UpcastingSerializer

NewUpcastingSerializer creates a new UpcastingSerializer wrapping the given serializer.

func (*UpcastingSerializer) BinaryFormat added in v1.1.7 

func (s *UpcastingSerializer) BinaryFormat() bool

BinaryFormat reports whether the wrapped serializer emits a binary (non-JSON) format, forwarding to the inner serializer's BinaryFormatReporter if it implements one (msgpack/ protobuf report true; the JSON serializer reports false). Forwarding here lets mink.New's serializer/adapter compatibility guard see through this decorator with a single interface check instead of an unbounded chain walk. Satisfies mink.BinaryFormatReporter.

func (*UpcastingSerializer) Chain 

func (s *UpcastingSerializer) Chain() *UpcasterChain

Chain returns the upcaster chain.

func (*UpcastingSerializer) Deserialize 

func (s *UpcastingSerializer) Deserialize(data []byte, eventType string) (interface{}, error)

Deserialize converts bytes back to an event. If upcasters are registered for the event type, the data is upcasted from DefaultSchemaVersion before deserialization.

func (*UpcastingSerializer) DeserializeWithVersion 

func (s *UpcastingSerializer) DeserializeWithVersion(data []byte, eventType string, schemaVersion int, metadata Metadata) (interface{}, error)

DeserializeWithVersion converts bytes back to an event, upcasting from the specified schema version. Metadata is provided as read-only context to upcasters.

func (*UpcastingSerializer) Inner 

func (s *UpcastingSerializer) Inner() Serializer

Inner returns the wrapped serializer.

func (*UpcastingSerializer) Serialize 

func (s *UpcastingSerializer) Serialize(event interface{}) ([]byte, error)

Serialize converts an event to bytes. Pass-through to the inner serializer.

type ValidationError 

type ValidationError struct {
	// CommandType is the type of command that failed validation.
	CommandType string

	// Field is the field that failed validation (optional).
	Field string

	// Message describes the validation failure.
	Message string

	// Cause is the underlying error (optional).
	Cause error
}

ValidationError represents a command validation failure.

func NewValidationError 

func NewValidationError(cmdType, field, message string) *ValidationError

NewValidationError creates a new ValidationError.

func NewValidationErrorWithCause 

func NewValidationErrorWithCause(cmdType, field, message string, cause error) *ValidationError

NewValidationErrorWithCause creates a new ValidationError with an underlying cause.

func (*ValidationError) Error 

func (e *ValidationError) Error() string

Error returns the error message.

func (*ValidationError) Is 

func (e *ValidationError) Is(target error) bool

Is reports whether this error matches the target error.

func (*ValidationError) Unwrap 

func (e *ValidationError) Unwrap() error

Unwrap returns the underlying cause for errors.Unwrap().

type Validator 

type Validator interface {
	// Validate validates a command and returns validation errors.
	Validate(cmd Command) error
}

Validator provides command validation functionality.

type ValidatorFunc 

type ValidatorFunc func(cmd Command) error

ValidatorFunc is a function that implements Validator.

func (ValidatorFunc) Validate 

func (f ValidatorFunc) Validate(cmd Command) error

Validate implements Validator.

type VerificationReport added in v1.1.3 

type VerificationReport struct {
	SubjectID      string
	Verified       bool // true iff no residual PII and the footprint is complete
	EventsChecked  int
	RedactedEvents int // encrypted events whose key is revoked (unrecoverable)

	// ResidualEncrypted lists "stream@version" of encrypted events whose key is still
	// live (PII recoverable).
	ResidualEncrypted []string

	// ResidualRecoverable lists "stream@version" of encrypted events whose key is only
	// SOFT-revoked — decryption is blocked but the key can still be restored via
	// UnrevokeKey within its grace window, so the PII is NOT yet permanently erased.
	// A non-empty set forces Verified=false: a certificate must never claim final
	// erasure while data is still recoverable.
	ResidualRecoverable []string

	// ResidualCleartext lists "stream@version" of events with no encryption (possible
	// legacy cleartext PII that crypto-shredding cannot reach).
	ResidualCleartext []string

	// Partial mirrors the footprint: untagged events may exist beyond what was checked.
	Partial bool
}

VerificationReport summarizes whether a subject's PII has been erased across its event footprint.

type VersionSetter 

type VersionSetter interface {
	SetVersion(v int64)
}

VersionSetter is an optional interface that aggregates can implement to allow the EventStore to set their version during loading. This is used for optimistic concurrency control in SaveAggregate. AggregateBase implements this interface.

type VersionedAggregate 

type VersionedAggregate interface {
	Aggregate

	// OriginalVersion returns the version when the aggregate was loaded.
	OriginalVersion() int64
}

VersionedAggregate provides versioning information for optimistic concurrency.

Source Files

View all Source files

Directories

Path Synopsis
Package adapters provides interfaces for event store backends.
 Package adapters provides interfaces for event store backends.
memory
Package memory provides an in-memory implementation of the event store adapter.
 Package memory provides an in-memory implementation of the event store adapter.
postgres
Package postgres provides a PostgreSQL implementation of the event store adapter.
 Package postgres provides a PostgreSQL implementation of the event store adapter.
cli
commands
Package commands provides the CLI command implementations for mink.
 Package commands provides the CLI command implementations for mink.
config
Package config provides configuration management for the mink CLI.
 Package config provides configuration management for the mink CLI.
styles
Package styles provides consistent styling for the go-mink CLI.
 Package styles provides consistent styling for the go-mink CLI.
ui
Package ui provides reusable UI components for the go-mink CLI.
 Package ui provides reusable UI components for the go-mink CLI.
cmd
mink command
mink is the command-line interface for the go-mink event sourcing library.
 mink is the command-line interface for the go-mink event sourcing library.
Package encryption defines interfaces and types for field-level encryption in event sourcing.
 Package encryption defines interfaces and types for field-level encryption in event sourcing.
kms
Package kms provides an AWS KMS encryption provider for field-level encryption.
 Package kms provides an AWS KMS encryption provider for field-level encryption.
local
Package local provides an in-memory AES-256-GCM encryption provider for testing.
 Package local provides an in-memory AES-256-GCM encryption provider for testing.
providertest
Package providertest provides shared test helpers for encryption.Provider implementations.
 Package providertest provides shared test helpers for encryption.Provider implementations.
vault
Package vault provides a HashiCorp Vault Transit encryption provider for field-level encryption.
 Package vault provides a HashiCorp Vault Transit encryption provider for field-level encryption.
examples
audit command
Package main demonstrates the command Audit Logging middleware.
 Package main demonstrates the command Audit Logging middleware.
basic command
Package main demonstrates the basic usage of go-mink for event sourcing.
 Package main demonstrates the basic usage of go-mink for event sourcing.
cqrs command
Package main demonstrates the CQRS and Command Bus features of go-mink (Phase 2).
 Package main demonstrates the CQRS and Command Bus features of go-mink (Phase 2).
cqrs-postgres command
Package main demonstrates Phase 2 CQRS & Command Bus features with PostgreSQL.
 Package main demonstrates Phase 2 CQRS & Command Bus features with PostgreSQL.
encryption command
Package main demonstrates field-level encryption in go-mink.
 Package main demonstrates field-level encryption in go-mink.
export command
Package main demonstrates GDPR data export (right to access / data portability) in go-mink.
 Package main demonstrates GDPR data export (right to access / data portability) in go-mink.
feed-filter command
Package main demonstrates filtered feed reads — reading the global event feed by an INDEXED axis for introspection.
 Package main demonstrates filtered feed reads — reading the global event feed by an INDEXED axis for introspection.
full-ecommerce command
Package main demonstrates a complete e-commerce order fulfillment system using go-mink.
 Package main demonstrates a complete e-commerce order fulfillment system using go-mink.
metrics command
Example: Metrics Middleware
 Example: Metrics Middleware
msgpack command
Example: MessagePack Serializer
 Example: MessagePack Serializer
projections command
Package main demonstrates the projection and read model features of go-mink.
 Package main demonstrates the projection and read model features of go-mink.
protobuf command
Example: Protocol Buffers Serializer
 Example: Protocol Buffers Serializer
reencrypt-inplace command
Package main demonstrates in-place stream re-encryption — the historical backfill for a store that enabled field encryption AFTER it already had plaintext data.
 Package main demonstrates in-place stream re-encryption — the historical backfill for a store that enabled field encryption AFTER it already had plaintext data.
sagas command
Example: Saga (Process Manager) Pattern
 Example: Saga (Process Manager) Pattern
tracing command
Example: Distributed Tracing with OpenTelemetry
 Example: Distributed Tracing with OpenTelemetry
versioning command
Package main demonstrates event versioning and upcasting in go-mink.
 Package main demonstrates event versioning and upcasting in go-mink.
middleware
metrics
Package metrics provides Prometheus metrics integration for mink.
 Package metrics provides Prometheus metrics integration for mink.
tracing
Package tracing provides OpenTelemetry integration for mink.
 Package tracing provides OpenTelemetry integration for mink.
outbox
kafka
Package kafka provides a Kafka publisher for the outbox pattern.
 Package kafka provides a Kafka publisher for the outbox pattern.
sns
Package sns provides an AWS SNS publisher for the outbox pattern.
 Package sns provides an AWS SNS publisher for the outbox pattern.
webhook
Package webhook provides a webhook publisher for the outbox pattern.
 Package webhook provides a webhook publisher for the outbox pattern.
serializer
msgpack
Package msgpack provides a MessagePack serializer implementation for mink.
 Package msgpack provides a MessagePack serializer implementation for mink.
protobuf
Package protobuf provides a Protocol Buffers serializer for mink events.
 Package protobuf provides a Protocol Buffers serializer for mink events.
testing
assertions
Package assertions provides event assertion utilities for testing event-sourced systems.
 Package assertions provides event assertion utilities for testing event-sourced systems.
bdd
Package bdd provides BDD-style test fixtures for event-sourced aggregates.
 Package bdd provides BDD-style test fixtures for event-sourced aggregates.
benchmarks
Package benchmarks provides a shared benchmark suite for EventStoreAdapter implementations.
 Package benchmarks provides a shared benchmark suite for EventStoreAdapter implementations.
containers
Package containers provides connection helpers for integration testing against infrastructure that is already running.
 Package containers provides connection helpers for integration testing against infrastructure that is already running.
projections
Package projections provides testing utilities for projection development.
 Package projections provides testing utilities for projection development.
sagas
Package sagas provides testing utilities for saga (process manager) development.
 Package sagas provides testing utilities for saga (process manager) development.
testutil
Package testutil provides test utilities and fixtures for go-mink.
 Package testutil provides test utilities and fixtures for go-mink.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.