API Design

Core Package (`mink`)

Event Store

package go-mink

// EventStore is the main entry point
type EventStore struct {
    adapter     EventStoreAdapter
    serializer  Serializer
    projections *ProjectionEngine
    middleware  []Middleware
}

// New creates a new event store
func New(adapter EventStoreAdapter, opts ...Option) *EventStore

// Options
func WithSerializer(s Serializer) Option
func WithMiddleware(m ...Middleware) Option
func WithSnapshots(adapter SnapshotAdapter, policy SnapshotPolicy) Option
func WithProjections(engine *ProjectionEngine) Option
func WithLogger(logger Logger) Option
func WithMetrics(provider MetricsProvider) Option

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

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

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

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

// Append options
func ExpectVersion(v int64) AppendOption
func WithMetadata(m Metadata) AppendOption

Aggregates

// Aggregate interface for domain objects
type Aggregate interface {
    AggregateID() string
    AggregateType() string
    Version() int64
    ApplyEvent(event interface{}) error
    UncommittedEvents() []interface{}
    ClearUncommittedEvents()
}

// AggregateBase provides default implementation
type AggregateBase struct {
    id               string
    version          int64
    uncommittedEvents []interface{}
}

func (a *AggregateBase) AggregateID() string { return a.id }
func (a *AggregateBase) SetID(id string)     { a.id = id }
func (a *AggregateBase) Version() int64      { return a.version }
func (a *AggregateBase) SetVersion(v int64)  { a.version = v }

func (a *AggregateBase) Apply(event interface{}) {
    a.uncommittedEvents = append(a.uncommittedEvents, event)
}

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

func (a *AggregateBase) ClearUncommittedEvents() {
    a.uncommittedEvents = nil
}

Commands

// Command represents intent to change state
type Command interface {
    CommandType() string
    Validate() error
}

// CommandBase provides common command functionality
type CommandBase struct {
    id            string
    correlationID string
    causationID   string
    tenantID      string
    metadata      map[string]string
}

// Accessors for command metadata
func (c *CommandBase) GetID() string
func (c *CommandBase) SetID(id string)
func (c *CommandBase) GetCorrelationID() string
func (c *CommandBase) SetCorrelationID(id string)
func (c *CommandBase) GetCausationID() string
func (c *CommandBase) SetCausationID(id string)
func (c *CommandBase) GetTenantID() string
func (c *CommandBase) SetTenantID(id string)
func (c *CommandBase) GetMetadata() map[string]string
func (c *CommandBase) SetMetadata(m map[string]string)

// CommandResult contains the result of command execution
type CommandResult struct {
    AggregateID string
    Version     int64
    Events      []interface{}
    Data        interface{}
}

func NewSuccessResult(aggregateID string, version int64) CommandResult

// IdempotentCommand provides custom idempotency keys
type IdempotentCommand interface {
    Command
    IdempotencyKey() string
}

Command Bus

// CommandBus routes commands to handlers
type CommandBus struct {
    handlers   map[string]CommandHandler
    middleware []CommandMiddleware
}

func NewCommandBus() *CommandBus

// Registration
func (b *CommandBus) Register(handler CommandHandler)
func (b *CommandBus) RegisterFunc(cmdType string, fn CommandHandlerFunc)

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

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

// CommandHandler interface
type CommandHandler interface {
    CommandType() string
    Handle(ctx context.Context, cmd Command) (CommandResult, error)
}

// CommandHandlerFunc for function-based handlers
type CommandHandlerFunc func(ctx context.Context, cmd Command) (CommandResult, error)

// CommandMiddleware wraps command handling
type CommandMiddleware func(CommandHandlerFunc) CommandHandlerFunc

Generic Handler

// GenericHandler provides type-safe command handling
type GenericHandler[T Command] struct {
    fn func(ctx context.Context, cmd T) (CommandResult, error)
}

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

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

Aggregate Handler

// AggregateHandler combines load/save with command handling
type AggregateHandler[C Command, A Aggregate] struct {
    store      AggregateStore
    idFunc     func(C) string
    factory    func() A
    handleFunc func(ctx context.Context, agg A, cmd C) error
}

func NewAggregateHandler[C Command, A Aggregate](
    store AggregateStore,
    idFunc func(C) string,
    factory func() A,
    handleFunc func(ctx context.Context, agg A, cmd C) error,
) *AggregateHandler[C, A]

Built-in Middleware

// Validation middleware - calls cmd.Validate()
func ValidationMiddleware() CommandMiddleware

// Recovery middleware - catches panics and returns a *PanicError.
// PanicError.Value is PanicValueSummary(r) — the panic value's type, plus its
// message only for a Go runtime error — and PanicError.CommandData is only
// {"commandType": ..., "aggregateId": ...}; PanicError.Error() never prints more
// than the summary. RecoveryMiddleware() keeps its original no-arg signature;
// RecoveryMiddlewareWithOptions() is the configurable form (identical with no
// options): WithPanicValueCapture() keeps the raw panic value in Value and
// WithPanicCommandCapture() records the full JSON-serialized command (both opt-in:
// handlers panic with the payloads they process, and commands carry PII/secrets).
func RecoveryMiddleware() CommandMiddleware
func RecoveryMiddlewareWithOptions(opts ...RecoveryOption) CommandMiddleware
type RecoveryOption func(*recoveryConfig)
func WithPanicCommandCapture() RecoveryOption
func WithPanicValueCapture() RecoveryOption

// Payload-safe description of a recovered panic value: "<type>: <message>" for a
// Go runtime error, "redacted panic value of type <type>" for anything else
// (strings included), "redacted panic value of type <type> (nil)" for a typed nil
// pointer (its Error() is never called), "nil panic value" for nil. Never panics.
// Used by RecoveryMiddleware, the projection engine and UpcasterChain; reusable by
// application-level recover().
func PanicValueSummary(v interface{}) string

// Logging middleware - logs command execution
func LoggingMiddleware(logger Logger) CommandMiddleware

// Metrics middleware - records command metrics
func MetricsMiddleware(metrics MetricsRecorder) CommandMiddleware

// Timeout middleware - adds context timeout
func TimeoutMiddleware(timeout time.Duration) CommandMiddleware

// Retry middleware - retries on transient failures
func RetryMiddleware(maxAttempts int, initialDelay time.Duration) CommandMiddleware

// Correlation ID middleware - sets/generates correlation ID (nil generator = random v4 UUID)
func CorrelationIDMiddleware(generator func() string) CommandMiddleware

// Causation ID middleware - tracks causation chain
func CausationIDMiddleware() CommandMiddleware

// Tenant middleware - uses the tenant already on the context (set server-side via
// WithTenantID by the auth layer), otherwise extracts it from the command; required=true
// rejects a command with no tenant. Never derive the tenant from a client-controlled header.
func TenantMiddleware(extractor func(Command) string, required bool) CommandMiddleware

// Idempotency middleware - prevents duplicate processing
func IdempotencyMiddleware(config IdempotencyConfig) CommandMiddleware

Idempotency Store

// IdempotencyStore tracks processed commands
type IdempotencyStore interface {
    Store(ctx context.Context, key string, response []byte, expiration time.Duration) error
    Get(ctx context.Context, key string) ([]byte, error)
    Close() error
}

// IdempotencyConfig configures idempotency middleware
type IdempotencyConfig struct {
    Store          IdempotencyStore     // Required: storage backend
    TTL            time.Duration        // Result expiration (default: 24h)
    ReservationTTL time.Duration        // In-flight reservation lease; bounds how long a duplicate is blocked (default: 5m)
    KeyGenerator   func(Command) string // Custom key generator (default: GetIdempotencyKey); "" = no guarantee, pass-through
    Scope          func(ctx context.Context, cmd Command) string // Principal scope, stored as "<scope>|<key>" (default: DefaultIdempotencyScope)
    StoreErrors    bool                 // Replay stored failures instead of retrying (default: false)
    FailClosed     bool                 // Fail the command on store outage (default: false = fail-open)
    SkipCommands   []string             // Command types that bypass the idempotency check
}

func DefaultIdempotencyConfig(store IdempotencyStore) IdempotencyConfig
func GenerateIdempotencyKey(cmd Command) string // "" when the command cannot be JSON-serialized

// Scoping and bounding of the stored key.
const MaxIdempotencyKeyLength = 255                                  // longer keys become "<len:scope|><CommandType>:sha256:<hex>" (digest covers the scoped key)
const MaxIdempotencyFieldLength = 255                                // NewIdempotencyRecord truncates CommandType / AggregateID to this many bytes (rune boundary)
func DefaultIdempotencyScope(ctx context.Context, _ Command) string  // TenantIDFromContext(ctx); "" when no tenant
func EffectiveIdempotencyKey(cmdType, scope, key string) string      // "<len(scope)>:<scope>|<key>" (e.g. "8:tenant-a|req-1"); key verbatim when scope == ""

Audit Logging

// AuditStore persists and queries the command audit trail.
type AuditStore interface {
    Append(ctx context.Context, entry *AuditEntry) error
    Find(ctx context.Context, q AuditQuery) ([]*AuditEntry, error) // honors Order/Limit/Offset
    Count(ctx context.Context, q AuditQuery) (int64, error)        // ignores Limit/Offset
    Cleanup(ctx context.Context, olderThan time.Duration) (int64, error)
    Initialize(ctx context.Context) error
    Close() error
}

// AuditEntry is one immutable record per dispatched command.
type AuditEntry struct {
    ID, CommandType, CommandID, AggregateID            string
    Actor, TenantID, CorrelationID, CausationID, Error string
    Version, DurationMs                                int64
    Timestamp                                          time.Time
    Success                                            bool
    Metadata                                           map[string]string
}

// AuditQuery filters and paginates the trail (zero value = all matching).
type AuditQuery struct {
    CommandType, Actor, TenantID, AggregateID, CorrelationID string
    From, To                                  time.Time
    Success                                   *bool      // nil = both
    Limit, Offset                             int
    Order                                     AuditOrder // AuditOrderTimestampDesc (default) | ...Asc
}

// AuditConfig configures the middleware.
type AuditConfig struct {
    Store           AuditStore
    ActorFunc       ActorFunc // nil -> ActorFromContext
    SkipCommands    []string
    FailClosed      bool      // default false = fail-open (auditing never breaks commands; drops are logged)
    IncludeMetadata bool
    MetadataFilter  func(map[string]string) map[string]string          // drop/mask captured metadata keys (plaintext trail)
    Logger          Logger                                             // fail-open drop warnings; nil -> slog.Default()
    OnError         func(ctx context.Context, entry *AuditEntry, err error) // every Store.Append failure, both modes
}

// String fields stored in VARCHAR(255) columns (CommandType, CommandID, AggregateID,
// Actor, TenantID, CorrelationID, CausationID) are truncated to this many bytes
// (rune boundary) before Append, so an over-long value cannot fail the row.
const MaxAuditFieldLength = 255

func AuditMiddleware(config AuditConfig) Middleware
func DefaultAuditConfig(store AuditStore) AuditConfig

// Actor context helpers.
func WithActor(ctx context.Context, actor string) context.Context
func ActorFromContext(ctx context.Context) string

See the Audit Logging guide for usage, failure semantics, querying, and the PostgreSQL schema.

Events

// Event represents a stored event
type Event struct {
    ID             string
    StreamID       string
    Type           string
    Data           interface{}
    Metadata       Metadata
    Version        int64
    GlobalPosition uint64
    Timestamp      time.Time
}

// Metadata carries event context
type Metadata struct {
    CorrelationID string
    CausationID   string
    UserID        string
    TenantID      string
    Custom        map[string]string
}

// EventRegistry maps type names to Go types
type EventRegistry struct {
    types map[string]reflect.Type
}

func NewEventRegistry() *EventRegistry

func (r *EventRegistry) Register(eventType string, example interface{}) // first registration of a name wins
func (r *EventRegistry) RegisterAll(events ...interface{}) // Uses struct name
func (r *EventRegistry) Lookup(eventType string) (reflect.Type, bool)
func (r *EventRegistry) Conflicts() []string // names a later Register tried to rebind to a different type (nil when none)

Projections

// Projection transforms events to read models
type Projection interface {
    Name() string
    HandledEvents() []string
}

// InlineProjection runs in same transaction
type InlineProjection interface {
    Projection
    Apply(ctx context.Context, tx Transaction, event Event) error
}

// AsyncProjection runs in background
type AsyncProjection interface {
    Projection
    Apply(ctx context.Context, events []Event) error
    Checkpoint() uint64
    SetCheckpoint(pos uint64) error
}

// ProjectionEngine manages all projections
type ProjectionEngine struct {
    inline []InlineProjection
    async  []asyncRunner
    store  EventStore
}

func NewProjectionEngine(store *EventStore) *ProjectionEngine

func (e *ProjectionEngine) RegisterInline(p InlineProjection)
func (e *ProjectionEngine) RegisterAsync(p AsyncProjection, opts AsyncOptions)
func (e *ProjectionEngine) Start(ctx context.Context) error
func (e *ProjectionEngine) Stop() error
func (e *ProjectionEngine) Rebuild(ctx context.Context, name string) error

// AsyncOptions configures async projection processing
type AsyncOptions struct {
    BatchSize    int
    BatchTimeout time.Duration
    Workers      int
    RetryPolicy  RetryPolicy
}

Subscriptions

// Subscribe to event streams
type Subscription interface {
    Events() <-chan StoredEvent // Closed when the subscription ends
    Close() error
    Err() error // Failure reason: nil after a clean Close(); the context error if your context was canceled
}

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

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

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

// Every subscription (catch-up and polling) delivers field-encrypted events
// DECRYPTED, like Load and projections, and filters on plaintext; a hard,
// unhandled decryption error ends the subscription with Err() set rather than
// delivering ciphertext.
//
// Example usage. Errors are not delivered on a channel: range over Events()
// (which closes when the subscription stops) and then inspect Err().
sub, err := store.SubscribeAll(ctx, 0)
if err != nil {
    log.Fatalf("subscribe: %v", err)
}
defer sub.Close()

for event := range sub.Events() {
    fmt.Printf("Event: %s\n", event.Type)
}
if err := sub.Err(); err != nil {
    log.Printf("Subscription stopped: %v\n", err)
}

Read Model Repository

// Generic repository for read models
type Repository[T any] interface {
    Get(ctx context.Context, id string) (*T, error)
    GetAll(ctx context.Context) ([]*T, error)
    Find(ctx context.Context, query Query) ([]*T, error)
    FindOne(ctx context.Context, query Query) (*T, error)
    Save(ctx context.Context, model *T) error
    Delete(ctx context.Context, id string) error
    Count(ctx context.Context, query Query) (int64, error)
}

// Query builder
type Query struct {
    filters  []Filter
    orderBy  []OrderBy
    limit    int
    offset   int
}

func NewQuery() *Query
func (q *Query) Where(field string, op string, value interface{}) *Query
func (q *Query) And(field string, op string, value interface{}) *Query
func (q *Query) Or(field string, op string, value interface{}) *Query
func (q *Query) OrderBy(field string, desc bool) *Query
func (q *Query) Limit(n int) *Query
func (q *Query) Offset(n int) *Query

// Usage
orders, _ := repo.Find(ctx,
    go-mink.NewQuery().
        Where("status", "=", "pending").
        And("total", ">", 100).
        OrderBy("created_at", true).
        Limit(10),
)

Event Versioning & Upcasting

// Upcaster transforms event data from one schema version to the next
type Upcaster interface {
    EventType() string
    FromVersion() int
    ToVersion() int   // Must equal FromVersion() + 1
    Upcast(data []byte, metadata Metadata) ([]byte, error)
}

// UpcasterChain manages and applies upcasters in sequence
func NewUpcasterChain() *UpcasterChain
func (c *UpcasterChain) Register(u Upcaster) error
func (c *UpcasterChain) Validate() error
func (c *UpcasterChain) Upcast(eventType string, fromVersion int, data []byte, metadata Metadata) ([]byte, int, error)
func (c *UpcasterChain) HasUpcasters(eventType string) bool
func (c *UpcasterChain) LatestVersion(eventType string) int
func (c *UpcasterChain) RegisteredEventTypes() []string

// EventStore integration
func WithUpcasters(chain *UpcasterChain) Option
func (s *EventStore) RegisterUpcasters(upcasters ...Upcaster) error

// Schema version helpers
func GetSchemaVersion(m Metadata) int
func SetSchemaVersion(m Metadata, version int) Metadata
func SerializeEventWithVersion(serializer Serializer, event interface{}, metadata Metadata, schemaVersion int) (EventData, error)

// UpcastingSerializer wraps any Serializer with upcasting on deserialize
func NewUpcastingSerializer(inner Serializer, chain *UpcasterChain) *UpcastingSerializer
func (s *UpcastingSerializer) Serialize(event interface{}) ([]byte, error)
func (s *UpcastingSerializer) Deserialize(data []byte, eventType string) (interface{}, error)
func (s *UpcastingSerializer) DeserializeWithVersion(data []byte, eventType string, schemaVersion int, metadata Metadata) (interface{}, error)
func (s *UpcastingSerializer) Inner() Serializer
func (s *UpcastingSerializer) Chain() *UpcasterChain

// SchemaRegistry tracks event schemas and checks compatibility
func NewSchemaRegistry() *SchemaRegistry
func (r *SchemaRegistry) Register(eventType string, schema SchemaDefinition) error
func (r *SchemaRegistry) GetSchema(eventType string, version int) (*SchemaDefinition, error)
func (r *SchemaRegistry) GetLatestVersion(eventType string) (int, error)
func (r *SchemaRegistry) CheckCompatibility(eventType string, oldVersion, newVersion int) (SchemaCompatibility, error)
func (r *SchemaRegistry) RegisteredEventTypes() []string

// Schema types
type SchemaCompatibility int
const (
    SchemaFullyCompatible    SchemaCompatibility = iota
    SchemaBackwardCompatible
    SchemaForwardCompatible
    SchemaBreaking
)

type SchemaDefinition struct {
    Version      int
    Fields       []FieldDefinition
    JSONSchema   json.RawMessage
    RegisteredAt time.Time
}

type FieldDefinition struct {
    Name     string
    Type     string
    Required bool
}

// Versioning errors (a panic inside a user upcaster is recovered by
// UpcasterChain.Upcast and returned as an *UpcastError carrying only
// PanicValueSummary of the panic value, never propagated)
var ErrUpcastFailed       = errors.New("mink: upcast failed")
var ErrSchemaVersionGap   = errors.New("mink: schema version gap")
var ErrIncompatibleSchema = errors.New("mink: incompatible schema")
var ErrSchemaNotFound     = errors.New("mink: schema not found")

Field-Level Encryption

package encryption

// Provider abstracts crypto operations for field-level encryption.
type Provider interface {
    Encrypt(ctx context.Context, keyID string, plaintext []byte) (ciphertext []byte, err error)
    Decrypt(ctx context.Context, keyID string, ciphertext []byte) (plaintext []byte, err error)
    GenerateDataKey(ctx context.Context, keyID string) (*DataKey, error)
    DecryptDataKey(ctx context.Context, keyID string, encryptedKey []byte) ([]byte, error)
    Close() error
}

type DataKey struct {
    Plaintext  []byte
    Ciphertext []byte
    KeyID      string
}

// Sentinel errors
var ErrEncryptionFailed = errors.New("mink: encryption failed")
var ErrDecryptionFailed = errors.New("mink: decryption failed")
var ErrKeyNotFound      = errors.New("mink: encryption key not found")
var ErrKeyRevoked       = errors.New("mink: encryption key revoked")
var ErrProviderClosed   = errors.New("mink: encryption provider closed")

// Typed errors (implement Is(), Unwrap())
type EncryptionError struct { Operation string; KeyID string; Field string; Cause error }
type KeyNotFoundError struct { KeyID string }
type KeyRevokedError struct { KeyID string }

func NewEncryptionError(keyID, field string, cause error) *EncryptionError
func NewDecryptionError(keyID, field string, cause error) *EncryptionError
func NewKeyNotFoundError(keyID string) *KeyNotFoundError
func NewKeyRevokedError(keyID string) *KeyRevokedError
func ClearBytes(b []byte)

// --- Providers ---

// Local (encryption/local)
func local.New(opts ...local.Option) *local.Provider
func local.WithKey(keyID string, key []byte) local.Option
func (p *local.Provider) AddKey(keyID string, key []byte) error
func (p *local.Provider) RevokeKey(keyID string) error

// AWS KMS (encryption/kms) — stamp key ARNs, never aliases
func kms.New(opts ...kms.Option) *kms.Provider
func kms.WithKMSClient(client kms.KMSClient) kms.Option
func kms.WithPendingDeletionWindow(days int32) kms.Option   // 7–30 (clamped)
func kms.WithRevocationTimeout(d time.Duration) kms.Option  // bounds context-free revocation calls; default kms.DefaultRevocationTimeout (30s)
// With a client implementing kms.KMSRevocationClient (+ kms.KMSDeletionCanceller for UnrevokeKey),
// *kms.Provider satisfies encryption.Revocable, StatefulRevocable and RecoverableRevocable:
func (p *kms.Provider) RevokeKey(keyID string) error
func (p *kms.Provider) IsRevoked(keyID string) (bool, error)
func (p *kms.Provider) RevocationState(keyID string) (encryption.RevocationState, error) // pending deletion = SoftRevoked
func (p *kms.Provider) SoftRevokeKey(keyID string, graceWindow time.Duration) error
func (p *kms.Provider) UnrevokeKey(keyID string) error                                    // state-driven: CancelKeyDeletion + EnableKey, EnableKey alone for a merely-disabled key; *kms.KeyPermanentlyDeletedError once deleted
var kms.ErrKeyPermanentlyDeleted                                                           // sentinel; KeyPermanentlyDeletedError{KeyID} also Is(encryption.ErrKeyRevoked)

// Vault Transit (encryption/vault) — key names validated before any request (non-empty, no "/", no control bytes, not "." / ".."), then url.PathEscape'd
func vault.New(opts ...vault.Option) *vault.Provider
func vault.WithVaultClient(client vault.VaultClient) vault.Option
func vault.WithRevocationTimeout(d time.Duration) vault.Option   // bounds RevokeKey / IsRevoked and the decrypt-path probe; default vault.DefaultRevocationTimeout (30s)

// --- Root package config ---

// FieldEncryptionConfig configures per-event-type field encryption
type FieldEncryptionConfig struct { /* ... */ }
type EncryptionOption func(*FieldEncryptionConfig)

func NewFieldEncryptionConfig(opts ...EncryptionOption) *FieldEncryptionConfig
func WithEncryptionProvider(p encryption.Provider) EncryptionOption
func WithDefaultKeyID(keyID string) EncryptionOption
func WithEncryptedFields(eventType string, fields ...string) EncryptionOption
func WithTenantKeyResolver(resolver func(tenantID string) string) EncryptionOption
func WithSubjectKeyResolver(resolver func(subjectID string) string) EncryptionOption // alias of WithTenantKeyResolver (last-applied wins)
func WithRequireKeyResolution() EncryptionOption                                     // resolver miss → EncryptionError{Cause: *KeyResolutionError} instead of the default key
func WithDecryptionErrorHandler(handler func(err error, eventType string, metadata Metadata) error) EncryptionOption
func (c *FieldEncryptionConfig) Validate() error                                     // first malformed field path (wraps ErrInvalidEncryptedFieldPath); overlapping paths are valid

// EventStore options
func WithFieldEncryption(config *FieldEncryptionConfig) Option
func WithCallerSchemaVersion() Option // honor an in-range caller $schema_version for a type with upcasters (default: dropped, latest stamped)
func WithCallerSubjectTags() Option   // merge caller $subjects with the tagger's output (default: tagger output replaces them)

// Metadata helpers
func IsEncrypted(m Metadata) bool            // $encrypted_fields present — the read-path signal (route through decryption)
func HasEncryptionEnvelope(m Metadata) bool  // complete envelope (fields + key id + wrapped DEK) — what every key-acting path (erasure, retention, guard, verify) uses
func SanitizeReservedMetadata(m Metadata) Metadata // copy-on-write strip of the four envelope keys
func GetEncryptedFields(m Metadata) []string
func GetEncryptionKeyID(m Metadata) string

Middleware

// Middleware wraps event store operations
type Middleware func(next Handler) Handler

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

// Built-in middleware
func LoggingMiddleware(logger Logger) Middleware
func MetricsMiddleware(provider MetricsProvider) Middleware
func TracingMiddleware(tracer Tracer) Middleware
func RetryMiddleware(policy RetryPolicy) Middleware
func ValidationMiddleware() Middleware

// Usage
store := go-mink.New(adapter,
    go-mink.WithMiddleware(
        go-mink.LoggingMiddleware(logger),
        go-mink.MetricsMiddleware(prometheus.NewProvider()),
        go-mink.TracingMiddleware(otel.Tracer("go-mink")),
    ),
)

Multi-tenancy

// TenantContext adds tenant awareness
type TenantContext struct {
    TenantID string
    store    *EventStore
}

func (s *EventStore) ForTenant(tenantID string) *TenantContext

// Tenant isolation strategies
type TenantStrategy int

const (
    // Shared tables with tenant_id column
    SharedTable TenantStrategy = iota

    // Separate schema per tenant
    SchemaPerTenant

    // Separate database per tenant
    DatabasePerTenant
)

// Configuration
config := go-mink.Config{
    MultiTenancy: go-mink.MultiTenancyConfig{
        Strategy: go-mink.SchemaPerTenant,
        SchemaPrefix: "tenant_",
    },
}

// Usage
tenantStore := store.ForTenant("acme-corp")
tenantStore.Append(ctx, "order-123", events)

Errors

// Sentinel errors
var (
    ErrStreamNotFound      = errors.New("stream not found")
    ErrConcurrencyConflict = errors.New("concurrency conflict")
    ErrEventNotFound       = errors.New("event not found")
    ErrProjectionNotFound  = errors.New("projection not found")
    ErrSerializationFailed = errors.New("serialization failed")
    ErrAdapterNotSupported = errors.New("adapter not supported")
)

// Typed errors with details
type ConcurrencyError struct {
    StreamID        string
    ExpectedVersion int64
    ActualVersion   int64
}

func (e *ConcurrencyError) Error() string {
    return fmt.Sprintf("concurrency conflict on stream %s: expected %d, got %d",
        e.StreamID, e.ExpectedVersion, e.ActualVersion)
}

// Error checking
if errors.Is(err, go-mink.ErrConcurrencyConflict) {
    // Handle conflict
}

var concErr *go-mink.ConcurrencyError
if errors.As(err, &concErr) {
    // Access error details
    fmt.Printf("Stream: %s\n", concErr.StreamID)
}

// Added by the security hardening (all "mink: "-prefixed, typed errors implement Is/Unwrap)
var ErrSagaTypeMismatch            // *SagaTypeMismatchError{CorrelationID, SagaID, ExpectedType, ActualType}
var ErrResidualCountUnsupported    // a subject store cannot count residual rows (SubjectResidualCounter)
var ErrRetentionUnscopedPolicy     // a RetentionPolicy with no matchers at all
var ErrRetentionSharedKey          // *RetentionSharedKeyError{KeyID, OutOfScope}: shred guard refused a shared key; Error() prints the count only (KeyID is programmatic, never logged)
var ErrRetentionUnencryptedMatches // a Shred policy matched events with no encryption envelope
var ErrRetentionMaxScanNeedsCheckpoint // WithRetentionMaxScan configured without WithRetentionCheckpoint (non-fatal; sweep runs unbounded)
var ErrAnonymizerSecretRequired    // Anonymizer.Validate() on an empty HMAC secret
var ErrKeyResolutionFailed         // *KeyResolutionError{EventType, TenantID, SubjectID}: WithRequireKeyResolution and the resolver yielded no key
var ErrInvalidEncryptedFieldPath   // FieldEncryptionConfig.Validate / first append of the affected type: malformed WithEncryptedFields path
var SharedKeyError                 // Error() prints counts only (shared keys, other subjects); SharedKeys / OtherSubjects are programmatic
// Sub-packages: postgres.ErrSubjectTagMalformed (*SubjectTagMalformedError{SubjectID, Rows, Streams}),
// kms.ErrKeyPermanentlyDeleted (*KeyPermanentlyDeletedError{KeyID}), webhook.ErrInvalidDestination / ErrHostNotAllowed /
// ErrHTTPSRequired / ErrInvalidHeader ("webhook: " prefix), sns.ErrReservedAttribute / ErrInvalidAttributeName

Testing Utilities

import (
    "go-mink.dev/testing/bdd"
    "go-mink.dev/testing/assertions"
    "go-mink.dev/testing/projections"
    "go-mink.dev/testing/sagas"
    "go-mink.dev/testing/containers"
)

// BDD-style aggregate testing
func TestOrderAggregate(t *testing.T) {
    order := NewOrder("order-123")

    bdd.Given(t, order).
        When(func() error {
            return order.Create("customer-456")
        }).
        Then(OrderCreated{OrderID: "order-123", CustomerID: "customer-456"})
}

// Event assertions
assertions.AssertEventTypes(t, events, "OrderCreated", "ItemAdded")
assertions.AssertEventsEqual(t, expected, actual)
assertions.AssertContainsEvent(t, events, OrderCreated{OrderID: "123"})

// Event diffing
diffs := assertions.DiffEvents(expected, actual)
t.Error(assertions.FormatDiffs(diffs))

// Projection testing
projections.TestProjection[OrderSummary](t, projection).
    GivenEvents(storedEvents...).
    ThenReadModel("order-123", expectedModel)

// Saga testing
sagas.TestSaga(t, saga).
    GivenEvents(events...).
    ThenCommands(expectedCommands...).
    ThenCompleted()

// PostgreSQL test containers
container := containers.StartPostgres(t)
db := container.MustDB(ctx)

Next: Advanced Patterns →

Core Package (mink)​

Event Store​

Aggregates​

Commands​

Command Bus​

Generic Handler​

Aggregate Handler​

Built-in Middleware​

Idempotency Store​

Audit Logging​

Events​

Projections​

Subscriptions​

Read Model Repository​

Event Versioning & Upcasting​

Field-Level Encryption​

Middleware​

Multi-tenancy​

Errors​

Testing Utilities​

Core Package (`mink`)

Event Store

Aggregates

Commands

Command Bus

Generic Handler

Aggregate Handler

Built-in Middleware

Idempotency Store

Audit Logging

Events

Projections

Subscriptions

Read Model Repository

Event Versioning & Upcasting

Field-Level Encryption

Middleware

Multi-tenancy

Errors

Testing Utilities