package eventbus

import (
        "context"
        "errors"
        "fmt"
        "hash/fnv"
        "reflect"
        "slices"
        "sync"
        "sync/atomic"
        "time"
)

// Handler is a generic event handler function
type Handler[T any] func(T)

// ContextHandler is a generic event handler function that accepts context
type ContextHandler[T any] func(context.Context, T)

// SubscribeOption configures a subscription
type SubscribeOption func(*internalHandler)

// internalHandler wraps a handler with metadata
type internalHandler struct {
        handler      any
        handlerType  reflect.Type
        once         bool
        async        bool
        sequential   bool
        filter       any // Predicate function for filtering events
        invoke       func(context.Context, any)
        filterInvoke func(any) bool
        mu           sync.Mutex
        executed     uint32 // For once handlers, atomically tracks if executed
        onceStateMu  sync.Mutex
        // onceInFlight is set only for durable dispatch. It distinguishes a Once
        // handler consumed by an earlier successful delivery from one whose
        // cancelled async invocation is still finishing and must not be
        // checkpointed past by an immediate follower restart.
        onceInFlight uint32

        // replayErrorPolicy controls how SubscribeWithReplay treats stored
        // events that cannot be decoded; it has no effect on live delivery.
        replayErrorPolicy ReplayErrorPolicy

        // internalDelivery is used by infrastructure subscriptions that need to
        // return a delivery error (currently the log-backed resumable coordinator).
        // suppressObservability prevents that signal handler from double-counting
        // the real user handler invocation performed by the coordinator. onRemove
        // lets infrastructure tear down state before its marker leaves the shard.
        internalDelivery      func(context.Context, any) error
        suppressObservability bool
        onRemove              func()
}

// PanicHandler is called when a handler panics
type PanicHandler func(event any, handlerType reflect.Type, panicValue any)

type handlerPanicError struct {
        value any
}

type durableOnceAttempt struct {
        handler  *internalHandler
        rollback atomic.Bool
}

func (a *durableOnceAttempt) complete() {
        a.handler.onceStateMu.Lock()
        atomic.StoreUint32(&a.handler.onceInFlight, 0)
        if a.rollback.Load() {
                atomic.StoreUint32(&a.handler.executed, 0)
        }
        a.handler.onceStateMu.Unlock()
}

func (a *durableOnceAttempt) requestRollback() {
        a.rollback.Store(true)
        a.handler.onceStateMu.Lock()
        if atomic.LoadUint32(&a.handler.onceInFlight) == 0 {
                atomic.StoreUint32(&a.handler.executed, 0)
        }
        a.handler.onceStateMu.Unlock()
}

func (e *handlerPanicError) Error() string {
        return fmt.Sprintf("handler panic: %v", e.value)
}

// PersistenceErrorHandler is called when event persistence fails
type PersistenceErrorHandler func(event any, eventType reflect.Type, err error)

// PublishHook is called when an event is published
type PublishHook func(eventType reflect.Type, event any)

// PublishHookContext is called when an event is published (with context)
type PublishHookContext func(ctx context.Context, eventType reflect.Type, event any)

const numShards = 32 // Power of 2 for efficient modulo

// shard represents a single shard with its own mutex
type shard struct {
        mu       sync.RWMutex
        handlers map[reflect.Type][]*internalHandler
}

// EventBus is a high-performance event bus with sharded locks
type EventBus struct {
        shards           [numShards]*shard
        panicHandler     PanicHandler
        beforePublish    PublishHook
        afterPublish     PublishHook
        beforePublishCtx PublishHookContext
        afterPublishCtx  PublishHookContext

        // Async handler tracking. A plain sync.WaitGroup would be racy here:
        // Publish (Add) may legally run concurrently with Wait/Shutdown, which
        // WaitGroup forbids when the counter is at zero.
        asyncMu    sync.Mutex
        asyncCond  *sync.Cond
        asyncCount int

        // Lifecycle admission combines the stop bit and active-operation count.
        lifecycleState atomic.Uint64
        shutdownDone   chan struct{}
        shutdownErr    error // published by closing shutdownDone
        stopContext    context.Context
        stopFollow     context.CancelFunc

        // asyncSem, when non-nil, bounds the number of concurrently running
        // async handler goroutines (see WithAsyncHandlerLimit).
        asyncSem chan struct{}

        // Optional persistence fields (nil if not using persistence)
        store                   EventStore
        subscriptionStore       SubscriptionStore
        persistenceErrorHandler PersistenceErrorHandler
        persistenceTimeout      time.Duration
        replayBatchSize         int // Batch size for Replay (default: 100)

        // strictPersistence, when set, makes a failed Append stop delivery:
        // handlers never observe an event the log did not record (see
        // WithStrictPersistence).
        strictPersistence bool

        // originID uniquely identifies this bus instance; it is stamped on every
        // persisted event's Origin field so processes sharing a log can tell
        // their own events apart from a peer's.
        originID string

        // logDelivery, when set, makes the store the only delivery path: Publish
        // appends to the log and never dispatches locally; handlers receive
        // events exclusively from a Follow loop (see WithLogDelivery).
        logDelivery bool

        // followDecoders maps a persisted type name to a closure that decodes a
        // stored event into its Go type and dispatches it locally. Entries are
        // registered by the generic Subscribe* entry points, which are the only
        // places the concrete type is statically known. A nil followTypes entry
        // marks a name made ambiguous on a nonpersistent bus; no decoder is retained
        // for that name.
        followMu       sync.RWMutex
        followDecoders map[string]followDecoder
        followTypes    map[string]reflect.Type

        // persistedTypes prevents a persistent bus from assigning the same durable
        // wire name to distinct Go types. configuring defers that audit until every
        // New option has run. replayIDs prevents two live coordinators on this bus
        // from racing the same scalar subscription checkpoint.
        persistedTypes    *persistedTypeRegistry
        configuring       bool
        replayMu          sync.Mutex
        replayIDs         map[string]struct{}
        replayMarkers     map[*internalHandler]struct{}
        replayMarkerCount atomic.Int64

        // Upcast registry for event migration
        upcastRegistry *upcastRegistry

        // Optional observability (metrics & tracing)
        observability Observability
}

// TypeNamer is an optional interface that events can implement to provide
// their own type name. This gives explicit control over event type naming,
// which is useful for:
//   - Stable type names across package refactoring
//   - Custom versioning schemes (e.g., "UserCreated.v2")
//   - Compatibility with external event stores
//
// If an event implements TypeNamer, EventType() will use the provided name
// instead of the reflection-based name. EventTypeName must depend only on the
// type, not instance fields: generic replay/follow registration may derive it
// from a fresh zero value.
// The bus may omit calls when the wire name is unused; callers must not rely
// on EventTypeName being invoked for every publication.
//
// Persisted or distributed event types should implement TypeNamer. Go's
// reflection fallback uses the declared package name (for example,
// "orders.Created"), not its full import path, and can therefore change after
// refactors or collide with another package that has the same name.
//
// Example:
//
//        type UserCreatedEvent struct {
//            UserID string
//        }
//
//        func (e UserCreatedEvent) EventTypeName() string {
//            return "user.created.v1"
//        }
type TypeNamer interface {
        EventTypeName() string
}

// EventType returns the type name of an event.
// If the event implements TypeNamer, it returns the custom name.
// Otherwise, it returns Go's reflection name, which uses the declared package
// name rather than the full import path. It returns "nil" for a nil event.
//
// This is useful for comparing with StoredEvent.Type during replay.
//
// Example with reflection (default):
//
//        eventType := EventType(MyEvent{})
//        // Returns: "mypackage.MyEvent"
//
// Example with TypeNamer:
//
//        type MyEvent struct{}
//        func (e MyEvent) EventTypeName() string { return "my-event.v1" }
//        eventType := EventType(MyEvent{})
//        // Returns: "my-event.v1"
//
// Usage in replay:
//
//        bus.Replay(ctx, OffsetOldest, func(event *StoredEvent) error {
//            if event.Type == EventType(MyEvent{}) {
//                // Process MyEvent
//            }
//            return nil
//        })
func EventType(event any) string {
        if event == nil {
                return "nil"
        }
        value := reflect.ValueOf(event)
        if value.Kind() == reflect.Pointer && value.IsNil() {
                return typeNameOf(value.Type())
        }
        if namer, ok := event.(TypeNamer); ok {
                return namer.EventTypeName()
        }
        return value.Type().String()
}

// Observability is an optional interface for metrics and tracing.
// Implementations can track event publishing, handler execution, and errors.
//
// When observability is not configured, the bus skips telemetry callbacks and
// handler timing. Persistent publishing still computes event names for storage.
//
// The context returned from each method can be used to propagate trace
// spans and other context-specific data through the event processing pipeline.
//
// Example implementation: see github.com/jilio/ebu/otel package for
// OpenTelemetry integration.
type Observability interface {
        // OnPublishStart is called when an event is about to be published.
        // Returns a context that will be passed to handlers and subsequent hooks.
        // The event parameter allows implementations to extract custom attributes.
        OnPublishStart(ctx context.Context, eventType string, event any) context.Context

        // OnPublishComplete is called after all synchronous handlers complete.
        // Note: This is called before async handlers complete.
        OnPublishComplete(ctx context.Context, eventType string)

        // OnHandlerStart is called before a handler executes.
        // Returns a context for the handler execution.
        OnHandlerStart(ctx context.Context, eventType string, async bool) context.Context

        // OnHandlerComplete is called after a handler completes.
        // The error parameter is non-nil if the handler panicked.
        OnHandlerComplete(ctx context.Context, eventType string, duration time.Duration, err error)

        // OnPersistStart is called before persisting an event. The assigned
        // offset is not known yet at this point; it is reported to
        // OnPersistComplete instead.
        OnPersistStart(ctx context.Context, eventType string) context.Context

        // OnPersistComplete is called after persisting an event. On success,
        // offset is the offset the store assigned; on failure it is empty and
        // err is non-nil.
        OnPersistComplete(ctx context.Context, eventType string, duration time.Duration, offset Offset, err error)
}

// Option is a function that configures the EventBus
type Option func(*EventBus)

// New creates a new EventBus with sharded locks for better performance.
// It panics when the final option set is internally inconsistent, including
// when typed registrations made by custom options assign one durable event
// name to distinct Go types on a bus configured with WithStore.
func New(opts ...Option) *EventBus {
        bus := &EventBus{
                upcastRegistry: newUpcastRegistry(),
                originID:       NewEventID(),
                followDecoders: make(map[string]followDecoder),
                followTypes:    make(map[string]reflect.Type),
                persistedTypes: newDeferredPersistedTypeRegistry(),
                configuring:    true,
                replayIDs:      make(map[string]struct{}),
                replayMarkers:  make(map[*internalHandler]struct{}),
        }
        bus.asyncCond = sync.NewCond(&bus.asyncMu)
        bus.shutdownDone = make(chan struct{})
        bus.stopContext, bus.stopFollow = context.WithCancel(context.Background())

        // Initialize shards
        for i := 0; i < numShards; i++ {
                bus.shards[i] = &shard{
                        handlers: make(map[reflect.Type][]*internalHandler),
                }
        }

        // Apply options
        for _, opt := range opts {
                opt(bus)
        }
        bus.configuring = false

        // Options are arbitrary functions and may perform typed registrations.
        // Audit their complete durable-name set only after every option has run so
        // WithStore has identical behavior regardless of its position in opts.
        if bus.store != nil {
                bus.activatePersistenceTypes()
        }

        // Log-delivery mode without a store would silently drop every publish
        // (nothing appends, nothing follows). That is a configuration bug, not a
        // runtime condition — fail loudly at construction, mirroring WithUpcast.
        if bus.logDelivery && bus.store == nil {
                panic("eventbus: WithLogDelivery requires WithStore: without a store there is no log to deliver from")
        }

        return bus
}

// getShard returns the shard for a given event type using FNV hash
func (bus *EventBus) getShard(eventType reflect.Type) *shard {
        h := fnv.New32a()
        h.Write([]byte(eventType.String()))
        shardIndex := h.Sum32() & (numShards - 1) // Fast modulo for power of 2
        return bus.shards[shardIndex]
}

// buildHandler is the single validation chokepoint for every subscription
// entry point: it constructs the internalHandler, applies each option exactly
// once, and runs all subscription-time validation (event type, options,
// filter). New entry points must go through it so no check can be missed.
func buildHandler(handler any, eventType reflect.Type, opts []SubscribeOption) (*internalHandler, error) {
        if err := validateEventType(eventType); err != nil {
                return nil, err
        }

        h := &internalHandler{
                handler:     handler,
                handlerType: reflect.TypeOf(handler),
        }

        for _, opt := range opts {
                if opt == nil {
                        return nil, fmt.Errorf("eventbus: subscribe option cannot be nil")
                }
                opt(h)
        }

        if err := validateFilter(h, eventType); err != nil {
                return nil, err
        }

        return h, nil
}

// buildPayloadHandler and buildContextHandler bind typed user functions once
// at subscription time. Dispatch routes by an event's dynamic concrete type,
// which can differ from Publish's static type when a value is passed through
// any or another interface. Keeping these adapters on internalHandler avoids
// both a static-generic type mismatch and reflection in the hot path.
func buildPayloadHandler[T any](handler Handler[T], eventType reflect.Type, opts []SubscribeOption) (*internalHandler, error) {
        h, err := buildHandler(handler, eventType, opts)
        if err != nil {
                return nil, err
        }
        h.invoke = func(_ context.Context, event any) { handler(event.(T)) }
        bindFilter[T](h)
        return h, nil
}

func buildContextHandler[T any](handler ContextHandler[T], eventType reflect.Type, opts []SubscribeOption) (*internalHandler, error) {
        h, err := buildHandler(handler, eventType, opts)
        if err != nil {
                return nil, err
        }
        h.invoke = func(ctx context.Context, event any) { handler(ctx, event.(T)) }
        bindFilter[T](h)
        return h, nil
}

func bindFilter[T any](h *internalHandler) {
        if h.filter == nil {
                return
        }
        predicate := h.filter.(func(T) bool)
        h.filterInvoke = func(event any) bool { return predicate(event.(T)) }
}

// addHandler registers a validated handler in the shard for its event type.
func (bus *EventBus) addHandler(eventType reflect.Type, h *internalHandler) {
        shard := bus.getShard(eventType)
        shard.mu.Lock()
        shard.handlers[eventType] = append(shard.handlers[eventType], h)
        shard.mu.Unlock()
}

// removeHandler unregisters a handler by identity (pointer equality on the
// internalHandler), which — unlike Unsubscribe's code-pointer matching —
// always removes exactly the registration it was given. It is a no-op when
// the handler is already gone (unsubscribed twice, or removed by Clear).
func (bus *EventBus) removeHandler(eventType reflect.Type, h *internalHandler) {
        shard := bus.getShard(eventType)
        shard.mu.Lock()

        handlers := shard.handlers[eventType]
        for i, existing := range handlers {
                if existing == h {
                        if existing.onRemove != nil {
                                existing.onRemove()
                        }
                        handlers = slices.Delete(handlers, i, i+1)
                        if len(handlers) == 0 {
                                delete(shard.handlers, eventType)
                        } else {
                                shard.handlers[eventType] = handlers
                        }
                        shard.mu.Unlock()
                        return
                }
        }
        shard.mu.Unlock()
}

// OriginID returns the unique identifier of this bus instance. Every event
// the bus persists carries it as Event.Origin, so consumers of a log shared
// by several processes can tell this instance's events from a peer's.
func (bus *EventBus) OriginID() string {
        return bus.originID
}

// Subscription is a handle to a single registration, returned by
// SubscribeWithHandle and SubscribeContextWithHandle. Unlike Unsubscribe —
// which matches handlers by function code pointer and therefore cannot tell
// two closures from the same function literal apart — a handle identifies
// exactly the registration that created it.
type Subscription struct {
        bus       *EventBus
        eventType reflect.Type
        h         *internalHandler
}

// Unsubscribe removes this subscription's handler from the bus. It is
// idempotent: calling it more than once (or after Clear/ClearAll already
// removed the handler) is a no-op. Safe for concurrent use.
func (s *Subscription) Unsubscribe() {
        s.bus.removeHandler(s.eventType, s.h)
}

// SubscribeWithHandle registers a handler for events of type T and returns a
// Subscription handle for precise removal. Semantics are identical to
// Subscribe otherwise; T must be a concrete type.
//
// Prefer this over Subscribe+Unsubscribe when the handler is a closure:
// handles remove exactly the registration they came from, with none of
// Unsubscribe's code-pointer ambiguity.
func SubscribeWithHandle[T any](bus *EventBus, handler Handler[T], opts ...SubscribeOption) (*Subscription, error) {
        if bus == nil {
                return nil, fmt.Errorf("eventbus: bus cannot be nil")
        }
        if handler == nil {
                return nil, fmt.Errorf("eventbus: handler cannot be nil")
        }

        if !bus.beginOperation() {
                return nil, ErrClosed
        }
        defer bus.endOperation()

        eventType := reflect.TypeOf((*T)(nil)).Elem()
        h, err := buildPayloadHandler(handler, eventType, opts)
        if err != nil {
                return nil, err
        }

        if err := registerFollowDecoder[T](bus); err != nil {
                return nil, err
        }
        bus.addHandler(eventType, h)
        return &Subscription{bus: bus, eventType: eventType, h: h}, nil
}

// SubscribeContextWithHandle registers a context-aware handler for events of
// type T and returns a Subscription handle for precise removal. Semantics are
// identical to SubscribeContext otherwise; T must be a concrete type.
func SubscribeContextWithHandle[T any](bus *EventBus, handler ContextHandler[T], opts ...SubscribeOption) (*Subscription, error) {
        if bus == nil {
                return nil, fmt.Errorf("eventbus: bus cannot be nil")
        }
        if handler == nil {
                return nil, fmt.Errorf("eventbus: handler cannot be nil")
        }

        if !bus.beginOperation() {
                return nil, ErrClosed
        }
        defer bus.endOperation()

        eventType := reflect.TypeOf((*T)(nil)).Elem()
        h, err := buildContextHandler(handler, eventType, opts)
        if err != nil {
                return nil, err
        }

        if err := registerFollowDecoder[T](bus); err != nil {
                return nil, err
        }
        bus.addHandler(eventType, h)
        return &Subscription{bus: bus, eventType: eventType, h: h}, nil
}

// Subscribe registers a handler for events of type T.
//
// T must be a concrete type. Interface types are rejected with an error:
// Publish routes events by their dynamic (concrete) type, so a handler
// registered under an interface type could never receive an event — even
// one published through a variable of that interface type.
func Subscribe[T any](bus *EventBus, handler Handler[T], opts ...SubscribeOption) error {
        if bus == nil {
                return fmt.Errorf("eventbus: bus cannot be nil")
        }
        if handler == nil {
                return fmt.Errorf("eventbus: handler cannot be nil")
        }

        if !bus.beginOperation() {
                return ErrClosed
        }
        defer bus.endOperation()

        eventType := reflect.TypeOf((*T)(nil)).Elem()
        h, err := buildPayloadHandler(handler, eventType, opts)
        if err != nil {
                return err
        }

        if err := registerFollowDecoder[T](bus); err != nil {
                return err
        }
        bus.addHandler(eventType, h)
        return nil
}

// SubscribeContext registers a context-aware handler for events of type T.
//
// T must be a concrete type; interface types are rejected (see Subscribe).
func SubscribeContext[T any](bus *EventBus, handler ContextHandler[T], opts ...SubscribeOption) error {
        if bus == nil {
                return fmt.Errorf("eventbus: bus cannot be nil")
        }
        if handler == nil {
                return fmt.Errorf("eventbus: handler cannot be nil")
        }

        if !bus.beginOperation() {
                return ErrClosed
        }
        defer bus.endOperation()

        eventType := reflect.TypeOf((*T)(nil)).Elem()
        h, err := buildContextHandler(handler, eventType, opts)
        if err != nil {
                return err
        }

        if err := registerFollowDecoder[T](bus); err != nil {
                return err
        }
        bus.addHandler(eventType, h)
        return nil
}

// validateEventType rejects event types that Publish could never route to.
// Publish looks handlers up by the event's dynamic (concrete) type, so a
// subscription registered under an interface type would be silently dead.
func validateEventType(eventType reflect.Type) error {
        if eventType.Kind() == reflect.Interface {
                return fmt.Errorf("eventbus: cannot subscribe to interface type %s: events are routed by their concrete type, so an interface subscription would never receive events; subscribe to each concrete event type instead", eventType)
        }
        return nil
}

// validateFilter ensures a WithFilter predicate matches the subscribed event type.
// Without this check, a mismatched predicate (e.g. WithFilter[U] on Subscribe[T])
// would silently never fire and the handler would receive all events unfiltered.
func validateFilter(h *internalHandler, eventType reflect.Type) error {
        if h.filter == nil {
                return nil
        }
        ft := reflect.TypeOf(h.filter)
        if ft.Kind() != reflect.Func || ft.NumIn() != 1 || ft.In(0) != eventType {
                return fmt.Errorf("eventbus: filter predicate type %s does not match event type %s", ft, eventType)
        }
        if reflect.ValueOf(h.filter).IsNil() {
                return fmt.Errorf("eventbus: filter predicate cannot be nil")
        }
        return nil
}

// Unsubscribe removes a handler for events of type T.
//
// Handlers are matched by function code pointer. This has two known
// limitations inherent to Go:
//   - Two closures created from the same function literal share a code
//     pointer and are indistinguishable; the first registered one is removed.
//   - Method values on different receivers share a code pointer.
//
// If you need precise unsubscription of closures, use SubscribeWithHandle
// and the returned Subscription's Unsubscribe method instead — handles are
// matched by identity, not code pointer. Alternatively keep a reference to
// the exact handler you registered and unsubscribe with it, or use Clear[T].
func Unsubscribe[T any, H any](bus *EventBus, handler H) error {
        if bus == nil {
                return fmt.Errorf("eventbus: bus cannot be nil")
        }
        eventType := reflect.TypeOf((*T)(nil)).Elem()
        handlerValue := reflect.ValueOf(handler)
        if !handlerValue.IsValid() || handlerValue.Kind() != reflect.Func || handlerValue.IsNil() {
                return fmt.Errorf("eventbus: handler must be a non-nil function")
        }
        handlerPtr := handlerValue.Pointer()

        shard := bus.getShard(eventType)
        shard.mu.Lock()
        defer shard.mu.Unlock()

        handlers := shard.handlers[eventType]
        for i, h := range handlers {
                // Infrastructure markers route through internalDelivery and deliberately
                // have no public handler function to compare.
                if h.handler == nil {
                        continue
                }
                registered := reflect.ValueOf(h.handler)
                if registered.Kind() == reflect.Func && !registered.IsNil() && registered.Pointer() == handlerPtr {
                        if h.onRemove != nil {
                                h.onRemove()
                        }
                        handlers = slices.Delete(handlers, i, i+1)
                        if len(handlers) == 0 {
                                delete(shard.handlers, eventType)
                        } else {
                                shard.handlers[eventType] = handlers
                        }
                        return nil
                }
        }

        return fmt.Errorf("handler not found")
}

// Publish publishes an event to all registered handlers.
// After Shutdown begins it silently discards the event; use TryPublish to detect rejection.
// It panics if bus is nil.
func Publish[T any](bus *EventBus, event T) {
        PublishContext(bus, context.Background(), event)
}

// PublishContext publishes an event with context to all registered handlers.
// After Shutdown begins it silently discards the event; use TryPublishContext to detect rejection.
// It panics if bus is nil.
//
// When the bus has a store configured (WithStore), the event is persisted
// before handlers run. Persistence is best-effort by default for ordinary
// Subscribe handlers: on failure they still receive the in-memory event and
// the error is reported to the PersistenceErrorHandler (see
// WithPersistenceErrorHandler). Log-backed resumable subscriptions consume
// only records visible in EventStore. An Append error can be ambiguous: if a
// remote store committed before its acknowledgement was lost, that durable
// record may still be delivered now or by a later replay. WithStrictPersistence
// makes every handler skip immediate delivery after a reported persist error. Use
// TryPublish/TryPublishContext to receive the persistence error directly.
func PublishContext[T any](bus *EventBus, ctx context.Context, event T) {
        publishContext(bus, ctx, event)
}

// TryPublish is Publish with the reported persistence outcome returned: it
// returns the marshal/Append error (nil on success or when no store is
// configured). An Append error may be ambiguous if a remote commit succeeded
// before its acknowledgement was lost. After Shutdown begins it returns ErrClosed
// without invoking callbacks. It panics if bus is nil.
func TryPublish[T any](bus *EventBus, event T) error {
        return publishContext(bus, context.Background(), event)
}

// TryPublishContext is PublishContext with the persistence outcome returned.
//
// The returned error does not by itself imply ordinary Subscribe handlers were
// skipped: in best-effort mode (the default) their delivery proceeds despite
// the error, while under WithStrictPersistence it is skipped. Log-backed
// resumable subscriptions consume only records visible in EventStore. Because
// an Append error may arrive after a remote commit, such a record can still be
// delivered now or by a later replay. Either way the error is also reported to
// the PersistenceErrorHandler, which remains the right channel for passive
// monitoring; TryPublishContext is for publishers that must act on the failure
// (e.g. fail the request that caused the publish).
// It returns ErrClosed without invoking callbacks if shutdown has begun.
// It panics if bus is nil.
func TryPublishContext[T any](bus *EventBus, ctx context.Context, event T) error {
        return publishContext(bus, ctx, event)
}

// publishContext is the single publish path: it persists (when configured),
// delivers, and returns the persistence error, if any. Publish/PublishContext
// discard the error; TryPublish/TryPublishContext surface it.
func publishContext[T any](bus *EventBus, ctx context.Context, event T) error {
        if bus == nil {
                panic("eventbus: Publish called with nil bus")
        }
        if !bus.beginOperation() {
                return ErrClosed
        }
        defer bus.endOperation()

        eventType := reflect.TypeOf(event)
        var eventTypeName string
        if bus.store != nil {
                // Persistent routing must use the reproducible type-derived name. The
                // persistence step separately rejects an instance-dependent TypeNamer.
                eventTypeName = typeNameOf(eventType)
        } else if bus.observability != nil {
                eventTypeName = EventType(event)
        }

        // Observability: Track publish start
        if bus.observability != nil {
                ctx = bus.observability.OnPublishStart(ctx, eventTypeName, event)
        }

        // Call before publish hooks
        if bus.beforePublish != nil {
                bus.beforePublish(eventType, event)
        }
        if bus.beforePublishCtx != nil {
                bus.beforePublishCtx(ctx, eventType, event)
        }

        // Persist the event before handlers run. On success the assigned offset
        // is attached to ctx and can be retrieved with OffsetFromContext.
        var persistErr error
        if bus.store != nil {
                ctx, persistErr = bus.persistEvent(ctx, eventType, event)
        }

        // In strict mode a failed persist skips delivery: handlers never observe
        // an event the log did not record, so replay and live handling can never
        // diverge. In log-delivery mode (WithLogDelivery) local dispatch is
        // always skipped: delivery happens exclusively through the follower
        // tailing the store (see Follow), so every process — including this one —
        // observes the same events in the same order. The after-publish hooks and
        // observability below still fire so monitoring sees the attempt.
        deliver := (persistErr == nil || !bus.strictPersistence) && !bus.logDelivery

        if deliver {
                dispatch(bus, ctx, eventType, eventTypeName, event)
        }
        if bus.store != nil && !bus.logDelivery && (persistErr == nil || deliver) {
                // A stored predecessor may upcast into a different concrete subscriber
                // type, whose shard the ordinary in-memory dispatch cannot know to wake.
                // Signal every resumable coordinator so it can scan the durable log and
                // apply the registry. Same-type wake-ups are intentionally idempotent:
                // ordinary dispatch may have skipped its marker when an earlier handler
                // cancelled ctx. Best-effort delivery also wakes after an ambiguous
                // Append error, matching its existing same-type behavior; strict failures
                // remain suppressed and can be recovered by a later Follow/replay.
                signalReplayMarkers(bus, ctx, eventTypeName, event)
        }

        // For async handlers, we don't wait inline to avoid blocking
        // Users can call bus.Wait() if they need to wait for completion

        // Call after publish hooks
        if bus.afterPublish != nil {
                bus.afterPublish(eventType, event)
        }
        if bus.afterPublishCtx != nil {
                bus.afterPublishCtx(ctx, eventType, event)
        }

        // Observability: Track publish complete (sync handlers done)
        if bus.observability != nil {
                bus.observability.OnPublishComplete(ctx, eventTypeName)
        }

        return persistErr
}

// signalReplayMarkers wakes log-backed resumable subscriptions whose concrete
// shard was not reached by the ordinary publish dispatch. Markers are tracked
// separately from shards so this path is O(active replay subscriptions), not a
// scan of every handler shard. The snapshot lock is released before any
// coordinator runs, preserving Clear's shard -> replay registry ordering.
func signalReplayMarkers[T any](bus *EventBus, ctx context.Context, eventTypeName string, event T) {
        if bus.replayMarkerCount.Load() == 0 {
                return
        }
        bus.replayMu.Lock()
        markers := make([]*internalHandler, 0, len(bus.replayMarkers))
        for marker := range bus.replayMarkers {
                markers = append(markers, marker)
        }
        bus.replayMu.Unlock()

        for _, marker := range markers {
                _ = callHandlerWithContext(marker, ctx, event, bus.panicHandler,
                        bus.observability, eventTypeName, false)
        }
}

// dispatch delivers an event to the handlers registered for eventType. It is
// the delivery half of the publish path, shared by publishContext (live
// publishes) and the follower (events read back from a shared store): all
// subscription options — filters, Once, Async, Sequential — behave
// identically on both paths. It does not run the publish hooks or
// publish-level observability; those belong to the publish, not to delivery.
func dispatch[T any](bus *EventBus, ctx context.Context, eventType reflect.Type, eventTypeName string, event T) {
        _ = dispatchWithMode(bus, ctx, eventType, eventTypeName, event, dispatchNonBlocking)
}

// dispatchMode controls whether delivery returns immediately after launching
// async handlers or waits for the async work belonging to this event. Local
// Publish uses the non-blocking mode; durable Follow uses the waiting mode so
// its checkpoint cannot overtake handler completion.
type dispatchMode uint8

const (
        dispatchNonBlocking dispatchMode = iota
        dispatchWaitForAsync
)

func dispatchWithMode[T any](bus *EventBus, ctx context.Context, eventType reflect.Type, eventTypeName string, event T, mode dispatchMode) error {
        // Get handlers from appropriate shard
        shard := bus.getShard(eventType)
        shard.mu.RLock()
        handlers := shard.handlers[eventType]
        // Create a copy to avoid holding the lock during handler execution
        handlersCopy := make([]*internalHandler, len(handlers))
        copy(handlersCopy, handlers)
        shard.mu.RUnlock()

        // Execute handlers without holding the lock
        var onceHandlersToRemove []*internalHandler
        var durableOnceAttempts []*durableOnceAttempt
        var handlerErrors []error
        var cancellationErr error
        var asyncDone chan error
        if mode == dispatchWaitForAsync {
                asyncDone = make(chan error, len(handlersCopy))
        }
        asyncCount := 0

handlerLoop:
        for _, h := range handlersCopy {
                // Check context cancellation before doing anything with the handler.
                // This must happen before the Once CAS so a cancelled publish never
                // consumes a once-handler without executing it.
                if err := ctx.Err(); err != nil {
                        cancellationErr = err
                        break
                }

                // Check filter if present. The predicate type is validated at
                // Subscribe time; if the assertion still fails, fail closed.
                if h.filter != nil {
                        matches, err := callFilter(h, event, bus.panicHandler)
                        if err != nil {
                                handlerErrors = append(handlerErrors, err)
                                continue
                        }
                        if !matches {
                                continue // Skip this handler as event doesn't match filter
                        }
                }

                // For Once handlers, use CompareAndSwap to ensure atomic execution. A
                // durable dispatch additionally publishes its in-flight state under a
                // small per-handler lock: an immediate follower restart must fail/retry,
                // not skip and checkpoint past an async invocation still unwinding from a
                // cancelled attempt.
                var onceAttempt *durableOnceAttempt
                if h.once {
                        if mode == dispatchWaitForAsync {
                                h.onceStateMu.Lock()
                                claimed := atomic.CompareAndSwapUint32(&h.executed, 0, 1)
                                if claimed {
                                        atomic.StoreUint32(&h.onceInFlight, 1)
                                        onceAttempt = &durableOnceAttempt{handler: h}
                                }
                                inFlight := atomic.LoadUint32(&h.onceInFlight) != 0
                                h.onceStateMu.Unlock()
                                if !claimed {
                                        if inFlight {
                                                handlerErrors = append(handlerErrors,
                                                        fmt.Errorf("eventbus: Once handler %v is still completing a previous durable attempt", h.handlerType))
                                                break handlerLoop
                                        }
                                        continue // Already completed and awaiting/removing its registration.
                                }
                        } else if !atomic.CompareAndSwapUint32(&h.executed, 0, 1) {
                                continue // Already executed.
                        }
                        // Mark for removal after execution
                        onceHandlersToRemove = append(onceHandlersToRemove, h)
                }

                // Bounded async capacity remains cancellable. A Once handler is selected
                // before waiting so concurrent publishes can fail its CAS immediately;
                // if cancellation wins before the goroutine starts, roll that selection
                // back so a later durable retry can still execute it.
                asyncSlot := false
                if h.async && bus.asyncSem != nil {
                        select {
                        case bus.asyncSem <- struct{}{}:
                                asyncSlot = true
                        case <-ctx.Done():
                                if h.once {
                                        if onceAttempt != nil {
                                                onceAttempt.requestRollback()
                                                onceAttempt.complete()
                                        }
                                        atomic.StoreUint32(&h.executed, 0)
                                        onceHandlersToRemove = onceHandlersToRemove[:len(onceHandlersToRemove)-1]
                                }
                                cancellationErr = ctx.Err()
                                break handlerLoop
                        }
                }
                if onceAttempt != nil {
                        durableOnceAttempts = append(durableOnceAttempts, onceAttempt)
                }

                if h.async {
                        if mode == dispatchWaitForAsync {
                                asyncCount++
                        }
                        bus.asyncStarted()
                        go func(handler *internalHandler, attempt *durableOnceAttempt, releaseSlot bool, completion chan<- error, waitForCompletion bool) {
                                var handlerErr error
                                defer func() {
                                        if attempt != nil {
                                                attempt.complete()
                                        }
                                        if releaseSlot {
                                                <-bus.asyncSem
                                        }
                                        bus.asyncFinished()
                                        if waitForCompletion {
                                                // Buffered for every snapshotted handler: cancellation may make
                                                // dispatch return before this handler, but completion never leaks
                                                // a goroutine waiting for a receiver.
                                                completion <- handlerErr
                                        }
                                }()

                                // Once handlers always run: their execution slot was already
                                // consumed by the CAS above, so skipping here would silently
                                // drop them forever. Other handlers re-check the context.
                                if !handler.once && ctx.Err() != nil {
                                        handlerErr = ctx.Err()
                                        return
                                }
                                handlerErr = callHandlerWithContext(handler, ctx, event, bus.panicHandler, bus.observability, eventTypeName, true)
                        }(h, onceAttempt, asyncSlot, asyncDone, mode == dispatchWaitForAsync)
                } else {
                        deliveryCtx := ctx
                        if mode == dispatchWaitForAsync && h.internalDelivery != nil {
                                // Only infrastructure delivery needs to know that its outer durable
                                // caller must wait. Keeping this marker off ordinary handler contexts
                                // avoids an allocation on every non-durable publish.
                                deliveryCtx = context.WithValue(ctx, durableDispatchCtxKey{}, true)
                        }
                        if err := callHandlerWithContext(h, deliveryCtx, event, bus.panicHandler, bus.observability, eventTypeName, false); err != nil {
                                handlerErrors = append(handlerErrors, err)
                        }
                        if onceAttempt != nil {
                                onceAttempt.complete()
                        }
                }
        }

        if cancellationErr != nil {
                if mode == dispatchWaitForAsync {
                        // The follower cannot know whether every selected handler completed
                        // before cancellation. Keep them retryable; at-least-once duplication
                        // is safer than silently checkpointing a failed one on a later run.
                        rollbackDurableOnceAttempts(durableOnceAttempts)
                } else {
                        removeOnceHandlers(shard, eventType, onceHandlersToRemove)
                }
                return errors.Join(append(handlerErrors, cancellationErr)...)
        }

        if mode == dispatchWaitForAsync {
                for asyncCount > 0 {
                        select {
                        case err := <-asyncDone:
                                asyncCount--
                                if err != nil {
                                        handlerErrors = append(handlerErrors, err)
                                }
                        case <-ctx.Done():
                                rollbackDurableOnceAttempts(durableOnceAttempts)
                                return errors.Join(append(handlerErrors, ctx.Err())...)
                        }
                }
                // A synchronous handler may cancel the delivery context itself, and an
                // async completion can win the final select at the same instant as
                // cancellation. In either case Follow will refuse to checkpoint this
                // attempt, so keep every selected Once handler eligible for its retry.
                if err := ctx.Err(); err != nil {
                        rollbackDurableOnceAttempts(durableOnceAttempts)
                        return errors.Join(append(handlerErrors, err)...)
                }
        }

        if len(handlerErrors) == 0 {
                commitDurableOnceAttempts(durableOnceAttempts)
                removeOnceHandlers(shard, eventType, onceHandlersToRemove)
                return nil
        }

        deliveryErr := errors.Join(handlerErrors...)
        if deliveryErr != nil && mode == dispatchWaitForAsync {
                // A durable event is retried as one unit. Keep every Once handler selected
                // for this failed attempt registered and make it eligible again, including
                // handlers that succeeded before a sibling failed.
                rollbackDurableOnceAttempts(durableOnceAttempts)
                return deliveryErr
        }

        commitDurableOnceAttempts(durableOnceAttempts)
        removeOnceHandlers(shard, eventType, onceHandlersToRemove)
        return deliveryErr
}

func rollbackDurableOnceAttempts(attempts []*durableOnceAttempt) {
        for _, attempt := range attempts {
                attempt.requestRollback()
        }
}

func commitDurableOnceAttempts(attempts []*durableOnceAttempt) {
        for _, attempt := range attempts {
                // A successful durable dispatch has waited for every async completion,
                // so complete has already cleared onceInFlight. Keeping executed at one
                // prevents a snapshotted concurrent dispatch from running the handler
                // again before removeOnceHandlers takes the shard lock.
                attempt.handler.onceStateMu.Lock()
                atomic.StoreUint32(&attempt.handler.onceInFlight, 0)
                attempt.handler.onceStateMu.Unlock()
        }
}

func removeOnceHandlers(shard *shard, eventType reflect.Type, onceHandlers []*internalHandler) {
        if len(onceHandlers) == 0 {
                return
        }
        shard.mu.Lock()
        handlers := shard.handlers[eventType]
        for _, onceHandler := range onceHandlers {
                for i, h := range handlers {
                        if h == onceHandler {
                                handlers = slices.Delete(handlers, i, i+1)
                                break
                        }
                }
        }
        if len(handlers) == 0 {
                delete(shard.handlers, eventType)
        } else {
                shard.handlers[eventType] = handlers
        }
        shard.mu.Unlock()
}

// Clear removes all handlers for events of type T
func Clear[T any](bus *EventBus) {
        eventType := reflect.TypeOf((*T)(nil)).Elem()

        shard := bus.getShard(eventType)
        shard.mu.Lock()
        for _, handler := range shard.handlers[eventType] {
                if handler.onRemove != nil {
                        handler.onRemove()
                }
        }
        delete(shard.handlers, eventType)
        shard.mu.Unlock()
}

// ClearAll removes all handlers for all event types
func ClearAll(bus *EventBus) {
        for i := 0; i < numShards; i++ {
                bus.shards[i].mu.Lock()
                for _, handlers := range bus.shards[i].handlers {
                        for _, handler := range handlers {
                                if handler.onRemove != nil {
                                        handler.onRemove()
                                }
                        }
                }
                bus.shards[i].handlers = make(map[reflect.Type][]*internalHandler)
                bus.shards[i].mu.Unlock()
        }
}

// HasHandlers checks if there are handlers for events of type T
func HasHandlers[T any](bus *EventBus) bool {
        eventType := reflect.TypeOf((*T)(nil)).Elem()

        shard := bus.getShard(eventType)
        shard.mu.RLock()
        defer shard.mu.RUnlock()

        return len(shard.handlers[eventType]) > 0
}

// HandlerCount returns the number of handlers for events of type T
func HandlerCount[T any](bus *EventBus) int {
        eventType := reflect.TypeOf((*T)(nil)).Elem()

        shard := bus.getShard(eventType)
        shard.mu.RLock()
        defer shard.mu.RUnlock()

        return len(shard.handlers[eventType])
}

// asyncStarted records the start of an async handler goroutine.
func (bus *EventBus) asyncStarted() {
        bus.asyncMu.Lock()
        bus.asyncCount++
        bus.asyncMu.Unlock()
}

// asyncFinished records the completion of an async handler goroutine and
// wakes Wait callers when the last one finishes.
func (bus *EventBus) asyncFinished() {
        bus.asyncMu.Lock()
        bus.asyncCount--
        if bus.asyncCount == 0 {
                bus.asyncCond.Broadcast()
        }
        bus.asyncMu.Unlock()
}

// Wait blocks until all async handlers and deferred resumable-subscription
// drains complete. Ordinary synchronous handlers still complete inline in
// their Publish call and are not independently tracked.
// It is safe to call concurrently with Publish.
func (bus *EventBus) Wait() {
        bus.asyncMu.Lock()
        for bus.asyncCount > 0 {
                bus.asyncCond.Wait()
        }
        bus.asyncMu.Unlock()
}

// callHandlerWithContext calls a handler with proper type checking and panic recovery
func callHandlerWithContext[T any](h *internalHandler, ctx context.Context, event T, panicHandler PanicHandler, obs Observability, eventTypeName string, async bool) (panicErr error) {
        if h.suppressObservability {
                obs = nil
        }
        var start time.Time
        if obs != nil {
                start = time.Now()
        }

        defer func() {
                var duration time.Duration
                if obs != nil {
                        duration = time.Since(start)
                }
                if r := recover(); r != nil {
                        panicErr = &handlerPanicError{value: r}
                        if panicHandler != nil {
                                panicHandler(event, h.handlerType, r)
                        }
                }
                // Observability: Track handler complete
                if obs != nil {
                        obs.OnHandlerComplete(ctx, eventTypeName, duration, panicErr)
                }
        }()

        // Observability: Track handler start
        if obs != nil {
                ctx = obs.OnHandlerStart(ctx, eventTypeName, async)
        }

        // Sequential handlers need locking
        if h.sequential {
                h.mu.Lock()
                defer h.mu.Unlock()
        }
        if h.internalDelivery != nil {
                return h.internalDelivery(ctx, event)
        }

        // Keep the common concrete-T path monomorphized and allocation-free. The
        // bound adapter is the correctness fallback when Publish's static T is any
        // or another interface while routing selected a concrete handler shard.
        switch fn := h.handler.(type) {
        case Handler[T]:
                fn(event)
        case ContextHandler[T]:
                fn(ctx, event)
        case nil:
                // Infrastructure handlers returned through internalDelivery above.
        default:
                if h.invoke != nil {
                        h.invoke(ctx, event)
                }
        }
        return nil
}

// callFilter keeps user predicate panics inside the same isolation contract as
// handler panics while deliberately remaining outside handler observability: a
// rejected event never starts the handler, and neither does a predicate that
// fails before that boundary.
func callFilter[T any](h *internalHandler, event T, panicHandler PanicHandler) (matches bool, panicErr error) {
        defer func() {
                if recovered := recover(); recovered != nil {
                        panicErr = &handlerPanicError{value: recovered}
                        if panicHandler != nil {
                                panicHandler(event, h.handlerType, recovered)
                        }
                }
        }()
        if filter, ok := h.filter.(func(T) bool); ok {
                return filter(event), nil
        }
        if h.filterInvoke != nil {
                return h.filterInvoke(event), nil
        }
        return false, nil
}

// Subscribe Options

// Once makes the handler execute only once
func Once() SubscribeOption {
        return func(h *internalHandler) {
                h.once = true
        }
}

// Async makes the handler execute asynchronously
func Async() SubscribeOption {
        return func(h *internalHandler) {
                h.async = true
        }
}

// Sequential ensures the handler executes sequentially (with mutex)
func Sequential() SubscribeOption {
        return func(h *internalHandler) {
                h.sequential = true
        }
}

// WithFilter configures the handler to only receive events that match the
// predicate. The predicate runs before handler observability. Its panics are
// isolated like handler panics and sent to PanicHandler; local/non-durable
// delivery continues, while durable Follow/replay retries without checkpointing.
func WithFilter[T any](predicate func(T) bool) SubscribeOption {
        return func(h *internalHandler) {
                h.filter = predicate
        }
}

// Bus Options

// WithPanicHandler sets a panic handler for the event bus
func WithPanicHandler(handler PanicHandler) Option {
        return func(bus *EventBus) {
                bus.panicHandler = handler
        }
}

// WithBeforePublish sets a hook that's called before publishing events
func WithBeforePublish(hook PublishHook) Option {
        return func(bus *EventBus) {
                bus.beforePublish = hook
        }
}

// WithAfterPublish sets a hook that's called after publishing events
func WithAfterPublish(hook PublishHook) Option {
        return func(bus *EventBus) {
                bus.afterPublish = hook
        }
}

// WithBeforePublishContext sets a context-aware hook that's called before publishing events
func WithBeforePublishContext(hook PublishHookContext) Option {
        return func(bus *EventBus) {
                bus.beforePublishCtx = hook
        }
}

// WithAfterPublishContext sets a context-aware hook that's called after publishing events
func WithAfterPublishContext(hook PublishHookContext) Option {
        return func(bus *EventBus) {
                bus.afterPublishCtx = hook
        }
}

// WithStrictPersistence makes persistence a delivery precondition: when a
// publish reports a persistence error (marshal or Append failure), it is NOT
// delivered to handlers immediately. Without this option persistence is
// best-effort — handlers can observe an event the log never recorded, so a
// later replay would diverge from what live handlers saw. Append errors can be
// ambiguous, so a record committed before an acknowledgement failure may
// still appear in a later Replay, Follow, or resumable-subscription delivery.
// Enable this option when the store is the source of truth (event sourcing,
// cross-process delivery).
//
// The failure is still reported to the PersistenceErrorHandler, and
// TryPublish/TryPublishContext return it to the publisher. It has no effect
// on a bus without a store.
func WithStrictPersistence() Option {
        return func(bus *EventBus) {
                bus.strictPersistence = true
        }
}

// WithPersistenceErrorHandler sets the error handler for persistence failures
func WithPersistenceErrorHandler(handler PersistenceErrorHandler) Option {
        return func(bus *EventBus) {
                bus.persistenceErrorHandler = handler
        }
}

// WithPersistenceTimeout sets the timeout for persistence operations
func WithPersistenceTimeout(timeout time.Duration) Option {
        return func(bus *EventBus) {
                bus.persistenceTimeout = timeout
        }
}

// WithReplayBatchSize sets the batch size for Replay operations.
// This controls how many events are read at a time when using a store
// that doesn't implement EventStoreStreamer.
// Default is 100 if not set or set to 0.
func WithReplayBatchSize(size int) Option {
        return func(bus *EventBus) {
                bus.replayBatchSize = size
        }
}

// WithObservability sets the observability implementation for metrics and tracing
func WithObservability(obs Observability) Option {
        return func(bus *EventBus) {
                bus.observability = obs
        }
}

// WithAsyncHandlerLimit bounds the number of concurrently running async
// handler goroutines. When the limit is reached, Publish blocks until a
// running async handler finishes, providing backpressure instead of
// unbounded goroutine growth during publish spikes.
//
// A limit <= 0 means unlimited (the default).
//
// Deadlock warning: do not publish events that have async subscribers from
// inside an async handler when a limit is set. Such a nested Publish blocks
// waiting for a free slot while the publishing handler occupies one; if all
// slots are held by handlers blocked the same way, none can ever be
// released. Publish re-entrantly only from synchronous handlers, or size
// the limit above the maximum possible nesting fan-out.
func WithAsyncHandlerLimit(n int) Option {
        return func(bus *EventBus) {
                if n > 0 {
                        bus.asyncSem = make(chan struct{}, n)
                }
        }
}

// Backward compatibility methods
//
// The Set* methods below write bus fields without synchronization. They are
// intended for configuration before the bus is shared across goroutines;
// calling them concurrently with Publish is a data race.

// SetPanicHandler sets the panic handler.
// Not safe to call concurrently with Publish; configure before use.
//
// Deprecated: use the WithPanicHandler option with New instead.
func (bus *EventBus) SetPanicHandler(handler PanicHandler) {
        bus.panicHandler = handler
}

// SetBeforePublishHook sets the before publish hook.
// Not safe to call concurrently with Publish; configure before use.
//
// Deprecated: use the WithBeforePublish option with New instead.
func (bus *EventBus) SetBeforePublishHook(hook PublishHook) {
        bus.beforePublish = hook
}

// SetAfterPublishHook sets the after publish hook.
// Not safe to call concurrently with Publish; configure before use.
//
// Deprecated: use the WithAfterPublish option with New instead.
func (bus *EventBus) SetAfterPublishHook(hook PublishHook) {
        bus.afterPublish = hook
}

// SetPersistenceErrorHandler sets the persistence error handler (for runtime configuration).
// Not safe to call concurrently with Publish; configure before use.
func (bus *EventBus) SetPersistenceErrorHandler(handler PersistenceErrorHandler) {
        bus.persistenceErrorHandler = handler
}

package eventbus

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "iter"
        "reflect"
        "time"
)

// EventStoreTailer is an optional interface for stores that can push new
// events as they arrive instead of being polled. When the bus's store
// implements it, Follow uses Tail; otherwise it falls back to polling Read
// on the configured interval (see FollowPollInterval).
//
// Contract:
//   - Tail yields events strictly after from, in order, as they become
//     available, blocking between events rather than returning at the tail.
//   - from accepts the same values as Read: OffsetOldest, OffsetNewest
//     (resolved to the tail at call time), or any offset the store issued.
//   - Every yielded StoredEvent.Offset is prefix-safe to persist immediately
//     after that event: resuming may redeliver an already yielded event or
//     indivisible unit, but must never skip a later yielded event. A protocol
//     with only chunk tokens must use the chunk-start token for non-last
//     members and the chunk-end token only for the last member.
//   - The iterator ends after yielding a non-nil error, or silently when ctx
//     is cancelled. Follow restarts a tail that ends for any other reason.
type EventStoreTailer interface {
        Tail(ctx context.Context, from Offset) iter.Seq2[*StoredEvent, error]
}

// followDecoder decodes a stored event's payload into its Go type and
// dispatches it to local handlers. Implementations are closures created by
// registerFollowDecoder, the only place the concrete type is known.
type followDecoder func(ctx context.Context, data json.RawMessage, mode dispatchMode) error

type followDecodeError struct {
        err error
}

func (e *followDecodeError) Error() string {
        return e.err.Error()
}

// registerFollowDecoder records how to decode and locally dispatch events of
// type T, keyed by T's persisted type name. Every generic subscribe entry
// point calls it, so by the time a Follow loop runs, each subscribed type
// can be delivered from the store. It is idempotent for the same Go type. An
// active persistent bus returns an error when two distinct types claim the
// same name; a nonpersistent bus records the ambiguity for a later activation
// audit without constraining reflect.Type delivery. Safe for concurrent use.
func registerFollowDecoder[T any](bus *EventBus) error {
        eventType := reflect.TypeOf((*T)(nil)).Elem()
        typeName := typeNameOf(eventType)
        reservation, err := bus.reservePersistedTypes(persistedTypeSpec{name: typeName, eventType: eventType})
        if err != nil {
                return err
        }
        defer reservation.Rollback()

        if err := installFollowDecoder[T](bus); err != nil {
                return err
        }
        reservation.Commit()
        return nil
}

// installFollowDecoder installs only the runtime decoder. The caller owns any
// durable type reservation; separating the operations lets
// SubscribeWithReplay roll back failed setup without leaving a decoder behind.
//
// Before persistence is enabled, conflicting names mark the decoder slot
// ambiguous instead of constraining ordinary reflect.Type delivery. Once the
// durable type registry is active, a conflict is always returned explicitly.
func installFollowDecoder[T any](bus *EventBus) error {
        eventType := reflect.TypeOf((*T)(nil)).Elem()
        typeName := typeNameOf(eventType)
        active := bus.persistedTypes.isActive()
        bus.followMu.Lock()
        defer bus.followMu.Unlock()
        if existing, ok := bus.followTypes[typeName]; ok {
                if existing == eventType {
                        return nil
                }
                if active {
                        return persistedTypeNameConflict(typeName, existing, eventType)
                }

                // An in-process bus never uses this map. Retain an explicit ambiguous
                // marker and remove the first decoder so enabling persistence cannot
                // silently select whichever conflicting subscription ran first.
                bus.followTypes[typeName] = nil
                delete(bus.followDecoders, typeName)
                return nil
        }
        bus.followTypes[typeName] = eventType
        bus.followDecoders[typeName] = func(ctx context.Context, data json.RawMessage, mode dispatchMode) error {
                var event T
                if err := json.Unmarshal(data, &event); err != nil {
                        // A resumable subscription is represented by an infrastructure marker
                        // in the same concrete-type shard. Wake those markers even though the
                        // ordinary typed decoder cannot produce T: each coordinator must apply
                        // its own ReplaySkip/ReplayAbort policy to this stored envelope. Valid
                        // events continue through dispatchWithMode exactly once below.
                        if markerErr := dispatchInternalHandlers(bus, ctx, eventType, typeName, event, mode); markerErr != nil {
                                return markerErr
                        }
                        return &followDecodeError{err: err}
                }
                return dispatchWithMode(bus, ctx, eventType, typeName, event, mode)
        }
        return nil
}

// dispatchInternalHandlers snapshots and invokes only infrastructure markers.
// It is used on a Follow decode failure, where no concrete user value exists
// but log-backed replay coordinators still need a wake-up. The shard lock is
// released before coordinator code runs, preserving Clear's lock ordering.
func dispatchInternalHandlers(
        bus *EventBus,
        ctx context.Context,
        eventType reflect.Type,
        eventTypeName string,
        event any,
        mode dispatchMode,
) error {
        shard := bus.getShard(eventType)
        shard.mu.RLock()
        var handlers []*internalHandler
        for _, handler := range shard.handlers[eventType] {
                if handler.internalDelivery != nil {
                        handlers = append(handlers, handler)
                }
        }
        shard.mu.RUnlock()

        var errs []error
        for _, handler := range handlers {
                if err := ctx.Err(); err != nil {
                        errs = append(errs, err)
                        break
                }
                deliveryCtx := ctx
                if mode == dispatchWaitForAsync {
                        deliveryCtx = context.WithValue(ctx, durableDispatchCtxKey{}, true)
                }
                if err := callHandlerWithContext(handler, deliveryCtx, event, bus.panicHandler,
                        bus.observability, eventTypeName, false); err != nil {
                        errs = append(errs, err)
                }
        }
        return errors.Join(errs...)
}

// WithLogDelivery makes the store the only delivery path: Publish appends to
// the log and returns without dispatching to local handlers; handlers receive
// events exclusively from a Follow loop tailing that log. Because every
// process — including the publisher — then observes the same log, all of them
// see the same events in the same order, and a publish that failed to persist
// is (by construction) never observed anywhere.
//
// Requirements: WithStore is mandatory (New panics without it), and the
// process must run bus.Follow, or published events are stored but never
// handled locally. Publish hooks and publish-level observability still fire
// at publish time; handler execution happens on the Follow goroutine. Follow
// defaults to OffsetOldest in this mode so events appended before it starts
// are not skipped; use a durable subscription ID to resume across restarts.
func WithLogDelivery() Option {
        return func(bus *EventBus) {
                bus.logDelivery = true
        }
}

// FollowOption configures a Follow loop.
type FollowOption func(*followConfig) error

type followConfig struct {
        from         Offset
        fromSet      bool
        subscription string
        pollInterval time.Duration
        includeOwn   bool
        dedupWindow  int
}

// FollowFrom sets the offset the follower starts reading after. The default
// is OffsetNewest (live-only), except with WithLogDelivery, where it is
// OffsetOldest so events published before Follow starts cannot be lost. With
// FollowWithSubscriptionID, a saved offset takes precedence and FollowFrom
// applies only to the first run (no saved offset yet); without a subscription
// ID it applies to every call. A custom SubscriptionStore needs the optional
// SubscriptionStoreLookup capability to combine a durable ID with an explicit
// starting offset other than OffsetOldest without confusing a legitimate
// OffsetOldest checkpoint with an absent one.
func FollowFrom(from Offset) FollowOption {
        return func(cfg *followConfig) error {
                cfg.from = from
                cfg.fromSet = true
                return nil
        }
}

// FollowWithSubscriptionID makes the follower durable: it resumes from the
// offset saved under id and saves its position as it processes events, so a
// restarted process continues where it left off. Requires a SubscriptionStore
// (WithSubscriptionStore, or a store that implements it). On the first run —
// no saved offset — it starts from FollowFrom if given, else OffsetOldest, and
// saves that concrete initial boundary before consuming. A failure to establish
// that first checkpoint aborts startup; later progress-save failures are
// reported and retain the usual at-least-once retry behavior.
//
// Progress is saved after each processed event; delivery is at-least-once
// across restarts. Use FollowDedupWindow and StoredEvent.ID to absorb the
// duplicates.
//
// One EventBus permits only one active durable owner for an ID across Follow
// and SubscribeWithReplay. Follow releases that ownership when it returns, so
// it can be cancelled and restarted on the same bus. Coordination across bus
// instances/processes still requires an external lease or a single owner.
func FollowWithSubscriptionID(id string) FollowOption {
        return func(cfg *followConfig) error {
                if id == "" {
                        return fmt.Errorf("eventbus: follow subscription ID cannot be empty")
                }
                cfg.subscription = id
                return nil
        }
}

// FollowPollInterval sets how long the follower sleeps between Read calls
// when the store does not implement EventStoreTailer, and how long it backs
// off after a read error on either path. Default 200ms.
func FollowPollInterval(d time.Duration) FollowOption {
        return func(cfg *followConfig) error {
                if d <= 0 {
                        return fmt.Errorf("eventbus: follow poll interval must be positive")
                }
                cfg.pollInterval = d
                return nil
        }
}

// FollowIncludeOwn delivers events this bus instance itself published
// (matched by Origin). By default the follower skips them, because in the
// default delivery mode they were already dispatched locally at publish time
// and would arrive twice. A bus in WithLogDelivery mode always receives its
// own events from the log regardless of this option — there, the log is the
// only delivery path.
func FollowIncludeOwn() FollowOption {
        return func(cfg *followConfig) error {
                cfg.includeOwn = true
                return nil
        }
}

// FollowDedupWindow sets how many recently seen event IDs the follower
// remembers to drop at-least-once duplicates (retried appends, chunk
// re-reads, replays after a crash). Default 1024; 0 disables deduplication.
// Events without an ID (written by pre-envelope versions or external
// producers) are never deduplicated.
func FollowDedupWindow(n int) FollowOption {
        return func(cfg *followConfig) error {
                if n < 0 {
                        return fmt.Errorf("eventbus: follow dedup window cannot be negative")
                }
                cfg.dedupWindow = n
                return nil
        }
}

// dedupRing remembers the last N event IDs seen. Follow runs single-threaded,
// so it needs no locking.
type dedupRing struct {
        ids  map[string]struct{}
        ring []string
        next int
}

func newDedupRing(n int) *dedupRing {
        return &dedupRing{
                ids:  make(map[string]struct{}, n),
                ring: make([]string, n),
        }
}

// observe records id and reports whether it had been seen already.
func (r *dedupRing) observe(id string) bool {
        if _, ok := r.ids[id]; ok {
                return true
        }
        if evicted := r.ring[r.next]; evicted != "" {
                delete(r.ids, evicted)
        }
        r.ring[r.next] = id
        r.next = (r.next + 1) % len(r.ring)
        r.ids[id] = struct{}{}
        return false
}

// forget removes an ID whose delivery failed so retrying the same stored
// event is not mistaken for a completed duplicate.
func (r *dedupRing) forget(id string) {
        delete(r.ids, id)
        for i, observed := range r.ring {
                if observed == id {
                        r.ring[i] = ""
                }
        }
}

// Follow tails the bus's event store and delivers each new event to the
// local handlers subscribed to its type. It is the primitive that turns a
// shared store into a cross-process bus: every process appends by
// publishing and receives by following.
//
// Follow blocks until ctx is cancelled (returning ctx.Err()) or startup
// validation fails. Run it on its own goroutine:
//
//        go func() {
//            if err := bus.Follow(ctx); err != nil && !errors.Is(err, context.Canceled) {
//                log.Printf("follower stopped: %v", err)
//            }
//        }()
//
// Delivery semantics:
//   - Events are dispatched with the same semantics as a local publish —
//     filters, Once, Async, Sequential, and panic recovery all apply — but
//     publish hooks and publish-level observability do not fire (they fired
//     in the publishing process). OffsetFromContext, EventIDFromContext, and
//     MetadataFromContext work inside handlers.
//   - Only event types with at least one prior Subscribe* call in this
//     process can be decoded; events of other types are skipped. Subscribe
//     BEFORE calling Follow — types subscribed later are only picked up from
//     that point in the stream onward.
//   - Events this bus itself published are skipped unless FollowIncludeOwn
//     is given or the bus is in WithLogDelivery mode (see those options).
//   - Registered upcasts are applied before decoding, exactly as in Replay.
//   - Delivery is at-least-once end to end. The follower deduplicates by
//     StoredEvent.ID within FollowDedupWindow; handlers that need stronger
//     guarantees must be idempotent.
//   - A follower with FollowWithSubscriptionID waits for Async handlers for
//     the current event before saving its checkpoint. Publish itself remains
//     non-blocking for Async handlers.
//
// Failure handling: startup checkpoint errors fail the call. Runtime read
// errors and undecodable (poison) events are reported to the
// PersistenceErrorHandler. Generic poison events are skipped; local resumable
// coordinators first apply their own ReplaySkip/ReplayAbort policy, and an
// abort is a durable delivery failure that keeps the outer follower checkpoint
// unchanged. For durable followers, handler panics and upcast failures are
// likewise reported and retried after FollowPollInterval. A non-durable
// follower reports and skips a failed upcast without dispatching the original
// schema. Follow itself returns on ctx cancellation or a startup/configuration
// failure. Shutdown also cancels active followers and waits for them to exit;
// it does not promise to consume the remaining log. Calls started after shutdown
// begins return ErrClosed.
func (bus *EventBus) Follow(ctx context.Context, opts ...FollowOption) error {
        if !bus.beginOperation() {
                return ErrClosed
        }
        defer bus.endOperation()
        ctx, cancel := context.WithCancel(ctx)
        stop := context.AfterFunc(bus.stopContext, cancel)
        defer stop()
        defer cancel()
        if bus.store == nil {
                return fmt.Errorf("eventbus: Follow requires persistence (use WithStore option)")
        }

        defaultFrom := OffsetNewest
        if bus.logDelivery {
                // Publish has no local delivery path in this mode, so starting at a
                // tail resolved inside Follow would create a startup race: an event
                // appended before the follower goroutine first runs would be skipped.
                // Replaying from the beginning is the only complete default when no
                // durable checkpoint or explicit starting position exists.
                defaultFrom = OffsetOldest
        }
        cfg := &followConfig{
                from:         defaultFrom,
                pollInterval: 200 * time.Millisecond,
                dedupWindow:  1024,
        }
        for _, opt := range opts {
                if opt == nil {
                        return fmt.Errorf("eventbus: follow option cannot be nil")
                }
                if err := opt(cfg); err != nil {
                        return err
                }
        }

        // Resolve the durable subscription store, mirroring SubscribeWithReplay.
        var subStore SubscriptionStore
        if cfg.subscription != "" {
                subStore = bus.subscriptionStore
                if subStore == nil {
                        if ss, ok := bus.store.(SubscriptionStore); ok {
                                subStore = ss
                        } else {
                                return fmt.Errorf("eventbus: FollowWithSubscriptionID requires a SubscriptionStore (use WithSubscriptionStore option or use a store that implements SubscriptionStore)")
                        }
                }
                // A scalar checkpoint has exactly one in-process owner. Unlike a
                // successful SubscribeWithReplay reservation (which lasts for the bus's
                // lifetime), a Follow reservation is active only for this blocking call so
                // callers can cancel and restart the follower on the same bus.
                if err := bus.reserveReplayID(cfg.subscription); err != nil {
                        return fmt.Errorf("eventbus: follow subscription %q: %w", cfg.subscription, err)
                }
                defer bus.releaseReplayID(cfg.subscription)
        }

        from := cfg.from
        if subStore != nil {
                var saved Offset
                var found bool
                var err error
                if lookup, ok := subStore.(SubscriptionStoreLookup); ok {
                        saved, found, err = lookup.LookupOffset(ctx, cfg.subscription)
                } else {
                        saved, err = subStore.LoadOffset(ctx, cfg.subscription)
                        found = saved != OffsetOldest
                        if err == nil && !found && cfg.fromSet && cfg.from != OffsetOldest {
                                return fmt.Errorf("eventbus: FollowWithSubscriptionID %q cannot safely combine legacy SubscriptionStore.LoadOffset with FollowFrom(%q): implement SubscriptionStoreLookup to distinguish a missing checkpoint from a saved OffsetOldest", cfg.subscription, cfg.from)
                        }
                }
                if err != nil {
                        return fmt.Errorf("eventbus: load follow offset for %q: %w", cfg.subscription, err)
                }
                if found {
                        if saved == OffsetNewest {
                                return fmt.Errorf("eventbus: load follow offset for %q: symbolic offset %q is not a durable checkpoint", cfg.subscription, OffsetNewest)
                        }
                        from = saved
                } else if !cfg.fromSet {
                        // First run of a durable follower: consume the full log so the
                        // subscription's view is complete, like SubscribeWithReplay.
                        from = OffsetOldest
                }

                // Every first-run boundary must be committed before Tail/Read begins so
                // FollowFrom remains a one-time choice even if no event is yielded. A
                // symbolic live-only boundary is first resolved to one concrete tail;
                // otherwise a restart could resolve "$" later and skip downtime events.
                if !found {
                        if from == OffsetNewest {
                                events, concrete, err := bus.store.Read(ctx, OffsetNewest, 0)
                                if err != nil {
                                        return fmt.Errorf("eventbus: resolve initial follow tail for %q: %w", cfg.subscription, err)
                                }
                                if len(events) != 0 {
                                        return fmt.Errorf("eventbus: resolve initial follow tail for %q: EventStore.Read(OffsetNewest) returned %d event(s), want none", cfg.subscription, len(events))
                                }
                                if concrete == OffsetNewest {
                                        return fmt.Errorf("eventbus: resolve initial follow tail for %q: EventStore.Read(OffsetNewest) returned symbolic offset %q, want a concrete checkpoint", cfg.subscription, OffsetNewest)
                                }
                                from = concrete
                        }
                        if err := ctx.Err(); err != nil {
                                return err
                        }
                        // Establish the first-run boundary even if no event is ever yielded.
                        // This makes FollowFrom a one-time choice and prevents a later call
                        // with different options from silently moving the starting point.
                        if err := subStore.SaveOffset(ctx, cfg.subscription, from); err != nil {
                                return fmt.Errorf("eventbus: save initial follow offset for %q: %w", cfg.subscription, err)
                        }
                }
        }

        var dedup *dedupRing
        if cfg.dedupWindow > 0 {
                dedup = newDedupRing(cfg.dedupWindow)
        }

        f := &follower{bus: bus, cfg: cfg, subStore: subStore, dedup: dedup, offset: from}
        if tailer, ok := bus.store.(EventStoreTailer); ok {
                return f.runTail(ctx, tailer)
        }
        return f.runPoll(ctx)
}

// follower is the running state of one Follow call.
type follower struct {
        bus      *EventBus
        cfg      *followConfig
        subStore SubscriptionStore
        dedup    *dedupRing
        offset   Offset
}

// runPoll reads batches in a sleep loop; the fallback for stores without
// EventStoreTailer support.
func (f *follower) runPoll(ctx context.Context) error {
        batchSize := f.bus.replayBatchSize
        if batchSize <= 0 {
                batchSize = 100
        }

pollLoop:
        for {
                if err := ctx.Err(); err != nil {
                        return err
                }

                readFrom := f.offset
                events, next, err := f.bus.store.Read(ctx, readFrom, batchSize)
                if err != nil {
                        if ctx.Err() != nil {
                                return ctx.Err()
                        }
                        f.reportError(fmt.Errorf("follow: read after offset %s: %w", readFrom, err))
                        if err := sleepCtx(ctx, f.cfg.pollInterval); err != nil {
                                return err
                        }
                        continue
                }

                for _, stored := range events {
                        if err := f.processEvent(ctx, stored); err != nil {
                                if ctx.Err() != nil {
                                        return ctx.Err()
                                }
                                if err := sleepCtx(ctx, f.cfg.pollInterval); err != nil {
                                        return err
                                }
                                continue pollLoop
                        }
                        // Checkpoint every successfully processed envelope. Stores with
                        // per-event offsets avoid replaying an already completed prefix when a
                        // later sibling in this batch fails; chunk-start tokens may repeat and
                        // safely leave the durable cursor unchanged until the chunk boundary.
                        f.advance(ctx, stored.Offset)
                }

                if next == readFrom {
                        // The read made no progress: at the tail (or the store cannot
                        // advance) — wait for new events. Compared against the offset the
                        // batch was read from, not the per-event advances above: a full
                        // batch whose last event carries the batch's resume token is
                        // progress, and the next read must follow immediately — sleeping
                        // per batch would cap catch-up at batchSize/pollInterval.
                        if err := sleepCtx(ctx, f.cfg.pollInterval); err != nil {
                                return err
                        }
                        continue
                }
                if next != f.offset {
                        // A store may advance beyond the last returned envelope (for example,
                        // an undecodable or empty chunk). Persist that concrete resume token
                        // too.
                        f.advance(ctx, next)
                }
        }
}

// runTail consumes a pushing store, restarting the tail after errors.
func (f *follower) runTail(ctx context.Context, tailer EventStoreTailer) error {
        for {
                if err := ctx.Err(); err != nil {
                        return err
                }

                var tailErr error
                for stored, err := range tailer.Tail(ctx, f.offset) {
                        if err != nil {
                                tailErr = err
                                break
                        }
                        if err := f.processEvent(ctx, stored); err != nil {
                                if ctx.Err() != nil {
                                        return ctx.Err()
                                }
                                break
                        }
                        f.advance(ctx, stored.Offset)
                }
                if ctx.Err() != nil {
                        return ctx.Err()
                }
                if tailErr != nil {
                        f.reportError(fmt.Errorf("follow: tail after offset %s: %w", f.offset, tailErr))
                }
                // Either an error or a tail that ended unexpectedly: back off and
                // re-tail from the last processed position.
                if err := sleepCtx(ctx, f.cfg.pollInterval); err != nil {
                        return err
                }
        }
}

// processEvent runs one stored event through dedup, origin filtering,
// upcasting, decoding, and local dispatch. Undeliverable events (duplicates,
// own-origin echoes, unsubscribed types, and generic poison payloads) are
// skipped so a follower can keep up with a heterogeneous stream. A poison
// payload still wakes resumable markers first; ReplayAbort and other durable
// delivery failures are returned so the caller retries without advancing.
func (f *follower) processEvent(ctx context.Context, stored *StoredEvent) error {
        if f.dedup != nil && stored.ID != "" && f.dedup.observe(stored.ID) {
                return nil
        }

        // Skip this bus's own events: in the default delivery mode they were
        // already dispatched locally at publish time. In log-delivery mode the
        // log is the only delivery path, so own events are always taken.
        if stored.Origin == f.bus.originID && !f.cfg.includeOwn && !f.bus.logDelivery {
                return nil
        }

        data, typeName := stored.Data, stored.Type
        if upcasted, upcastedType, err := f.bus.upcastRegistry.apply(data, typeName); err == nil {
                data, typeName = upcasted, upcastedType
        } else {
                failure := fmt.Errorf("follow: upcast failed at offset %s: %w", stored.Offset, err)
                // The typed decoder never runs on an upcast failure, so target-type
                // replay markers would otherwise receive no signal. Wake every resumable
                // coordinator before the outer follower chooses retry (durable) or skip
                // (non-durable); each coordinator scans the stored envelope and owns its
                // own autonomous retry policy.
                signalReplayMarkers(f.bus, contextForStoredEvent(ctx, stored), stored.Type, stored)
                f.reportEvent(stored, failure)
                if f.subStore != nil {
                        f.forget(stored)
                        return failure
                }
                // A non-durable follower has no checkpoint to retain for retry. Skip the
                // failed migration after reporting it, but never dispatch the original
                // schema as though the configured upcast had succeeded.
                return nil
        }

        f.bus.followMu.RLock()
        decode := f.bus.followDecoders[typeName]
        f.bus.followMu.RUnlock()
        if decode == nil {
                return nil // No local subscriber for this type.
        }

        // Hand handlers the same context values a live publish would carry.
        dctx := contextForStoredEvent(ctx, stored)

        mode := dispatchNonBlocking
        if f.subStore != nil {
                // A durable checkpoint is a promise that all work for this event
                // finished. Await only this dispatch's async handlers; bus.Wait would
                // also wait for unrelated publishes and cannot provide that boundary.
                mode = dispatchWaitForAsync
        }
        if err := decode(dctx, data, mode); err != nil {
                if _, ok := err.(*followDecodeError); ok {
                        // Poison event: report and move on. The payload travels with the
                        // report for out-of-band recovery, as in SubscribeWithReplay.
                        f.reportEvent(stored, fmt.Errorf("follow: skipping undecodable event at offset %s: %w", stored.Offset, err))
                        return nil
                }
                if f.subStore == nil {
                        // Non-durable followers retain live-dispatch panic semantics: recovery
                        // isolates the handler and the stream continues. Only a durable
                        // checkpoint needs delivery failure to stop advancement.
                        return nil
                }

                f.forget(stored)
                failure := fmt.Errorf("follow: handler delivery failed at offset %s: %w", stored.Offset, err)
                var panicErr *handlerPanicError
                if ctx.Err() == nil || errors.As(err, &panicErr) {
                        f.reportEvent(stored, failure)
                }
                return failure
        }
        return nil
}

func (f *follower) forget(stored *StoredEvent) {
        if f.dedup != nil && stored.ID != "" {
                f.dedup.forget(stored.ID)
        }
}

// advance records the follower's position, persisting it for durable
// followers. SaveOffset failures are reported and do not stop the follower:
// the position is redundant with the events themselves (at-least-once).
func (f *follower) advance(ctx context.Context, offset Offset) {
        f.offset = offset
        if f.subStore == nil {
                return
        }
        if err := f.subStore.SaveOffset(ctx, f.cfg.subscription, offset); err != nil {
                f.reportError(fmt.Errorf("follow: save offset for %q: %w", f.cfg.subscription, err))
        }
}

// reportError forwards a follower failure to the PersistenceErrorHandler,
// the bus's monitoring channel for storage problems. The event argument is
// nil for failures not tied to a specific event.
func (f *follower) reportError(err error) {
        if f.bus.persistenceErrorHandler != nil {
                f.bus.persistenceErrorHandler(nil, nil, err)
        }
}

// reportEvent forwards a per-event failure, carrying the *StoredEvent so the
// payload can be recovered out of band.
func (f *follower) reportEvent(stored *StoredEvent, err error) {
        if f.bus.persistenceErrorHandler != nil {
                f.bus.persistenceErrorHandler(stored, nil, err)
        }
}

// sleepCtx sleeps for d or until ctx is done, returning ctx.Err() in the
// latter case.
func sleepCtx(ctx context.Context, d time.Duration) error {
        timer := time.NewTimer(d)
        defer timer.Stop()
        select {
        case <-ctx.Done():
                return ctx.Err()
        case <-timer.C:
                return nil
        }
}

package eventbus

import (
        "context"
        "errors"
        "fmt"
)

// ErrClosed is returned when an operation starts after Shutdown begins.
var ErrClosed = errors.New("eventbus: bus is shutting down or closed")

// The high bit closes admission; lower bits count accepted operations. Keeping
// both in one atomic word makes admission race-free without locking the publish
// path. asyncMu still protects the condition variable and async handler count.
const lifecycleStopping uint64 = 1 << 63

func (bus *EventBus) beginOperation() bool {
        for {
                state := bus.lifecycleState.Load()
                if state&lifecycleStopping != 0 {
                        return false
                }
                if bus.lifecycleState.CompareAndSwap(state, state+1) {
                        return true
                }
        }
}

func (bus *EventBus) endOperation() {
        if bus.lifecycleState.Add(^uint64(0)) == lifecycleStopping {
                // The final accepted operation exited during shutdown. Take the waiter's
                // mutex before signaling so a transition to zero cannot lose a wakeup.
                bus.asyncMu.Lock()
                bus.asyncCond.Broadcast()
                bus.asyncMu.Unlock()
        }
}

// Shutdown stops admission of publishes, subscriptions, Replay and Follow.
// New operations return ErrClosed; Publish and PublishContext discard it.
// This includes nested publishes from handlers once shutdown has begun.
// Active Follow calls are canceled. Accepted operations, async handlers and
// deferred resumable-subscription drains finish before the store is closed.
// In log-delivery mode this does not guarantee that Follow consumed the log.
//
// Shutdown is terminal and safe to call concurrently or repeatedly. The store's
// Close method, if present, runs once; all successful waits return its result.
// ctx bounds only the caller's wait: after timeout, draining and eventual close
// continue in the background. A later Shutdown can wait for completion.
// Permanent replay failures or operations ignoring cancellation can prevent
// completion. A blocking store Close also remains subject to the caller's wait.
//
// Do not call Shutdown synchronously from a handler, hook or store operation:
// shutdown waits for that operation to return. Stop external users of GetStore
// and shared stores separately; their operations are not tracked by the bus.
func (bus *EventBus) Shutdown(ctx context.Context) error {
        bus.asyncMu.Lock()
        if bus.lifecycleState.Load()&lifecycleStopping == 0 {
                bus.lifecycleState.Or(lifecycleStopping)
                bus.stopFollow()
                go bus.finishShutdown()
        }
        bus.asyncMu.Unlock()

        select {
        case <-bus.shutdownDone:
                return bus.shutdownErr
        case <-ctx.Done():
                return ctx.Err()
        }
}

func (bus *EventBus) finishShutdown() {
        bus.asyncMu.Lock()
        for bus.lifecycleState.Load() != lifecycleStopping || bus.asyncCount > 0 {
                bus.asyncCond.Wait()
        }
        bus.asyncMu.Unlock()
        // No accepted operation can start another handler or deferred drain now.
        if closer, ok := bus.store.(interface{ Close() error }); ok {
                if err := closer.Close(); err != nil {
                        bus.shutdownErr = fmt.Errorf("failed to close store: %w", err)
                }
        }
        close(bus.shutdownDone)
}

package eventbus

import (
        "context"
        "fmt"
        "time"
)

// MirrorOption configures a Mirror loop.
type MirrorOption func(*mirrorConfig) error

type mirrorConfig struct {
        pollInterval  time.Duration
        dedupWindow   int
        resetOnRewind bool
        onForward     func(*StoredEvent)
        onError       func(error)
        progress      *Replicator
}

// MirrorPollInterval sets how long the mirror sleeps between Read calls when
// the source store does not implement EventStoreTailer, and how long it backs
// off after any error on either path. Default 200ms.
func MirrorPollInterval(d time.Duration) MirrorOption {
        return func(cfg *mirrorConfig) error {
                if d <= 0 {
                        return fmt.Errorf("eventbus: mirror poll interval must be positive")
                }
                cfg.pollInterval = d
                return nil
        }
}

// MirrorDedupWindow sets how many recently forwarded event IDs the mirror
// remembers so that a re-read source batch (a chunk-token re-read after a
// restart or tail restart) does not append the same event to the destination
// twice. Default 1024; 0 disables deduplication. Events without an ID
// (written by pre-envelope versions or external producers) are never
// deduplicated. Failed appends are retried in place and never re-read the
// batch, so the window does not need to cover a whole batch. The window only
// absorbs duplicates seen by one Mirror call: a restart resumes from the
// durable checkpoint, and the events between that checkpoint and the crash
// may be appended again — consumers of the destination deduplicate on
// StoredEvent.ID exactly as they would on the source.
func MirrorDedupWindow(n int) MirrorOption {
        return func(cfg *mirrorConfig) error {
                if n < 0 {
                        return fmt.Errorf("eventbus: mirror dedup window cannot be negative")
                }
                cfg.dedupWindow = n
                return nil
        }
}

// MirrorOnForward registers an observer invoked after each event is
// successfully appended to the destination, before the checkpoint for it is
// saved. Use it for metrics such as forwarded-event counts or replication lag
// (compare StoredEvent.Timestamp with the clock). The callback runs on the
// mirror goroutine: keep it fast, and treat the event as read-only.
func MirrorOnForward(fn func(*StoredEvent)) MirrorOption {
        return func(cfg *mirrorConfig) error {
                if fn == nil {
                        return fmt.Errorf("eventbus: mirror OnForward callback cannot be nil")
                }
                cfg.onForward = fn
                return nil
        }
}

// MirrorOnError registers an observer for runtime failures the mirror absorbs
// and retries: source read errors, destination append errors, checkpoint save
// errors, and reconcile failures. Without it those failures are silent (the
// mirror still retries). The callback runs on the mirror goroutine.
func MirrorOnError(fn func(error)) MirrorOption {
        return func(cfg *mirrorConfig) error {
                if fn == nil {
                        return fmt.Errorf("eventbus: mirror OnError callback cannot be nil")
                }
                cfg.onError = fn
                return nil
        }
}

// MirrorResetOnSourceRewind makes the mirror recover when its saved checkpoint
// no longer resolves in the source because the source log was rebuilt or
// restored from an older copy: a checkpoint past the source's current tail
// would otherwise read as "at the tail" forever and the mirror would silently
// stop forwarding. With this option the mirror compares its checkpoint against
// the source tail and, when the checkpoint is ahead, resets to OffsetOldest
// and re-reads the source from the beginning, reporting the reset to
// MirrorOnError.
//
// When the comparison runs: at startup; on the poll path, on reads that make
// no progress and on read errors; on the tail path, at every tail restart (a
// tail that ends or yields an error — a source whose Tail treats an
// ahead-of-tail offset as a permanently blocked idle long-poll is not
// detected mid-run). Checks are throttled to roughly one per ten poll
// intervals, and the reset itself requires two consecutive checks to agree
// (one transiently stale tail read must not destroy the checkpoint), so
// detection latency is around twenty poll intervals.
//
// What it can and cannot detect: the reset fires only while the restored
// source's tail is still behind the saved checkpoint. A source that was
// restored AND then refilled past the checkpoint is indistinguishable from
// ordinary progress by offsets alone — the mirror resumes at the checkpoint
// and the replaced events below it are not re-forwarded. Detecting that
// requires out-of-band versioning (for example, a per-rebuild log name or
// epoch), which is the caller's design decision, not an offset property.
//
// Requires the source store to implement EventStoreOffsetComparer; Mirror
// fails at startup otherwise. Off by default because the reset re-appends the
// entire surviving source log to the destination — duplicates that downstream
// consumers must absorb by StoredEvent.ID.
func MirrorResetOnSourceRewind() MirrorOption {
        return func(cfg *mirrorConfig) error {
                cfg.resetOnRewind = true
                return nil
        }
}

// Mirror copies every event from src to dst, preserving the stored envelope
// verbatim: ID, Origin, Type, Data, Metadata, and Timestamp are appended to
// dst exactly as read from src (dst assigns its own offsets). It is the
// primitive for replicating one log into another — a local write-ahead log
// shipped into a shared store, a store migration run alongside live traffic,
// or a backup log rebuilt on another machine.
//
// Mirror tails src forever: it blocks until ctx is cancelled (returning
// ctx.Err()) or startup validation fails, using Tail when src implements
// EventStoreTailer and polling Read otherwise. Run it on its own goroutine.
//
// Progress is durable. The mirror resumes from the offset saved under
// subscriptionID in offsets, and checkpoints after each forwarded event; on
// the first run (no saved checkpoint) it starts from OffsetOldest, because a
// mirror's job is the whole log. The checkpoint is a SOURCE offset; offsets
// may be backed by src, dst, or a third store, provided that store accepts
// the source's offset tokens verbatim. A SubscriptionStore that parses or
// validates offsets in its own native format (the bundled SQLite store's
// SaveOffset accepts only its integer offsets) cannot checkpoint a source
// with a different token format: the very first save fails and Mirror ends
// with an error, turning the misconfiguration into an immediate failure
// instead of a run that looks healthy while every restart silently starts
// over from OffsetOldest.
//
// Semantics:
//   - Delivery into dst is at-least-once: a crash between Append and the
//     checkpoint save re-forwards that event on restart. Event IDs are
//     preserved, so consumers of dst deduplicate exactly as they would on src
//     (see FollowDedupWindow). Within one call, a failed append is retried in
//     place — the already forwarded prefix of a batch is never re-read — and
//     MirrorDedupWindow absorbs chunk-token re-reads after restarts. When the
//     source implements EventStoreOffsetComparer, the checkpoint also never
//     moves backward, even when a read redelivers earlier events.
//   - Events are forwarded as raw envelopes: no type registry, no decoding,
//     and no upcasts are involved, so the mirroring process does not need the
//     producers' event types and never skips an unregistered type.
//   - Per-source ordering is preserved: events are appended to dst in source
//     log order, one at a time.
//   - Mirror is one-directional. Two mirrors forming a cycle between stores
//     will copy events forever; nothing filters previously mirrored events.
//   - Runtime failures (read, append, checkpoint save) are reported to
//     MirrorOnError and retried after MirrorPollInterval; the checkpoint does
//     not advance past a failed event. One exception is fatal: a failure to
//     persist the very first checkpoint of a fresh subscription ends Mirror
//     with an error, because it is the signature of a checkpoint store that
//     cannot store the source's offset tokens (see below) — continuing would
//     look healthy while every restart re-copies the whole source.
//   - Source retention must not outpace the mirror. If the source drops
//     events past the saved checkpoint (a remote stream head-trimmed by
//     server-side retention), those events are unrecoverable and the mirror
//     will not silently skip the gap: it keeps reporting the failing read
//     and retrying. Recovery is an operator decision — clear or reseed the
//     checkpoint to pick a new starting point.
//
// Mirror does not coordinate writers: run one mirror per subscriptionID, and
// enforce that yourself — unlike Follow, Mirror is not bound to a bus, so it
// cannot detect a second concurrent mirror on the same ID even in the same
// process. Two concurrent mirrors interleave checkpoint saves; the result is
// re-forwarded duplicates in the destination (each mirror only checkpoints a
// prefix it has itself forwarded), never skipped events. Across processes,
// use an external lease or a single owner.
func Mirror(ctx context.Context, src, dst EventStore, subscriptionID string, offsets SubscriptionStore, opts ...MirrorOption) error {
        if src == nil || dst == nil {
                return fmt.Errorf("eventbus: mirror requires both a source and a destination store")
        }
        if subscriptionID == "" {
                return fmt.Errorf("eventbus: mirror subscription ID cannot be empty")
        }
        if offsets == nil {
                return fmt.Errorf("eventbus: mirror requires a SubscriptionStore for its checkpoint")
        }

        cfg := &mirrorConfig{
                pollInterval: 200 * time.Millisecond,
                dedupWindow:  1024,
        }
        for _, opt := range opts {
                if opt == nil {
                        return fmt.Errorf("eventbus: mirror option cannot be nil")
                }
                if err := opt(cfg); err != nil {
                        return err
                }
        }

        return mirrorRun(ctx, src, dst, subscriptionID, offsets, cfg)
}

func mirrorRun(ctx context.Context, src, dst EventStore, subscriptionID string, offsets SubscriptionStore, cfg *mirrorConfig) error {
        // The comparer is used opportunistically whenever the source can order
        // offsets (the checkpoint monotonicity guard in advance); the rewind
        // option additionally requires it.
        comparer, _ := src.(EventStoreOffsetComparer)
        if cfg.resetOnRewind && comparer == nil {
                return fmt.Errorf("eventbus: MirrorResetOnSourceRewind requires the source store to implement EventStoreOffsetComparer")
        }

        // Resolve the durable checkpoint. The first-run default is OffsetOldest,
        // which doubles as the legacy LoadOffset "absent" value, so a store without
        // SubscriptionStoreLookup resolves identically.
        from := OffsetOldest
        var saved Offset
        var found bool
        var err error
        if lookup, ok := offsets.(SubscriptionStoreLookup); ok {
                saved, found, err = lookup.LookupOffset(ctx, subscriptionID)
        } else {
                saved, err = offsets.LoadOffset(ctx, subscriptionID)
                found = saved != OffsetOldest
        }
        if err != nil {
                return fmt.Errorf("eventbus: load mirror offset for %q: %w", subscriptionID, err)
        }
        if found {
                if saved == OffsetNewest {
                        return fmt.Errorf("eventbus: load mirror offset for %q: symbolic offset %q is not a durable checkpoint", subscriptionID, OffsetNewest)
                }
                from = saved
        }

        var dedup *dedupRing
        if cfg.dedupWindow > 0 {
                dedup = newDedupRing(cfg.dedupWindow)
        }

        m := &mirror{
                src:          src,
                dst:          dst,
                subscription: subscriptionID,
                offsets:      offsets,
                comparer:     comparer,
                cfg:          cfg,
                dedup:        dedup,
                offset:       from,
                // A found checkpoint proves the offsets store accepts the source's
                // tokens; without one, the first save must prove it (see advance).
                saveProven: found,
        }

        if cfg.progress != nil {
                if err := cfg.progress.initialize(ctx, from); err != nil {
                        return err
                }
        }

        // A rebuilt source is detectable before the first read. Like every later
        // reconcile, the check is best-effort: a transient source error here is
        // reported, not fatal — the same error one loop iteration later would be
        // retried forever, and the check reruns on non-progressing polls, read
        // errors, and tail restarts.
        if err := m.maybeReconcile(ctx); err != nil {
                return err
        }

        if tailer, ok := src.(EventStoreTailer); ok && cfg.progress == nil {
                return m.runTail(ctx, tailer)
        }
        return m.runPoll(ctx)
}

// mirror is the running state of one Mirror call.
type mirror struct {
        src          EventStore
        dst          EventStore
        subscription string
        offsets      SubscriptionStore
        comparer     EventStoreOffsetComparer
        cfg          *mirrorConfig
        dedup        *dedupRing
        offset       Offset

        // saveProven records that at least one checkpoint save succeeded (or a
        // saved checkpoint was found at startup): the first save failing is
        // fatal, later failures are tolerated (see advance).
        saveProven bool

        // rewindSeen/rewindFrom implement the two-check confirmation of a source
        // rewind: a reset only fires when two consecutive reconciles observe the
        // same checkpoint ahead of the tail, so one stale tail read (a source
        // behind a cache or replica) cannot destroy the checkpoint.
        rewindSeen bool
        rewindFrom Offset

        // lastReconcile throttles reconcile to roughly one check per ten poll
        // intervals: at the tail the reconcile path runs every poll, and each
        // check costs a tail-resolution round-trip against the source.
        lastReconcile time.Time
}

// runPoll reads batches in a sleep loop; the fallback for sources without
// EventStoreTailer support.
func (m *mirror) runPoll(ctx context.Context) error {
        const batchSize = 100

        for {
                if err := ctx.Err(); err != nil {
                        return err
                }

                readFrom := m.offset
                events, next, err := m.src.Read(ctx, readFrom, batchSize)
                if err != nil {
                        if ctx.Err() != nil {
                                return ctx.Err()
                        }
                        m.report(fmt.Errorf("mirror: read after offset %s: %w", readFrom, err))
                        // A rebuilt source may reject the stale checkpoint with an error
                        // rather than answer with an empty read, so the rewind check must
                        // run on this path too (runTail already reconciles after tail
                        // errors for the same reason).
                        if err := m.maybeReconcile(ctx); err != nil {
                                return err
                        }
                        if err := sleepCtx(ctx, m.cfg.pollInterval); err != nil {
                                return err
                        }
                        continue
                }

                for _, stored := range events {
                        // A failed append retries THIS event in place rather than
                        // re-reading the batch: a re-read redelivers the already forwarded
                        // prefix, and a resumable unit larger than the dedup window would
                        // then land in the destination again on every retry cycle.
                        for {
                                err := m.forward(ctx, stored)
                                if err == nil {
                                        break
                                }
                                if ctx.Err() != nil {
                                        return ctx.Err()
                                }
                                if err := sleepCtx(ctx, m.cfg.pollInterval); err != nil {
                                        return err
                                }
                        }
                        // Checkpoint every forwarded envelope so a later sibling's failure
                        // does not replay an already completed prefix (see follower.runPoll).
                        if err := m.advance(ctx, stored.Offset); err != nil {
                                return err
                        }
                }

                if next == readFrom {
                        // The read made no progress: the mirror is at the tail — or holds a
                        // checkpoint the source no longer knows; only an explicit tail
                        // comparison tells the two apart. (Compared against the offset the
                        // batch was read from, not the per-event advances above: a full
                        // batch whose last event carries the batch's resume token is
                        // progress, and the next read must follow immediately — sleeping
                        // per batch would cap bulk replication at batchSize/pollInterval.)
                        if err := m.maybeReconcile(ctx); err != nil {
                                return err
                        }
                        if err := m.waitForPoll(ctx); err != nil {
                                return err
                        }
                        continue
                }
                if next != m.offset {
                        // A store may advance beyond the last returned envelope (for
                        // example, an empty chunk). Persist that concrete resume token too.
                        if err := m.advance(ctx, next); err != nil {
                                return err
                        }
                }
        }
}

// A Replicator uses Read even for tail-capable stores: Tail cannot expose an
// empty chunk's advanced cursor, so it cannot confirm every captured boundary.
// An explicit waiter wakes idle polling without bypassing failure backoff.
func (m *mirror) waitForPoll(ctx context.Context) error {
        if m.cfg.progress == nil {
                return sleepCtx(ctx, m.cfg.pollInterval)
        }
        timer := time.NewTimer(m.cfg.pollInterval)
        defer timer.Stop()
        select {
        case <-m.cfg.progress.wake:
                return nil
        case <-timer.C:
                return nil
        case <-ctx.Done():
                return ctx.Err()
        }
}

// runTail consumes a pushing source, restarting the tail after errors.
func (m *mirror) runTail(ctx context.Context, tailer EventStoreTailer) error {
        for {
                if err := ctx.Err(); err != nil {
                        return err
                }

                var tailErr, fatal error
                for stored, err := range tailer.Tail(ctx, m.offset) {
                        if err != nil {
                                tailErr = err
                                break
                        }
                        // In-place retry, as in runPoll: re-tailing from the checkpoint
                        // would redeliver the already forwarded prefix of the chunk.
                        for {
                                err := m.forward(ctx, stored)
                                if err == nil {
                                        break
                                }
                                if ctx.Err() != nil {
                                        return ctx.Err()
                                }
                                if err := sleepCtx(ctx, m.cfg.pollInterval); err != nil {
                                        return err
                                }
                        }
                        if err := m.advance(ctx, stored.Offset); err != nil {
                                fatal = err
                                break
                        }
                }
                if fatal != nil {
                        return fatal
                }
                if ctx.Err() != nil {
                        return ctx.Err()
                }
                if tailErr != nil {
                        m.report(fmt.Errorf("mirror: tail after offset %s: %w", m.offset, tailErr))
                }
                // Either an error or a tail that ended unexpectedly: a rebuilt source is
                // one cause of both, so reconcile before re-tailing from the checkpoint.
                if err := m.maybeReconcile(ctx); err != nil {
                        return err
                }
                if err := sleepCtx(ctx, m.cfg.pollInterval); err != nil {
                        return err
                }
        }
}

// forward appends one stored event to the destination with its envelope
// preserved. The destination assigns its own offset; the source offset is
// carried only in the mirror's checkpoint.
func (m *mirror) forward(ctx context.Context, stored *StoredEvent) error {
        if m.dedup != nil && stored.ID != "" && m.dedup.observe(stored.ID) {
                return nil
        }

        event := &Event{
                ID:        stored.ID,
                Origin:    stored.Origin,
                Type:      stored.Type,
                Data:      stored.Data,
                Metadata:  stored.Metadata,
                Timestamp: stored.Timestamp,
        }
        if _, err := m.dst.Append(ctx, event); err != nil {
                // Forget the ID so the retry of this same stored event is not mistaken
                // for a completed duplicate.
                if m.dedup != nil && stored.ID != "" {
                        m.dedup.forget(stored.ID)
                }
                failure := fmt.Errorf("mirror: append event at source offset %s: %w", stored.Offset, err)
                if ctx.Err() == nil {
                        m.report(failure)
                }
                return failure
        }
        if m.cfg.onForward != nil {
                m.cfg.onForward(stored)
        }
        return nil
}

// advance records the mirror's position durably. When the source can order
// offsets (EventStoreOffsetComparer), the checkpoint never moves backward:
// the EventStore contract lets reads redeliver events at or before the
// resume point, and a chunk protocol can interleave an exact embedded offset
// with a lower chunk-start token — persisting the lower one would re-append
// an already checkpointed range after a crash.
//
// For plain Mirror, SaveOffset failures after the first proven save are reported
// and do not stop copying. Replicator instead retries the save in place before
// copying any further event or announcing confirmed progress.
// A failure of the FIRST save is fatal: it is the signature of a checkpoint
// store that cannot store the source's offset tokens at all, and continuing
// would look healthy while every restart re-copies the whole source.
func (m *mirror) advance(ctx context.Context, offset Offset) error {
        if m.comparer != nil && (m.offset != OffsetOldest || m.cfg.progress != nil) {
                cmp, err := m.comparer.CompareOffsets(offset, m.offset)
                if err != nil && m.cfg.progress != nil {
                        return fmt.Errorf("replication: compare progress: %w", err)
                }
                if err == nil && cmp <= 0 {
                        return nil
                }
        }
        // Plain Mirror retains its best-effort checkpoint policy. Confirmed
        // replication cannot advertise progress until the checkpoint save succeeds.
        if m.cfg.progress == nil {
                m.offset = offset
        }
        for {
                err := m.offsets.SaveOffset(ctx, m.subscription, offset)
                if err == nil {
                        break
                }
                if !m.saveProven {
                        return fmt.Errorf("eventbus: mirror %q: initial checkpoint save failed (does the offsets store accept the source's offset tokens?): %w", m.subscription, err)
                }
                m.report(fmt.Errorf("mirror: save offset for %q: %w", m.subscription, err))
                if m.cfg.progress == nil {
                        return nil
                }
                if err := sleepCtx(ctx, m.cfg.pollInterval); err != nil {
                        return err
                }
        }
        m.offset = offset
        m.saveProven = true
        if m.cfg.progress != nil {
                m.cfg.progress.confirm(offset)
        }
        return nil
}

// maybeReconcile runs reconcile when MirrorResetOnSourceRewind is active,
// reporting failures instead of returning them: mid-run they are as transient
// as a read error. Returns only ctx.Err(). Checks are throttled to roughly
// one per ten poll intervals — at the tail this path runs every poll, and
// each check costs a tail-resolution round-trip against the source.
func (m *mirror) maybeReconcile(ctx context.Context) error {
        if !m.cfg.resetOnRewind {
                return nil
        }
        if !m.lastReconcile.IsZero() && time.Since(m.lastReconcile) < 10*m.cfg.pollInterval {
                return nil
        }
        m.lastReconcile = time.Now()
        if err := m.reconcile(ctx); err != nil {
                if ctx.Err() != nil {
                        return ctx.Err()
                }
                m.report(fmt.Errorf("mirror: %w", err))
        }
        return nil
}

// reconcile compares the checkpoint with the source's current tail and resets
// to OffsetOldest when the checkpoint is ahead — the signature of a source log
// that was rebuilt or restored from an older copy. The reset is destructive
// (it re-appends the surviving source history), so it requires TWO
// consecutive checks to observe the same checkpoint ahead of the tail: one
// transiently stale tail read (a source answering from a lagging replica or
// cache) must not destroy the checkpoint.
func (m *mirror) reconcile(ctx context.Context) error {
        if m.offset == OffsetOldest {
                return nil // Nothing can be ahead of the beginning.
        }
        events, tail, err := m.src.Read(ctx, OffsetNewest, 0)
        if err != nil {
                return fmt.Errorf("resolve source tail: %w", err)
        }
        if len(events) != 0 {
                return fmt.Errorf("resolve source tail: EventStore.Read(OffsetNewest) returned %d event(s), want none", len(events))
        }
        if tail == OffsetNewest {
                return fmt.Errorf("resolve source tail: EventStore.Read(OffsetNewest) returned symbolic offset %q, want a concrete checkpoint", OffsetNewest)
        }
        cmp, err := m.comparer.CompareOffsets(m.offset, tail)
        if err != nil {
                return fmt.Errorf("compare checkpoint %q with source tail %q: %w", m.offset, tail, err)
        }
        if cmp <= 0 {
                m.rewindSeen = false
                return nil
        }
        if !m.rewindSeen || m.rewindFrom != m.offset {
                m.rewindSeen, m.rewindFrom = true, m.offset
                m.report(fmt.Errorf("mirror: checkpoint %q is ahead of source tail %q; resetting %q to the start if the next check agrees", m.offset, tail, m.subscription))
                return nil
        }
        m.rewindSeen = false
        m.report(fmt.Errorf("mirror: checkpoint %q is ahead of source tail %q (source log was rebuilt); resetting %q to the start", m.offset, tail, m.subscription))
        // Save directly: the reset intentionally moves the checkpoint backward,
        // which advance's monotonicity guard would refuse.
        m.offset = OffsetOldest
        if err := m.offsets.SaveOffset(ctx, m.subscription, OffsetOldest); err != nil {
                m.report(fmt.Errorf("mirror: save offset for %q: %w", m.subscription, err))
        }
        return nil
}

// report forwards a runtime failure to the MirrorOnError observer, if any.
func (m *mirror) report(err error) {
        if m.cfg.onError != nil {
                m.cfg.onError(err)
        }
}

package eventbus

import (
        "bytes"
        "context"
        "encoding/json"
        "fmt"
        "iter"
        "maps"
        "reflect"
        "sort"
        "sync"
        "time"
)

// Offset represents an opaque position in an event stream.
// Implementations define the format (e.g., "123", "abc_456", timestamp-based).
//
// Offsets are resumption tokens. The bus treats them as opaque: it normally
// performs only equality checks and passes them back to the store that issued
// them. When a store implements EventStoreOffsetComparer, the bus delegates
// same-store ordering checks to it without interpreting token contents. Treat
// offsets from one store as meaningless to any other store.
type Offset string

const (
        // OffsetOldest represents the beginning of the stream.
        // When passed to Read, returns events from the start.
        OffsetOldest Offset = ""

        // OffsetNewest represents the current end of the stream.
        // Useful for subscribing to only new events.
        //
        // OffsetNewest is a query sentinel, not a durable offset. Every EventStore
        // must resolve it, at call time, to the current concrete tail: Read returns
        // no historical events and returns that resumable tail (never "$" itself)
        // as nextOffset. The empty stream resolves to OffsetOldest. Implementations
        // must use an efficient tail lookup rather than scanning or returning the
        // stream's history. ReadStream likewise yields no events and terminates; use
        // EventStoreTailer when later appends should be followed.
        // Persist the concrete nextOffset returned by Read, not OffsetNewest.
        OffsetNewest Offset = "$"
)

// EventStore defines the core interface for persisting events.
// This is a minimal interface with just 2 methods for basic event storage.
// Additional capabilities are provided through optional interfaces.
type EventStore interface {
        // Append stores an event and returns a store-issued, resumable offset.
        // Offsets need not be numeric or unique per event: multiple events may share
        // one token when the store's smallest resumable unit is a chunk. Resuming
        // after the returned token must never skip an event appended later. A returned
        // error does not prove the event is absent: a remote commit may succeed
        // before its acknowledgement is lost. Implementations should use Event.ID
        // to make internal retries idempotent.
        // A replication barrier made from this offset assumes the returned boundary
        // covers the appended event; stores must not return an earlier chunk-start
        // boundary for Append. Durability of a successful Append is store-specific:
        // replication cannot strengthen an in-memory or buffered acknowledgement.
        Append(ctx context.Context, event *Event) (Offset, error)

        // Read returns events starting after the given offset.
        // Use OffsetOldest to read from the beginning. OffsetNewest must be
        // resolved according to its sentinel contract above: return no events and
        // the current concrete tail as nextOffset, without scanning stream history.
        // A positive limit is the requested batch target; zero requests no limit.
        // A store should not exceed a positive limit unless splitting its smallest
        // resumable unit (for example, a chunk with one shared offset) would make
        // nextOffset unsafe. It may return that whole unit to preserve resume-token
        // integrity. If such a unit can advance past an earlier concrete tail token
        // without returning that token as a StoredEvent.Offset or nextOffset, the
        // store must also implement EventStoreOffsetComparer so a bounded consumer
        // can detect the overrun.
        //
        // Every returned StoredEvent.Offset must be prefix-safe to persist
        // immediately after that event is handled: Read(ctx, event.Offset, ...) may
        // redeliver events at or before that point, but must never skip a later event
        // from this result. For a chunk protocol without per-event tokens, assign the
        // read-from/chunk-start token to each non-last member and the chunk-end token
        // only to the last member; assigning the end token to every member can lose
        // the unfinished suffix after a crash. nextOffset must likewise be safe after
        // every returned event has completed.
        // Returns the events, the offset to use for the next read, and any error.
        Read(ctx context.Context, from Offset, limit int) ([]*StoredEvent, Offset, error)
}

// EventStoreOffsetComparer is an optional capability for stores whose opaque
// offsets nevertheless have a store-defined order. CompareOffsets compares two
// concrete offsets issued by the same store and returns a negative value when
// left is before right, zero when equal, and a positive value when left is
// after right. OffsetOldest is a valid concrete boundary; OffsetNewest is a
// symbolic query sentinel and must return an error.
//
// A store must implement this interface when one indivisible Read unit can
// cross a previously issued concrete tail without exposing that exact token as
// either a StoredEvent.Offset or nextOffset. Resumable subscriptions then stop
// immediately after the one unit that crossed their captured tail instead of
// chasing concurrent appends.
// Concrete boundaries must identify stable prefixes within one log generation:
// appending more history must not change the meaning of an already issued tail
// token. Several read events can share an earlier token for safe redelivery;
// that does not make that token evidence of their inclusion in the prefix.
// CompareOffsets must be safe to call concurrently.
type EventStoreOffsetComparer interface {
        CompareOffsets(left, right Offset) (int, error)
}

// EventStoreStreamer is an optional interface for memory-efficient streaming.
// When implemented, the Replay method will automatically use streaming.
//
// Implementation notes:
//   - Database-backed stores should use cursor-based iteration to minimize memory
//   - In-memory stores may need to take a snapshot to avoid holding locks during iteration,
//     trading memory for deadlock safety (see MemoryStore.ReadStream for an example)
type EventStoreStreamer interface {
        // ReadStream returns an iterator yielding events starting after the given offset.
        // Use OffsetOldest to read from the beginning.
        // The iterator checks ctx.Done() before each yield and returns ctx.Err() when cancelled.
        // A yielded error terminates iteration.
        ReadStream(ctx context.Context, from Offset) iter.Seq2[*StoredEvent, error]
}

// SubscriptionStore tracks subscription progress separately from event storage.
// This interface is optional and enables resumable subscriptions.
type SubscriptionStore interface {
        // SaveOffset persists the current offset for a subscription.
        SaveOffset(ctx context.Context, subscriptionID string, offset Offset) error

        // LoadOffset retrieves the last saved offset for a subscription.
        // Returns OffsetOldest if the subscription has no saved offset.
        LoadOffset(ctx context.Context, subscriptionID string) (Offset, error)
}

// SubscriptionStoreLookup is an optional capability that distinguishes a
// missing subscription checkpoint from a checkpoint whose concrete value is
// OffsetOldest. That distinction matters when a durable Follow has an explicit
// first-run starting offset: OffsetOldest is valid for stores whose first
// resume-safe unit starts at the empty token, so it cannot also prove absence.
//
// Implementations should perform the lookup atomically. When found is false,
// offset is ignored and should conventionally be OffsetOldest. A found
// OffsetNewest is invalid because symbolic offsets are not durable resume
// tokens. The bundled MemoryStore and SQLite store implement this interface.
type SubscriptionStoreLookup interface {
        LookupOffset(ctx context.Context, subscriptionID string) (offset Offset, found bool, err error)
}

// EventStoreSnapshotter is an optional interface for stores that can persist a
// materialized projection snapshot, keyed by an id and tagged with the Offset the
// snapshot reflects. It lets a caller compact a high-churn log: save the reduced
// state, then (with EventStoreTruncator) drop the events the snapshot subsumes,
// bounding cold-start replay. The blob is opaque to ebu — callers define its
// encoding. A store that does not implement this interface is simply not
// compactable; callers fall back to a full replay from OffsetOldest.
//
// Implementations may reject snapshot IDs they cannot represent or
// distinguish safely (a store whose snapshot log is externally writable
// rejects the empty ID, which it could not tell apart from foreign data).
// Portable callers should use non-empty IDs.
type EventStoreSnapshotter interface {
        // SaveSnapshot upserts the snapshot for snapshotID, recording that blob
        // reflects the projection state as of (and including) atOffset. atOffset MUST
        // be a real, resumable Offset previously returned by Append/Read for THIS
        // store (never OffsetNewest, never synthetic): a caller resumes with
        // Replay(ctx, atOffset, ...), which reads only events strictly after it.
        SaveSnapshot(ctx context.Context, snapshotID string, atOffset Offset, blob json.RawMessage) error

        // LoadSnapshot returns the last saved snapshot for snapshotID. When none
        // exists it returns (OffsetOldest, nil, nil) — "replay from the beginning",
        // never an error — mirroring LoadOffset's OffsetOldest default.
        LoadSnapshot(ctx context.Context, snapshotID string) (atOffset Offset, blob json.RawMessage, err error)
}

// EventStoreTruncator is an optional interface for log compaction: it deletes
// events at or before a given Offset. It is only safe to call once a snapshot
// covering that Offset is durably saved AND no live reader/subscription still
// needs the truncated prefix. Stores whose Offsets are not deletable positions
// (e.g. a remote append-only broker) MUST NOT implement this interface.
type EventStoreTruncator interface {
        // TruncateBefore deletes every event whose Offset <= beforeOffset and returns
        // the number deleted. beforeOffset == OffsetOldest is a no-op. Idempotent:
        // re-running with the same offset deletes nothing more.
        TruncateBefore(ctx context.Context, beforeOffset Offset) (deleted int64, err error)
}

// Event represents an event to be stored (before it has an offset).
//
// ID, Origin, and Metadata form the event envelope. All three are optional
// at the storage level (`omitempty`), so streams written by earlier ebu
// versions or by external producers decode cleanly with empty values.
type Event struct {
        // ID is a globally unique identifier for the publish that produced this
        // event (a ULID, see NewEventID). The bus assigns it once, before the
        // first Append attempt, so a store that retries a failed append writes
        // the SAME ID for every attempt — consumers can deduplicate at-least-once
        // deliveries on it.
        ID string `json:"id,omitempty"`

        // Origin identifies the bus instance that published the event (each
        // EventBus gets a unique origin at New). On a log shared by several
        // processes it tells "my own event echoed back" apart from a peer's.
        Origin string `json:"origin,omitempty"`

        Type string          `json:"type"`
        Data json.RawMessage `json:"data"`

        // Metadata carries publisher-supplied key/value pairs (correlation IDs,
        // causation IDs, tenant tags, ...). Attach it to the publish context with
        // ContextWithMetadata. The bus never reads or writes keys itself.
        Metadata map[string]string `json:"metadata,omitempty"`

        Timestamp time.Time `json:"timestamp"`
}

// StoredEvent represents an event that has been persisted with an offset.
// ID, Origin, and Metadata mirror the Event envelope; they are empty for
// events written before the envelope existed or by external producers.
type StoredEvent struct {
        // Offset is a store-issued, prefix-safe checkpoint immediately after this
        // event's successful handling; resuming from it may redeliver this event or
        // an earlier indivisible unit, but must not skip any later event.
        Offset    Offset            `json:"offset"`
        ID        string            `json:"id,omitempty"`
        Origin    string            `json:"origin,omitempty"`
        Type      string            `json:"type"`
        Data      json.RawMessage   `json:"data"`
        Metadata  map[string]string `json:"metadata,omitempty"`
        Timestamp time.Time         `json:"timestamp"`
}

// WithStore enables persistence with the given store.
//
// Events are persisted in the publish path, after the before-publish hooks
// and before handlers run. Persistence is best-effort for ordinary Subscribe
// handlers: a failed Append does not prevent their delivery, and the error is
// reported to the PersistenceErrorHandler instead. Log-backed resumable
// subscriptions consume only records visible in EventStore. Because an Append
// error may follow a durable remote commit, that record can still be delivered.
//
// New applies all options before enforcing durable type-name uniqueness, so
// custom options that make typed registrations behave identically on either
// side of WithStore. Applying this option directly to an existing bus panics
// if its provisional typed registrations contain a durable-name collision;
// the bus remains nonpersistent in that case.
func WithStore(store EventStore) Option {
        return func(bus *EventBus) {
                if !bus.configuring && store != nil {
                        bus.activatePersistenceTypes()
                }
                bus.store = store
        }
}

// offsetCtxKey is the context key under which the persisted event's offset is
// stored for live, followed, and context-aware replay delivery.
type offsetCtxKey struct{}

// OffsetFromContext returns the persisted offset of the event currently being
// handled. It is available after a successful live persist, in followed
// handlers, and in handlers registered with SubscribeContextWithReplay.
func OffsetFromContext(ctx context.Context) (Offset, bool) {
        offset, ok := ctx.Value(offsetCtxKey{}).(Offset)
        return offset, ok
}

// eventIDCtxKey is the context key under which the persisted event ID is
// stored for live, followed, and context-aware replay delivery.
type eventIDCtxKey struct{}

// EventIDFromContext returns the ID assigned to the event currently being
// handled. It is set for every publish on a bus configured with WithStore —
// including publishes whose Append failed in best-effort mode, since the ID
// identifies the publish, not the storage outcome — and restored for Follow
// and SubscribeContextWithReplay delivery. Context-aware handlers can use it
// for idempotency keys and correlation.
func EventIDFromContext(ctx context.Context) (string, bool) {
        id, ok := ctx.Value(eventIDCtxKey{}).(string)
        return id, ok
}

// metadataCtxKey is the context key under which publisher-supplied event
// metadata travels from PublishContext to persistEvent.
type metadataCtxKey struct{}

// ContextWithMetadata returns a context that attaches metadata to every event
// subsequently published with it on a bus configured with WithStore. The map
// is stored on the persisted event's envelope (Event.Metadata) and surfaces
// through StoredEvent.Metadata, Follow, and SubscribeContextWithReplay.
//
// The map is not copied: callers must not mutate it after publishing.
// Publishing with a context that already carries metadata replaces the whole
// map (no merging). A nil or empty map attaches nothing.
func ContextWithMetadata(ctx context.Context, md map[string]string) context.Context {
        if len(md) == 0 {
                return ctx
        }
        return context.WithValue(ctx, metadataCtxKey{}, md)
}

// MetadataFromContext returns the event metadata attached to ctx with
// ContextWithMetadata, if any. Inside live, followed, or context-aware replay
// handlers it returns the metadata of the event being delivered.
func MetadataFromContext(ctx context.Context) (map[string]string, bool) {
        md, ok := ctx.Value(metadataCtxKey{}).(map[string]string)
        return md, ok
}

// WithSubscriptionStore enables subscription position tracking
func WithSubscriptionStore(store SubscriptionStore) Option {
        return func(bus *EventBus) {
                bus.subscriptionStore = store
        }
}

// persistEvent saves an event to storage. It returns a context that carries
// the event's assigned ID (always) and its offset (on success, see
// OffsetFromContext), plus the persistence error, if any. Every error is
// reported to the PersistenceErrorHandler before being returned; the caller
// decides — based on WithStrictPersistence — whether the error also stops
// delivery.
//
// Concurrency note: the bus does not serialize Append calls; stores must be
// safe for concurrent use (all bundled stores are). Once an Append call is
// attempted, the event type's durable wire-name claim is retained even when
// Append returns an error, because a remote commit may have succeeded before
// its acknowledgement was lost.
func (bus *EventBus) persistEvent(ctx context.Context, eventType reflect.Type, event any) (context.Context, error) {
        // The ID identifies this publish: it is minted before the first Append
        // attempt so retries inside a store write the same ID, and it is attached
        // to ctx even when persistence fails (see EventIDFromContext).
        eventID := NewEventID()
        ctx = context.WithValue(ctx, eventIDCtxKey{}, eventID)

        // Marshal the event first
        data, err := json.Marshal(event)
        if err != nil {
                err = fmt.Errorf("failed to marshal event: %w", err)
                if bus.persistenceErrorHandler != nil {
                        bus.persistenceErrorHandler(event, eventType, err)
                }
                return ctx, err
        }

        // Use the same type-derived name as subscriptions and typed upcasts. A
        // TypeNamer value is intentionally evaluated on a fresh zero value so one
        // Go type cannot fragment across instance-dependent wire names.
        typeName := typeNameOf(eventType)
        if actualName := EventType(event); actualName != typeName {
                err = fmt.Errorf("eventbus: EventTypeName for Go type %s depends on instance state (%q for this value, %q for the type); EventTypeName must be a pure, immutable function of the type", eventType, actualName, typeName)
                if bus.persistenceErrorHandler != nil {
                        bus.persistenceErrorHandler(event, eventType, err)
                }
                return ctx, err
        }
        typeReservation, err := bus.reservePersistedTypes(persistedTypeSpec{name: typeName, eventType: eventType})
        if err != nil {
                err = fmt.Errorf("failed to reserve persisted event type: %w", err)
                if bus.persistenceErrorHandler != nil {
                        bus.persistenceErrorHandler(event, eventType, err)
                }
                return ctx, err
        }
        defer typeReservation.Rollback()

        metadata, _ := MetadataFromContext(ctx)
        toStore := &Event{
                ID:        eventID,
                Origin:    bus.originID,
                Type:      typeName,
                Data:      data,
                Metadata:  metadata,
                Timestamp: time.Now(),
        }

        // Apply timeout if configured. The timeout applies to Append only; the
        // original context is what gets returned to the publish path.
        appendCtx := ctx
        if bus.persistenceTimeout > 0 {
                var cancel context.CancelFunc
                appendCtx, cancel = context.WithTimeout(ctx, bus.persistenceTimeout)
                defer cancel()
        }

        // Observability: Track persistence start
        if bus.observability != nil {
                appendCtx = bus.observability.OnPersistStart(appendCtx, typeName)
        }

        start := time.Now()

        // Once Append is invoked, its error outcome is inherently ambiguous: a
        // remote store may commit the event and then time out while returning the
        // response. Keep the wire-name mapping sticky before making the call so a
        // later publish cannot reinterpret a possibly-durable payload as another Go
        // type. Pre-append validation and observability failures still roll back via
        // the deferred cleanup above.
        typeReservation.Commit()

        // Append the event - the store assigns the offset.
        offset, saveErr := bus.store.Append(appendCtx, toStore)

        // Observability: Track persistence complete
        if bus.observability != nil {
                bus.observability.OnPersistComplete(appendCtx, typeName, time.Since(start), offset, saveErr)
        }

        if saveErr != nil {
                saveErr = fmt.Errorf("failed to save event: %w", saveErr)
                if bus.persistenceErrorHandler != nil {
                        bus.persistenceErrorHandler(event, eventType, saveErr)
                }
                return ctx, saveErr
        }

        return context.WithValue(ctx, offsetCtxKey{}, offset), nil
}

// Replay replays events from an offset. Shutdown waits for accepted replays;
// replays started after shutdown begins return ErrClosed.
func (bus *EventBus) Replay(ctx context.Context, from Offset, handler func(*StoredEvent) error) error {
        if !bus.beginOperation() {
                return ErrClosed
        }
        defer bus.endOperation()
        if bus.store == nil {
                return fmt.Errorf("replay requires persistence (use WithStore option)")
        }
        if from == OffsetNewest {
                // Replay("$") is an immediate tail boundary, not a moving follower.
                // Resolve it exactly once for parity with ReadStream("$"), then return;
                // continuing from that concrete token could consume an append that races
                // a later fallback Read.
                if err := ctx.Err(); err != nil {
                        return err
                }
                events, concrete, err := bus.store.Read(ctx, OffsetNewest, 0)
                if err != nil {
                        return fmt.Errorf("resolve replay tail: %w", err)
                }
                if err := ctx.Err(); err != nil {
                        return err
                }
                if len(events) != 0 {
                        return fmt.Errorf("resolve replay tail: EventStore.Read(OffsetNewest) returned %d event(s), want none", len(events))
                }
                if concrete == OffsetNewest {
                        return fmt.Errorf("resolve replay tail: EventStore.Read(OffsetNewest) returned symbolic offset %q, want a concrete tail", OffsetNewest)
                }
                return nil
        }

        // Use streaming if available for memory efficiency
        if streamer, ok := bus.store.(EventStoreStreamer); ok {
                for event, err := range streamer.ReadStream(ctx, from) {
                        if err != nil {
                                return fmt.Errorf("stream events: %w", err)
                        }
                        if err := handler(event); err != nil {
                                return fmt.Errorf("handle event at offset %s: %w", event.Offset, err)
                        }
                }
                return nil
        }

        // Fallback to Read for stores that don't support streaming
        batchSize := bus.replayBatchSize
        if batchSize <= 0 {
                batchSize = 100 // Default batch size
        }

        offset := from
        for {
                // Check context cancellation before each batch
                select {
                case <-ctx.Done():
                        return ctx.Err()
                default:
                }

                events, nextOffset, err := bus.store.Read(ctx, offset, batchSize)
                if err != nil {
                        return fmt.Errorf("read events: %w", err)
                }

                if len(events) == 0 {
                        // Zero events does not mean the tail was reached: a store may
                        // advance the offset past a stretch it cannot deliver (e.g. a
                        // remote chunk whose events were all skipped as undecodable).
                        // Only a non-advancing offset marks the end of the stream.
                        if nextOffset == offset {
                                break
                        }
                        offset = nextOffset
                        continue
                }

                for _, event := range events {
                        if err := handler(event); err != nil {
                                return fmt.Errorf("handle event at offset %s: %w", event.Offset, err)
                        }
                }

                // Protect against infinite loop if offset doesn't advance
                if nextOffset == offset {
                        return fmt.Errorf("store returned non-advancing offset %s: possible bug in EventStore.Read implementation", offset)
                }
                offset = nextOffset
        }

        return nil
}

// ReplayWithUpcast replays events from an offset, applying upcasts before
// passing them to handler. An upcast failure aborts replay without delivering
// the original schema as if migration had succeeded.
func (bus *EventBus) ReplayWithUpcast(ctx context.Context, from Offset, handler func(*StoredEvent) error) error {
        return bus.Replay(ctx, from, func(event *StoredEvent) error {
                // Apply upcasts if available
                if bus.upcastRegistry != nil {
                        upcastedData, upcastedType, err := bus.upcastRegistry.apply(event.Data, event.Type)
                        if err != nil {
                                // Delivering the original schema after a configured migration failed
                                // would turn corruption or an upcaster contract violation into a
                                // successful replay. Leave the caller's offset unchanged instead.
                                return err
                        }
                        // Preserve the complete envelope. A shallow copy intentionally
                        // carries ID, Origin, Metadata, and any future StoredEvent fields;
                        // only the schema-dependent type and payload are replaced.
                        upcastedEvent := *event
                        upcastedEvent.Type = upcastedType
                        upcastedEvent.Data = upcastedData
                        return handler(&upcastedEvent)
                }
                return handler(event)
        })
}

// IsPersistent returns true if persistence is enabled
func (bus *EventBus) IsPersistent() bool {
        return bus.store != nil
}

// GetStore returns the event store (or nil if not persistent)
func (bus *EventBus) GetStore() EventStore {
        return bus.store
}

// typeNamerType is the reflect.Type of the TypeNamer interface.
var typeNamerType = reflect.TypeOf((*TypeNamer)(nil)).Elem()

// typeNameOf returns the persisted type name for an event type, honoring the
// TypeNamer interface the same way EventType does for event values.
func typeNameOf(t reflect.Type) string {
        if t.Implements(typeNamerType) {
                if t.Kind() == reflect.Pointer {
                        // Calling a value-receiver method through a typed nil pointer
                        // panics before the method body runs. Use a fresh value so pointer
                        // event types are as safe here as they are at Publish time.
                        return reflect.New(t.Elem()).Interface().(TypeNamer).EventTypeName()
                }
                return reflect.Zero(t).Interface().(TypeNamer).EventTypeName()
        }
        // Do not use a pointer-receiver EventTypeName for a non-pointer T. A value
        // of T does not implement TypeNamer, so EventType(value) persists it under
        // the reflection name. Deriving a custom name here would make replay and
        // Follow look under a name Publish never wrote.
        return t.String()
}

// ReplayErrorPolicy determines how SubscribeWithReplay handles a stored
// event that cannot be decoded into the subscription's event type.
type ReplayErrorPolicy int

const (
        // ReplayAbort stops the replay and returns the decode error (default).
        // The subscription's saved offset does not advance past the failing
        // event, so the next SubscribeWithReplay hits it again. Use this when a
        // decode failure means a bug that must be fixed before proceeding.
        ReplayAbort ReplayErrorPolicy = iota

        // ReplaySkip reports the decode error to the PersistenceErrorHandler
        // (with the *StoredEvent as the event argument) and continues with the
        // next event. The skip is durable: the poison event's offset is saved,
        // so it is not re-scanned and re-reported on later restarts — recover
        // its payload from the reported *StoredEvent if needed.
        //
        // Scope: the policy fires only when a stored event OF THE SUBSCRIBED
        // TYPE fails to decode (malformed or type-incompatible JSON). It does
        // not cover events stored under a different type name — those are
        // always skipped silently, by design, since streams may carry many
        // event types — nor JSON that decodes leniently despite schema drift
        // (unknown fields are dropped, missing fields zero-filled; use upcasts
        // for schema evolution).
        ReplaySkip
)

// WithReplayErrorPolicy sets how SubscribeWithReplay and
// SubscribeContextWithReplay treat stored events that fail to decode in any
// phase, including their log-backed live phase. It has no effect on ordinary
// Subscribe/SubscribeContext delivery. The default is ReplayAbort.
func WithReplayErrorPolicy(policy ReplayErrorPolicy) SubscribeOption {
        return func(h *internalHandler) {
                h.replayErrorPolicy = policy
        }
}

// SubscribeWithReplay subscribes a payload-only handler and replays missed
// events. Use SubscribeContextWithReplay when the handler needs the persisted
// event ID, offset, or metadata for idempotency and correlation.
//
// Both functions require an EventStore and a SubscriptionStore. If the event
// store also implements SubscriptionStore, it is used automatically.
//
// Durable live contract: resumable subscriptions are log consumers, not
// ordinary in-memory callbacks. Every phase drains toward a concrete, bounded
// tail barrier using finite requested batch targets. A store may exceed a
// target only when required to preserve an indivisible resume token. If that
// unit crosses the captured barrier without exposing its exact token, the store
// must implement EventStoreOffsetComparer; the handler can then see at most the
// one crossing unit of concurrent post-barrier events before the drain stops at
// its actual resume token. The handler receives a freshly JSON-decoded value
// plus the exact stored ID, offset, and metadata. Therefore json:"-" fields,
// unexported fields, pointer identity, and other transient properties of the
// published object are intentionally unavailable. An event absent from
// EventStore is not delivered by a resumable subscription; an Append call that
// committed before returning an error may still be delivered. Ordinary
// Subscribe handlers retain the bus's configured best-effort behavior.
//
// Ordering and offsets: one coordinator scans the store in log order and
// saves an event's offset only after successful handler completion. A blocked
// or panicking event therefore prevents a newer checkpoint from leapfrogging
// it. A crash between handling and offset save still redelivers the event, so
// delivery is at-least-once and handlers must remain idempotent.
//
// Context usage: ctx controls setup replay and is visible to historical
// handlers. Live delivery preserves cancellation/deadline behavior from the
// leading publish/follow signal but deliberately strips its arbitrary values;
// envelope accessors always describe the stored event being delivered.
// Coalesced work scheduled after a concurrent or same-type reentrant publish
// uses a background operation context. Under that contention, Publish may
// return before this resumable handler runs; bus.Wait/Shutdown waits for the
// scheduled drain. A live read/handler failure retains that work and retries
// automatically with capped exponential backoff; Clear/ClearAll cancels a
// delayed retry. Ordinary Subscribe handlers retain their configured
// synchronous or asynchronous behavior.
//
// Handoff starts with a concrete, bounded tail barrier. When no local signal
// races that replay, setup also completes the second barrier synchronously. If
// a concurrent or reentrant publish is already pending, setup activates after
// the initial barrier and hands catch-up to the tracked background drain, so a
// hot local producer cannot starve SubscribeWithReplay from returning. In that
// case bus.Wait/Shutdown observes completion. Signals remain behind an inactive
// marker until every setup step succeeds, so a failed setup cannot race live
// user-handler side effects.
//
// SaveOffset failures do not stop delivery; they are reported to the
// PersistenceErrorHandler.
//
// Async and Once are rejected. A resumable subscription can only checkpoint
// safely after synchronous handler completion, and Once has no unambiguous
// durable meaning across process restarts.
//
// Validation: all argument and option validation (nil bus/handler/options,
// empty/previously-used subscription ID, interface event types, filter shape)
// happens before replay, so a validation error has delivered no events and
// saved no offsets. After setup succeeds, its subscription ID remains reserved
// for the EventBus's lifetime (create a new bus to restart it). A setup failure
// releases the ID so the same call can be retried on that bus. Cross-process
// coordination still requires one owner per ID.
//
// Filtering: a WithFilter predicate applies to the replay and catch-up
// passes exactly as it does to live delivery — the handler never sees
// non-matching events, and their offsets are not saved.
//
// Note: If the saved offset is OffsetOldest (""), replay starts from the beginning.
// The OffsetNewest ("$") constant is typically not stored and is only used for live subscriptions.
func SubscribeWithReplay[T any](
        ctx context.Context,
        bus *EventBus,
        subscriptionID string,
        handler Handler[T],
        opts ...SubscribeOption,
) error {
        if handler == nil {
                return fmt.Errorf("eventbus: handler cannot be nil")
        }
        return subscribeContextWithReplay(ctx, bus, subscriptionID,
                func(_ context.Context, event T) { handler(event) }, opts...)
}

// SubscribeContextWithReplay is SubscribeWithReplay for a context-aware
// handler. During historical and catch-up replay, the context exposes the
// StoredEvent envelope through EventIDFromContext, OffsetFromContext, and
// MetadataFromContext, matching live delivery.
func SubscribeContextWithReplay[T any](
        ctx context.Context,
        bus *EventBus,
        subscriptionID string,
        handler ContextHandler[T],
        opts ...SubscribeOption,
) error {
        if handler == nil {
                return fmt.Errorf("eventbus: handler cannot be nil")
        }
        return subscribeContextWithReplay(ctx, bus, subscriptionID, handler, opts...)
}

func subscribeContextWithReplay[T any](
        ctx context.Context,
        bus *EventBus,
        subscriptionID string,
        handler ContextHandler[T],
        opts ...SubscribeOption,
) error {
        if bus == nil {
                return fmt.Errorf("eventbus: bus cannot be nil")
        }
        if !bus.beginOperation() {
                return ErrClosed
        }
        defer bus.endOperation()

        if bus.store == nil {
                return fmt.Errorf("SubscribeWithReplay requires persistence (use WithStore option)")
        }

        // Get subscription store - either explicit or from the event store
        subStore := bus.subscriptionStore
        if subStore == nil {
                if ss, ok := bus.store.(SubscriptionStore); ok {
                        subStore = ss
                } else {
                        return fmt.Errorf("SubscribeWithReplay requires a SubscriptionStore (use WithSubscriptionStore option or use a store that implements SubscriptionStore)")
                }
        }

        eventType := reflect.TypeOf((*T)(nil)).Elem()

        // Build the real user handler before reserving any external identity. The
        // separate signal handler installed later has no user options: its only job
        // is to wake the ordered log consumer.
        h, err := buildContextHandler(handler, eventType, opts)
        if err != nil {
                return err
        }
        if h.async {
                return fmt.Errorf("eventbus: Async cannot be used with SubscribeWithReplay or SubscribeContextWithReplay: durable offsets require synchronous handler completion")
        }
        if h.once {
                return fmt.Errorf("eventbus: Once cannot be used with SubscribeWithReplay or SubscribeContextWithReplay: one-time delivery has no durable meaning across restarts")
        }
        // typeNameOf may instantiate a TypeNamer. Keep it strictly after
        // buildHandler's interface-type validation so a zero interface value can
        // never reach that assertion.
        typeName := typeNameOf(eventType)
        if err := ctx.Err(); err != nil {
                return err
        }

        if err := bus.reserveReplayID(subscriptionID); err != nil {
                return err
        }
        setupSucceeded := false
        defer func() {
                if !setupSucceeded {
                        bus.releaseReplayID(subscriptionID)
                }
        }()

        typeReservation, err := bus.reservePersistedTypes(persistedTypeSpec{name: typeName, eventType: eventType})
        if err != nil {
                return err
        }
        defer typeReservation.Rollback()

        // A failed offset load is not equivalent to a new subscription. Replaying
        // from the beginning in that case could repeat the entire side-effect
        // history and then overwrite the last known checkpoint.
        lastOffset, err := subStore.LoadOffset(ctx, subscriptionID)
        if err != nil {
                return fmt.Errorf("eventbus: load offset for subscription %q: %w", subscriptionID, err)
        }
        if err := ctx.Err(); err != nil {
                return err
        }
        if lastOffset == OffsetNewest {
                return fmt.Errorf("eventbus: load offset for subscription %q: symbolic offset %q is not a durable checkpoint", subscriptionID, OffsetNewest)
        }

        coordinator := &replayCoordinator[T]{
                bus:            bus,
                subStore:       subStore,
                subscriptionID: subscriptionID,
                eventType:      eventType,
                typeName:       typeName,
                userHandler:    h,
                typeClaim:      typeReservation,
                scanOffset:     lastOffset,
                waitCh:         make(chan struct{}),
                done:           make(chan struct{}),
        }

        // The shard contains only an unconditional, silent wake-up handler. The
        // coordinator reads and decodes the durable event representation in store
        // order, then invokes userHandler exactly once through the normal panic,
        // Sequential, and observability boundary.
        marker := &internalHandler{
                handlerType: h.handlerType,
                internalDelivery: func(signalCtx context.Context, _ any) error {
                        return coordinator.signal(signalCtx)
                },
                suppressObservability: true,
        }
        marker.onRemove = func() {
                bus.unregisterReplayMarker(marker)
                coordinator.deactivate()
        }
        // Register before exposing the shard marker. A concurrent predecessor
        // publish can then wake this inactive coordinator; its fixed setup replay
        // will consume that durable record before activation.
        bus.registerReplayMarker(marker)
        bus.addHandler(eventType, marker)

        // Register the inactive marker before the first barrier-bounded replay. Signals
        // coalesce behind setup without waiting, including repeated same-type
        // reentrant publishes from the resumable handler itself.
        if err := coordinator.drainToCapturedTail(ctx); err != nil {
                bus.removeHandler(eventType, marker)
                return fmt.Errorf("replay events: %w", err)
        }

        // Prepare activation under a fresh tail barrier when setup is uncontended.
        // A signal already pending at handoff defers that second catch-up to the
        // tracked drain instead. Either way the marker remains gated until decoder
        // installation and the durable type claim below both succeed.
        needsCatchUp, err := coordinator.activate(ctx)
        if err != nil {
                bus.removeHandler(eventType, marker)
                return fmt.Errorf("catch-up replay: %w", err)
        }

        if err := completeReplaySetup(bus, eventType, marker, coordinator, typeReservation, needsCatchUp); err != nil {
                bus.removeHandler(eventType, marker)
                return err
        }
        setupSucceeded = true
        return nil
}

// completeReplaySetup linearizes successful setup against Clear/ClearAll.
// Those operations remove markers while holding the same shard lock, so they
// either win before this section (and setup rolls back) or run after the
// subscription is fully active. Setup must never return nil for a marker that
// was already removed.
func completeReplaySetup[T any](
        bus *EventBus,
        eventType reflect.Type,
        marker *internalHandler,
        coordinator *replayCoordinator[T],
        typeReservation *persistedTypeReservation,
        needsCatchUp bool,
) error {
        shard := bus.getShard(eventType)
        shard.mu.Lock()
        markerPresent := false
        for _, registered := range shard.handlers[eventType] {
                if registered == marker {
                        markerPresent = true
                        break
                }
        }
        if !markerPresent {
                shard.mu.Unlock()
                coordinator.deactivate()
                return fmt.Errorf("eventbus: resumable subscription %q was removed during setup", coordinator.subscriptionID)
        }

        if err := installFollowDecoder[T](bus); err != nil {
                shard.mu.Unlock()
                coordinator.deactivate()
                return fmt.Errorf("eventbus: install follow decoder: %w", err)
        }
        if !coordinator.finalizeActivation(needsCatchUp) {
                shard.mu.Unlock()
                return fmt.Errorf("eventbus: resumable subscription %q was removed during setup", coordinator.subscriptionID)
        }
        typeReservation.Commit()
        shard.mu.Unlock()
        return nil
}

// replayCoordinator turns SubscribeWithReplay into an ordered log consumer.
// scanOffset is the store read cursor and advances only after a whole batch is
// processed; this is deliberately separate from the durable per-event
// checkpoint because some stores assign several events the same resume-safe
// offset.
type replayCoordinator[T any] struct {
        bus            *EventBus
        subStore       SubscriptionStore
        subscriptionID string
        eventType      reflect.Type
        typeName       string
        userHandler    *internalHandler
        typeClaim      *persistedTypeReservation
        scanOffset     Offset

        mu      sync.Mutex
        active  bool
        closed  bool
        running bool
        pending bool
        tracked bool // pending-work token spanning first deferral through drain/drop
        waitCh  chan struct{}
        done    chan struct{}
        // retryWaiting coalesces new signals behind the single delayed retry instead
        // of allocating one timer per publish during a store outage.
        retryWaiting bool

        // retryDelay backs off autonomous retries after a live drain failure. A
        // successful drain resets it. The pending-work token remains held across
        // the timer, so Wait and Shutdown cannot mistake a failed wake-up for
        // completed delivery.
        retryDelay time.Duration
}

const (
        replayRetryInitialDelay = 10 * time.Millisecond
        replayRetryMaxDelay     = time.Second
)

// durableDispatchCtxKey marks a dispatch whose caller must not checkpoint
// until every internal delivery layer (including a coalescing replay
// coordinator) has actually finished.
type durableDispatchCtxKey struct{}

func (c *replayCoordinator[T]) saveOffset(ctx context.Context, event T, offset Offset) {
        if err := c.subStore.SaveOffset(ctx, c.subscriptionID, offset); err != nil && c.bus.persistenceErrorHandler != nil {
                c.bus.persistenceErrorHandler(event, c.eventType,
                        fmt.Errorf("failed to save offset for subscription %q: %w", c.subscriptionID, err))
        }
}

func (c *replayCoordinator[T]) processStored(ctx context.Context, stored *StoredEvent) error {
        eventData, eventTypeName := stored.Data, stored.Type
        if c.bus.upcastRegistry != nil {
                var err error
                eventData, eventTypeName, err = c.bus.upcastRegistry.apply(eventData, eventTypeName)
                if err != nil {
                        return fmt.Errorf("upcast event at offset %s: %w", stored.Offset, err)
                }
        }
        if eventTypeName != c.typeName {
                return nil
        }

        var event T
        if err := json.Unmarshal(eventData, &event); err != nil {
                if c.userHandler.replayErrorPolicy != ReplaySkip {
                        return fmt.Errorf("decode event at offset %s: %w", stored.Offset, err)
                }
                if c.bus.persistenceErrorHandler != nil {
                        c.bus.persistenceErrorHandler(stored, c.eventType,
                                fmt.Errorf("skipping undecodable event at offset %s for subscription %q: %w", stored.Offset, c.subscriptionID, err))
                }
                var zero T
                skipCtx := contextForStoredEvent(ctx, stored)
                if err := skipCtx.Err(); err != nil {
                        return err
                }
                c.saveOffset(skipCtx, zero, stored.Offset)
                return nil
        }

        // A successful decode is durable evidence for this Go type even if a
        // filter rejects it or a later handler/read fails. Keep that identity
        // sticky so another type can never reinterpret the proven history.
        c.typeClaim.Commit()
        if c.userHandler.filter != nil {
                matches, err := callFilter(c.userHandler, event, c.bus.panicHandler)
                if err != nil {
                        return err
                }
                if !matches {
                        return nil
                }
        }

        deliveryCtx := contextForStoredEvent(ctx, stored)
        if err := callHandlerWithContext(c.userHandler, deliveryCtx, event, c.bus.panicHandler,
                c.bus.observability, eventTypeName, false); err != nil {
                return err
        }
        if err := deliveryCtx.Err(); err != nil {
                return err
        }
        c.saveOffset(deliveryCtx, event, stored.Offset)
        return nil
}

// reachedCapturedTail asks the store to order its own opaque offsets when it
// exposes that optional capability. Equality remains sufficient for stores
// whose Read results always expose the exact captured tail token.
func (c *replayCoordinator[T]) reachedCapturedTail(offset, barrier Offset) (bool, error) {
        if offset == barrier {
                return true, nil
        }
        comparer, ok := c.bus.store.(EventStoreOffsetComparer)
        if !ok {
                return false, nil
        }
        order, err := comparer.CompareOffsets(offset, barrier)
        if err != nil {
                return false, fmt.Errorf("compare replay offset %q with captured tail %q: %w", offset, barrier, err)
        }
        return order >= 0, nil
}

// drainToCapturedTail snapshots the current concrete tail, then issues
// EventStore.Read calls with finite batch targets until that barrier is
// processed. A store may exceed a target to keep an indivisible resume token
// safe; a store-provided offset comparer bounds any overrun to that one crossing
// unit. It never uses ReadStream: a streaming implementation may keep observing
// concurrent appends and starve subscription setup indefinitely.
func (c *replayCoordinator[T]) drainToCapturedTail(ctx context.Context) (err error) {
        // EventStore, upcast, filter, and lifecycle callbacks are extension points.
        // A panic from any of them must become a failed durable attempt rather than
        // unwinding past signal's running/token cleanup and wedging the coordinator.
        defer func() {
                if recovered := recover(); recovered != nil {
                        err = fmt.Errorf("resumable subscription %q drain panic: %v", c.subscriptionID, recovered)
                }
        }()

        if err := ctx.Err(); err != nil {
                return err
        }
        if c.isClosed() {
                return nil
        }
        events, barrier, err := c.bus.store.Read(ctx, OffsetNewest, 0)
        if err != nil {
                return fmt.Errorf("capture replay tail: %w", err)
        }
        if err := ctx.Err(); err != nil {
                return err
        }
        if barrier == OffsetNewest {
                return fmt.Errorf("capture replay tail: EventStore.Read(OffsetNewest) returned symbolic offset %q; a concrete resumable tail is required", OffsetNewest)
        }
        if len(events) != 0 {
                return fmt.Errorf("capture replay tail: EventStore.Read(OffsetNewest) returned %d event(s), want none", len(events))
        }
        reached, err := c.reachedCapturedTail(c.scanOffset, barrier)
        if err != nil {
                return err
        }
        if reached {
                return nil
        }

        batchSize := c.bus.replayBatchSize
        if batchSize <= 0 {
                batchSize = 100
        }
        for {
                if err := ctx.Err(); err != nil {
                        return err
                }
                if c.isClosed() {
                        return nil
                }
                batchStart := c.scanOffset
                events, next, err := c.bus.store.Read(ctx, batchStart, batchSize)
                if err != nil {
                        return fmt.Errorf("read events after %s: %w", batchStart, err)
                }
                if len(events) == 0 {
                        if next == batchStart {
                                return nil
                        }
                        c.scanOffset = next
                        reached, err := c.reachedCapturedTail(c.scanOffset, barrier)
                        if err != nil {
                                return err
                        }
                        if reached {
                                return nil
                        }
                        continue
                }

                for _, stored := range events {
                        if err := ctx.Err(); err != nil {
                                return err
                        }
                        if c.isClosed() {
                                return nil
                        }
                        if err := c.processStored(ctx, stored); err != nil {
                                // Keep scanOffset at batchStart. Stores with chunk-level offsets
                                // can then redeliver the whole batch without skipping a failed
                                // sibling that shares the same offset.
                                return err
                        }
                        if err := ctx.Err(); err != nil {
                                return err
                        }
                        reached, err := c.reachedCapturedTail(stored.Offset, barrier)
                        if err != nil {
                                return err
                        }
                        if reached {
                                // Keep the actual processed, resume-safe token. A comparer may
                                // report that an indivisible Read unit crossed the captured tail.
                                c.scanOffset = stored.Offset
                                return nil
                        }
                }
                if next == batchStart {
                        return fmt.Errorf("store returned non-advancing offset %s while draining subscription %q", batchStart, c.subscriptionID)
                }
                c.scanOffset = next
                reached, err := c.reachedCapturedTail(c.scanOffset, barrier)
                if err != nil {
                        return err
                }
                if reached {
                        return nil
                }
        }
}

func (c *replayCoordinator[T]) isClosed() bool {
        c.mu.Lock()
        closed := c.closed
        c.mu.Unlock()
        return closed
}

// notifyWaitersLocked advances the completion generation. Closing and
// replacing the channel while holding mu gives deactivate and a completing
// leader a single owner for every close, even when Clear races a live drain.
func (c *replayCoordinator[T]) notifyWaitersLocked() {
        close(c.waitCh)
        c.waitCh = make(chan struct{})
}

// activate prepares the live handoff while the coordinator remains gated.
// When setup-time publishes are already pending, their local marker proves a
// second synchronous barrier could chase a hot producer, so final setup hands
// that catch-up to the tracked background drain instead.
func (c *replayCoordinator[T]) activate(ctx context.Context) (needsCatchUp bool, err error) {
        c.mu.Lock()
        if c.closed {
                c.mu.Unlock()
                return false, fmt.Errorf("resumable subscription %q was removed during setup", c.subscriptionID)
        }
        if c.pending {
                c.mu.Unlock()
                return true, nil
        }
        c.running = true
        c.mu.Unlock()

        err = c.drainToCapturedTail(ctx)
        c.mu.Lock()
        c.running = false
        if c.closed && err == nil {
                err = fmt.Errorf("resumable subscription %q was removed during setup", c.subscriptionID)
        }
        if err != nil {
                c.closed = true
        }
        needsCatchUp = err == nil && c.pending
        finishTracked := c.tracked && err != nil
        if finishTracked {
                c.tracked = false
        }
        if err != nil {
                c.notifyWaitersLocked()
        }
        c.mu.Unlock()
        if finishTracked {
                c.bus.asyncFinished()
        }
        return needsCatchUp, err
}

// finalizeActivation opens the marker only after every remaining setup step
// has succeeded. A pending-work token already bridges to scheduleDrain; create
// one defensively if a synthetic coordinator violates that invariant.
func (c *replayCoordinator[T]) finalizeActivation(needsCatchUp bool) bool {
        c.mu.Lock()
        if c.closed {
                c.mu.Unlock()
                return false
        }
        c.active = true
        needsCatchUp = needsCatchUp || c.pending
        if needsCatchUp && !c.tracked {
                c.bus.asyncStarted()
                c.tracked = true
        }
        if needsCatchUp {
                // Reserve the goroutine token before waking contenders. One of them may
                // otherwise consume the pending token before scheduleDrain is accounted.
                c.scheduleDrain(0)
        }
        c.notifyWaitersLocked()
        c.mu.Unlock()
        return true
}

func (c *replayCoordinator[T]) deactivate() {
        c.mu.Lock()
        if c.closed {
                c.mu.Unlock()
                return
        }
        c.closed = true
        c.active = false
        c.pending = false
        c.retryWaiting = false
        if c.done != nil {
                close(c.done)
        }
        finishTracked := c.tracked && !c.running
        if finishTracked {
                c.tracked = false
        }
        c.notifyWaitersLocked()
        c.mu.Unlock()
        if finishTracked {
                c.bus.asyncFinished()
        }
}

// signal coalesces concurrent and reentrant publishes. The leader drains to a
// captured tail barrier; followers return after recording pending work,
// avoiding a self-deadlock when a resumable handler publishes its own event
// type.
func (c *replayCoordinator[T]) signal(ctx context.Context) error {
        mustWait, _ := ctx.Value(durableDispatchCtxKey{}).(bool)
        for {
                c.mu.Lock()
                if c.closed {
                        c.mu.Unlock()
                        return nil
                }
                if !c.active || c.running || c.retryWaiting {
                        c.pending = true
                        if !c.tracked {
                                // Register before unlocking: a coalesced publisher may return as
                                // soon as mu is released, and Wait must already see its deferred
                                // replay work at that point.
                                c.bus.asyncStarted()
                                c.tracked = true
                        }
                        if !mustWait {
                                c.mu.Unlock()
                                return nil
                        }
                        waitCh := c.waitCh
                        c.mu.Unlock()
                        select {
                        case <-waitCh:
                                continue
                        case <-ctx.Done():
                                return ctx.Err()
                        }
                }
                c.running = true
                c.pending = false
                c.mu.Unlock()
                break
        }

        // Live resumable delivery is defined by the stored envelope, not arbitrary
        // values on whichever publisher happened to lead this drain. Preserve only
        // cancellation/deadline behavior from that operation context.
        err := c.drainToCapturedTail(valueStrippedContext{Context: ctx})
        c.mu.Lock()
        c.running = false
        pending := c.pending
        c.pending = false
        closed := c.closed
        schedule := !closed && (pending || err != nil)
        retryDelay := time.Duration(0)
        if err != nil && !closed {
                if c.retryDelay == 0 {
                        c.retryDelay = replayRetryInitialDelay
                } else {
                        c.retryDelay = min(c.retryDelay*2, replayRetryMaxDelay)
                }
                retryDelay = c.retryDelay
                c.retryWaiting = true
        } else if err == nil {
                c.retryDelay = 0
                c.retryWaiting = false
        }
        if schedule && !c.tracked {
                // The leading synchronous signal has no token of its own. Once it
                // defers a retry, account for that pending work before it returns.
                c.bus.asyncStarted()
                c.tracked = true
        }
        finishTracked := c.tracked && !schedule
        if finishTracked {
                c.tracked = false
        }
        if schedule {
                // Start accounting while leadership is still closed under mu. The new
                // goroutine can safely block on mu until this completion is published.
                c.scheduleDrain(retryDelay)
        }
        c.notifyWaitersLocked()
        c.mu.Unlock()
        if closed {
                err = nil
        }
        if err != nil && c.bus.persistenceErrorHandler != nil {
                c.bus.persistenceErrorHandler(nil, c.eventType,
                        fmt.Errorf("resumable subscription %q live drain: %w", c.subscriptionID, err))
        }
        if finishTracked {
                c.bus.asyncFinished()
        }
        return err
}

func (c *replayCoordinator[T]) scheduleDrain(delay time.Duration) {
        // The pending-work token bridges the gap up to this call. Keep a separate
        // goroutine token as well: a woken durable waiter can win leadership and
        // release the pending token before this goroutine gets scheduled.
        c.bus.asyncStarted()
        go func() {
                defer c.bus.asyncFinished()
                if delay > 0 {
                        timer := time.NewTimer(delay)
                        defer timer.Stop()
                        select {
                        case <-timer.C:
                        case <-c.done:
                                return
                        }
                        c.mu.Lock()
                        c.retryWaiting = false
                        c.mu.Unlock()
                }
                _ = c.signal(context.Background())
        }()
}

type valueStrippedContext struct {
        context.Context
}

func (valueStrippedContext) Value(any) any { return nil }

// storedDeliveryContext masks any envelope values carried by the operation
// context before restoring the exact stored event. Without masking, an empty
// field on an older event could accidentally inherit ID/metadata from the
// publisher whose signal happened to lead a coalesced drain.
type storedDeliveryContext struct {
        context.Context
        stored *StoredEvent
}

func (c storedDeliveryContext) Value(key any) any {
        switch key.(type) {
        case offsetCtxKey:
                // OffsetOldest is also a valid resume token on stores whose first
                // chunk assigns chunk-start offsets. Preserve that exact empty Offset
                // while still masking any stale offset carried by the signal context.
                return c.stored.Offset
        case eventIDCtxKey:
                if c.stored.ID == "" {
                        return nil
                }
                return c.stored.ID
        case metadataCtxKey:
                if len(c.stored.Metadata) == 0 {
                        return nil
                }
                return c.stored.Metadata
        default:
                return c.Context.Value(key)
        }
}

// contextForStoredEvent gives replayed and followed handlers the same envelope
// accessors as handlers on a successful live publish while preserving
// cancellation, deadlines, and unrelated caller values.
func contextForStoredEvent(ctx context.Context, stored *StoredEvent) context.Context {
        return storedDeliveryContext{Context: ctx, stored: stored}
}

// MemoryStore is a simple in-memory implementation of EventStore and
// SubscriptionStore. It owns copies of mutable payload and metadata values:
// callers may reuse an Event after Append or mutate a Read/ReadStream result
// without changing the stored history.
type MemoryStore struct {
        events        []*StoredEvent
        subscriptions map[string]Offset
        nextOffset    int64
        mu            sync.RWMutex
}

// Ensure MemoryStore implements all required interfaces
var _ EventStore = (*MemoryStore)(nil)
var _ EventStoreOffsetComparer = (*MemoryStore)(nil)
var _ EventStoreStreamer = (*MemoryStore)(nil)
var _ SubscriptionStore = (*MemoryStore)(nil)
var _ SubscriptionStoreLookup = (*MemoryStore)(nil)

// NewMemoryStore creates a new in-memory event store
func NewMemoryStore() *MemoryStore {
        return &MemoryStore{
                events:        make([]*StoredEvent, 0),
                subscriptions: make(map[string]Offset),
        }
}

// cloneStoredEvent transfers ownership of a stored envelope across a
// MemoryStore boundary. StoredEvent itself contains a mutable byte slice and
// map, so copying only its pointer (or struct header) would let callers rewrite
// the store's history after Append or through a Read result.
func cloneStoredEvent(event *StoredEvent) *StoredEvent {
        cloned := *event
        cloned.Data = bytes.Clone(event.Data)
        cloned.Metadata = maps.Clone(event.Metadata)
        return &cloned
}

// Append implements EventStore
func (m *MemoryStore) Append(ctx context.Context, event *Event) (Offset, error) {
        m.mu.Lock()
        defer m.mu.Unlock()

        m.nextOffset++
        // Use zero-padded format for correct lexicographic ordering
        offset := Offset(fmt.Sprintf("%020d", m.nextOffset))

        stored := &StoredEvent{
                Offset:    offset,
                ID:        event.ID,
                Origin:    event.Origin,
                Type:      event.Type,
                Data:      bytes.Clone(event.Data),
                Metadata:  maps.Clone(event.Metadata),
                Timestamp: event.Timestamp,
        }
        m.events = append(m.events, stored)
        return offset, nil
}

// searchAfter returns the index of the first event with offset > from.
// Events are stored in ascending, zero-padded offset order, so binary
// search applies. Caller must hold at least a read lock.
func (m *MemoryStore) searchAfter(from Offset) int {
        if from == OffsetOldest {
                return 0
        }
        if from == OffsetNewest {
                // "$" is not a stored offset ("$" sorts before the zero-padded
                // digits, so the binary search would wrongly return 0 = replay
                // everything); the tail means "after every current event".
                return len(m.events)
        }
        return sort.Search(len(m.events), func(i int) bool {
                return m.events[i].Offset > from
        })
}

// tailOffset returns the offset of the last stored event, or OffsetOldest
// when the store is empty. Caller must hold at least a read lock.
func (m *MemoryStore) tailOffset() Offset {
        if len(m.events) == 0 {
                return OffsetOldest
        }
        return m.events[len(m.events)-1].Offset
}

// Read implements EventStore
func (m *MemoryStore) Read(ctx context.Context, from Offset, limit int) ([]*StoredEvent, Offset, error) {
        m.mu.RLock()
        defer m.mu.RUnlock()

        start := m.searchAfter(from)
        end := len(m.events)
        if limit > 0 && start+limit < end {
                end = start + limit
        }

        if start == end {
                if from == OffsetNewest {
                        // Resolve "$" to a concrete, resumable position: echoing the
                        // symbolic offset back would make the caller chase a
                        // perpetually moving tail.
                        return nil, m.tailOffset(), nil
                }
                return nil, from, nil
        }

        result := make([]*StoredEvent, end-start)
        for i, event := range m.events[start:end] {
                result[i] = cloneStoredEvent(event)
        }
        return result, result[len(result)-1].Offset, nil
}

// CompareOffsets orders two concrete offsets issued by this MemoryStore.
// OffsetNewest is a symbolic query sentinel and cannot be compared.
func (m *MemoryStore) CompareOffsets(left, right Offset) (int, error) {
        if left == OffsetNewest || right == OffsetNewest {
                return 0, fmt.Errorf("memory store cannot compare symbolic offset %q", OffsetNewest)
        }
        switch {
        case left < right:
                return -1, nil
        case left > right:
                return 1, nil
        default:
                return 0, nil
        }
}

// SaveOffset implements SubscriptionStore.
// OffsetNewest is resolved to the current tail at save time, so the stored
// value is always a concrete, resumable position.
func (m *MemoryStore) SaveOffset(ctx context.Context, subscriptionID string, offset Offset) error {
        m.mu.Lock()
        defer m.mu.Unlock()

        if offset == OffsetNewest {
                offset = m.tailOffset()
        }
        m.subscriptions[subscriptionID] = offset
        return nil
}

// LookupOffset implements SubscriptionStoreLookup.
func (m *MemoryStore) LookupOffset(ctx context.Context, subscriptionID string) (Offset, bool, error) {
        m.mu.RLock()
        defer m.mu.RUnlock()

        offset, ok := m.subscriptions[subscriptionID]
        if !ok {
                return OffsetOldest, false, nil
        }
        return offset, true, nil
}

// LoadOffset implements SubscriptionStore.
func (m *MemoryStore) LoadOffset(ctx context.Context, subscriptionID string) (Offset, error) {
        offset, _, err := m.LookupOffset(ctx, subscriptionID)
        return offset, err
}

// ReadStream implements EventStoreStreamer for memory-efficient event iteration.
// Note: This takes a filtered snapshot of matching events to avoid holding the lock
// during iteration, which could cause deadlocks if handlers call other store methods.
func (m *MemoryStore) ReadStream(ctx context.Context, from Offset) iter.Seq2[*StoredEvent, error] {
        return func(yield func(*StoredEvent, error) bool) {
                // Take a snapshot to avoid holding lock during iteration
                m.mu.RLock()
                start := m.searchAfter(from)
                events := make([]*StoredEvent, len(m.events)-start)
                for i, event := range m.events[start:] {
                        events[i] = cloneStoredEvent(event)
                }
                m.mu.RUnlock()

                for _, event := range events {
                        // Check context cancellation
                        select {
                        case <-ctx.Done():
                                yield(nil, ctx.Err())
                                return
                        default:
                        }

                        if !yield(event, nil) {
                                return // Consumer stopped iteration
                        }
                }
        }
}

package eventbus

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "sync"
        "time"
        "unicode/utf8"
)

var (
        // ErrReplicationPosition identifies an invalid position or a position from a
        // different replication identity/generation.
        ErrReplicationPosition = errors.New("eventbus: invalid replication position")
        // ErrReplicationHistory means a saved checkpoint is ahead of the source tail.
        // Restore the matching source or start a new generation; do not reuse offsets.
        ErrReplicationHistory = errors.New("eventbus: replication history changed")
        // ErrReplicatorStarted means Run was already called on this object.
        ErrReplicatorStarted = errors.New("eventbus: replicator already started")
        // ErrReplicationStopped means Run ended before the requested point was confirmed.
        ErrReplicationStopped = errors.New("eventbus: replication stopped")
)

// ReplicationPosition identifies a SOURCE boundary, scoped to one immutable
// source/destination pair and generation. It can be serialized across restarts.
// Use Replicator.Position for an Append result or Capture for a group of writes.
// Never use a non-last chunk member's StoredEvent.Offset as proof of that
// member's replication: such offsets may refer to the start of the chunk.
type ReplicationPosition struct {
        ID         string `json:"id"`
        Generation string `json:"generation"`
        Offset     Offset `json:"offset"`
}

// ReplicatorConfig binds one replication relationship. ID and Generation must
// be valid UTF-8 strings for unambiguous persisted/serialized identity. ID identifies the
// source/destination pair. Generation is an application-owned immutable epoch:
// change it after replacing, rebuilding or restoring EITHER log. Epoch changes
// require a new Replicator and checkpoint namespace; they cannot be inferred
// from opaque offsets, especially after a rebuilt log has caught up again.
//
// Source must implement EventStoreOffsetComparer, expose every event without
// filtering/skipping decode failures, and retain all history not yet replicated.
// A fresh generation starts at OffsetOldest and requires the full source history. Source and destination must be different logs with no
// cyclic mirror topology. Destination must preserve all acknowledged records (or
// a recovery-equivalent snapshot). There must be one owner of the checkpoint
// namespace across processes; Replicator does not acquire distributed leases.
//
// Wait inherits Destination.Append's acknowledgement guarantees. For survival
// of source-worker loss, Destination must be outside that worker's failure
// domain and acknowledge only after meeting your storage durability policy.
// An in-memory or buffered destination does not become durable through this API.
// Checkpoints must durably store SOURCE offsets verbatim; losing a checkpoint
// causes redelivery. Both replicas must tolerate duplicate event IDs on restart.
type ReplicatorConfig struct {
        Source      EventStore
        Destination EventStore
        Checkpoints SubscriptionStore
        ID          string
        Generation  string
}

// Replicator runs Mirror's ordered copy loop and exposes acknowledged prefix
// barriers. Construct with NewReplicator. Run is single-use; after it ends,
// construct another instance with the same config to resume the checkpoint.
// It owns no store lifetimes and is independent of EventBus.Shutdown.
type Replicator struct {
        config       ReplicatorConfig
        comparer     EventStoreOffsetComparer
        mirrorConfig *mirrorConfig
        checkpointID string
        mu           sync.Mutex
        changed      chan struct{}
        wake         chan struct{}
        started      bool
        stopped      bool
        runErr       error
        confirmed    Offset
        ready        bool
}

// NewReplicator prepares a replicator without I/O or background goroutines.
// Mirror options configure retry timing, deduplication and observers. Automatic
// rewind reset is rejected: reusing an acknowledged generation after history
// replacement would invalidate its barriers. Ordinary Mirror is unchanged.
func NewReplicator(config ReplicatorConfig, opts ...MirrorOption) (*Replicator, error) {
        if config.Source == nil || config.Destination == nil || config.Checkpoints == nil {
                return nil, fmt.Errorf("eventbus: replication requires source, destination and checkpoint stores")
        }
        if config.ID == "" || config.Generation == "" {
                return nil, fmt.Errorf("eventbus: replication ID and generation are required")
        }
        if !utf8.ValidString(config.ID) || !utf8.ValidString(config.Generation) {
                return nil, fmt.Errorf("eventbus: replication ID and generation must be valid UTF-8")
        }
        comparer, ok := config.Source.(EventStoreOffsetComparer)
        if !ok {
                return nil, fmt.Errorf("eventbus: replication source must implement EventStoreOffsetComparer")
        }
        cfg := &mirrorConfig{pollInterval: 200 * time.Millisecond, dedupWindow: 1024}
        for _, opt := range opts {
                if opt == nil {
                        return nil, fmt.Errorf("eventbus: mirror option cannot be nil")
                }
                if err := opt(cfg); err != nil {
                        return nil, err
                }
        }
        if cfg.resetOnRewind {
                return nil, fmt.Errorf("eventbus: confirmed replication cannot reset history; use a new generation")
        }
        // A structured tuple prevents IDs/generations containing separators from
        // aliasing another relationship's checkpoints. Plain Mirror IDs stay separate.
        key, _ := json.Marshal([2]string{config.ID, config.Generation})
        r := &Replicator{config: config, comparer: comparer, mirrorConfig: cfg,
                checkpointID: "ebu:replication:" + string(key), changed: make(chan struct{}), wake: make(chan struct{}, 1)}
        cfg.progress = r
        return r, nil
}

// Position scopes an offset returned by Source.Append to this relationship.
// It delegates token comparisons to the store; this is not an existence or
// provenance check. Pass only offsets actually issued by this source generation.
// OffsetNewest is rejected; use Capture to resolve it.
func (r *Replicator) Position(offset Offset) (ReplicationPosition, error) {
        if offset == OffsetNewest {
                return ReplicationPosition{}, fmt.Errorf("%w: newest is symbolic", ErrReplicationPosition)
        }
        cmp, err := r.comparer.CompareOffsets(offset, OffsetOldest)
        if err != nil {
                return ReplicationPosition{}, fmt.Errorf("%w: %w", ErrReplicationPosition, err)
        }
        if cmp < 0 {
                return ReplicationPosition{}, fmt.Errorf("%w: offset precedes the start", ErrReplicationPosition)
        }
        return ReplicationPosition{ID: r.config.ID, Generation: r.config.Generation, Offset: offset}, nil
}

// Capture resolves the current source tail once. Call it after successful local
// writes, then Wait for the returned position to establish a prefix barrier.
// Concurrent later appends do not extend that barrier. This method does not
// itself wait for replication or start Run.
func (r *Replicator) Capture(ctx context.Context) (ReplicationPosition, error) {
        if err := ctx.Err(); err != nil {
                return ReplicationPosition{}, err
        }
        events, offset, err := r.config.Source.Read(ctx, OffsetNewest, 0)
        if err != nil {
                return ReplicationPosition{}, fmt.Errorf("replication: capture source tail: %w", err)
        }
        if err := ctx.Err(); err != nil {
                return ReplicationPosition{}, err
        }
        if len(events) != 0 {
                return ReplicationPosition{}, fmt.Errorf("replication: tail lookup returned historical events")
        }
        return r.Position(offset)
}

// Run copies until canceled or a fatal startup/checkpoint error. Transient
// mirror errors are retried in place. Run uses batched Read so cursor-only
// advances (empty chunks) are visible; Wait wakes idle polling without bypassing
// retry backoff. Concurrent/repeated Run calls fail with
// ErrReplicatorStarted; use a new instance to resume after this call returns.
// Observers run on this goroutine: never call Wait on this replicator from one.
func (r *Replicator) Run(ctx context.Context) (err error) {
        r.mu.Lock()
        if r.started {
                r.mu.Unlock()
                return ErrReplicatorStarted
        }
        r.started = true
        r.mu.Unlock()
        defer func() {
                r.mu.Lock()
                r.stopped = true
                r.runErr = errors.Join(ErrReplicationStopped, err)
                close(r.changed)
                r.mu.Unlock()
        }()
        return mirrorRun(ctx, r.config.Source, r.config.Destination, r.checkpointID, r.config.Checkpoints, r.mirrorConfig)
}

// Confirmed returns the latest confirmed SOURCE prefix and whether startup has
// established one. The empty prefix is valid. A previously confirmed prefix
// remains confirmed after Run stops, subject to the configured storage contract.
func (r *Replicator) Confirmed() (ReplicationPosition, bool) {
        r.mu.Lock()
        defer r.mu.Unlock()
        return ReplicationPosition{ID: r.config.ID, Generation: r.config.Generation, Offset: r.confirmed}, r.ready
}

// Wait waits until every source event through target has been acknowledged by
// Destination.Append and its SOURCE checkpoint has been saved. It never treats
// the local append, a read attempt, or an observer callback as acknowledgement.
// This is a replication barrier, not evidence that a destination projection has
// applied the events. Configure destination durability separately (see config).
//
// Wait may start before Run, and multiple waiters are independent. Canceling a
// wait neither cancels replication nor rolls back any writes: a timeout has an
// ambiguous outcome, so retry the same position instead of reissuing the write.
// Positions from another ID/generation fail immediately. If Run ends before
// target is confirmed, the error matches ErrReplicationStopped and Run's error.
func (r *Replicator) Wait(ctx context.Context, target ReplicationPosition) error {
        if target.ID != r.config.ID || target.Generation != r.config.Generation {
                return fmt.Errorf("%w: ID or generation mismatch", ErrReplicationPosition)
        }
        if _, err := r.Position(target.Offset); err != nil {
                return err
        }
        for {
                if err := ctx.Err(); err != nil {
                        return err
                }
                r.mu.Lock()
                ready, confirmed, stopped, runErr, changed := r.ready, r.confirmed, r.stopped, r.runErr, r.changed
                r.mu.Unlock()
                if ready {
                        cmp, err := r.comparer.CompareOffsets(confirmed, target.Offset)
                        if err != nil {
                                return fmt.Errorf("replication: compare wait target: %w", err)
                        }
                        if cmp >= 0 {
                                return nil
                        }
                }
                if stopped {
                        return runErr
                }
                select {
                case r.wake <- struct{}{}:
                default:
                }
                select {
                case <-changed:
                case <-ctx.Done():
                        return ctx.Err()
                }
        }
}

func (r *Replicator) initialize(ctx context.Context, from Offset) error {
        point, err := r.Position(from)
        if err != nil {
                return fmt.Errorf("replication: invalid saved checkpoint: %w", err)
        }
        tail, err := r.Capture(ctx)
        if err != nil {
                return err
        }
        cmp, err := r.comparer.CompareOffsets(point.Offset, tail.Offset)
        if err != nil {
                return fmt.Errorf("replication: compare saved checkpoint: %w", err)
        }
        if cmp > 0 {
                return fmt.Errorf("%w: checkpoint %q is ahead of tail %q", ErrReplicationHistory, from, tail.Offset)
        }
        r.confirm(from)
        return nil
}

func (r *Replicator) confirm(offset Offset) {
        r.mu.Lock()
        r.confirmed = offset
        r.ready = true
        close(r.changed)
        r.changed = make(chan struct{})
        r.mu.Unlock()
}

package state

import (
        "encoding/json"
        "fmt"
        "time"
)

// Insert creates an insert change message for an entity.
// The type parameter T determines the entity type name (unless overridden with WithEntityType).
//
// Example:
//
//        msg, err := state.Insert("user:1", User{Name: "Alice"})
func Insert[T any](key string, value T, opts ...ChangeOption) (*ChangeMessage, error) {
        return newChangeMessage[T](OperationInsert, key, &value, nil, opts...)
}

// Update creates an update change message for an entity.
// The type parameter T determines the entity type name (unless overridden with WithEntityType).
//
// Example:
//
//        msg, err := state.Update("user:1", User{Name: "Alice Smith"})
func Update[T any](key string, value T, opts ...ChangeOption) (*ChangeMessage, error) {
        return newChangeMessage[T](OperationUpdate, key, &value, nil, opts...)
}

// UpdateWithOldValue creates an update change message with the old value for conflict detection.
// The old value can be used by consumers to detect concurrent modifications.
//
// Example:
//
//        msg, err := state.UpdateWithOldValue("user:1", newUser, oldUser)
func UpdateWithOldValue[T any](key string, value, oldValue T, opts ...ChangeOption) (*ChangeMessage, error) {
        return newChangeMessage[T](OperationUpdate, key, &value, &oldValue, opts...)
}

// Delete creates a delete change message for an entity.
// The type parameter T determines the entity type name.
//
// Example:
//
//        msg, err := state.Delete[User]("user:1")
func Delete[T any](key string, opts ...ChangeOption) (*ChangeMessage, error) {
        return newChangeMessage[T](OperationDelete, key, nil, nil, opts...)
}

// DeleteWithOldValue creates a delete change message with the old value preserved.
// This is useful for consumers that need to know what was deleted.
//
// Example:
//
//        msg, err := state.DeleteWithOldValue("user:1", user)
func DeleteWithOldValue[T any](key string, oldValue T, opts ...ChangeOption) (*ChangeMessage, error) {
        return newChangeMessage[T](OperationDelete, key, nil, &oldValue, opts...)
}

// newChangeMessage is the internal constructor for change messages.
func newChangeMessage[T any](op Operation, key string, value, oldValue *T, opts ...ChangeOption) (*ChangeMessage, error) {
        if key == "" {
                return nil, fmt.Errorf("state: key cannot be empty")
        }

        cfg := &changeConfig{}
        for _, opt := range opts {
                opt(cfg)
        }

        // Determine entity type. entityTypeFor derives the name from the type
        // itself, so pointer entity types with a value-receiver StateTypeName
        // don't panic on a nil zero value.
        entityType := entityTypeFor[T]()
        if cfg.entityType != "" {
                entityType = cfg.entityType
        }

        msg := &ChangeMessage{
                Type: entityType,
                Key:  key,
                Headers: Headers{
                        Operation: op,
                        TxID:      cfg.txID,
                },
        }

        // Set timestamp
        if cfg.timestamp != nil {
                msg.Headers.Timestamp = cfg.timestamp.Format(time.RFC3339Nano)
        } else if cfg.autoTimestamp {
                msg.Headers.Timestamp = time.Now().UTC().Format(time.RFC3339Nano)
        }

        // Marshal value for insert/update
        if value != nil {
                valueData, err := json.Marshal(value)
                if err != nil {
                        return nil, fmt.Errorf("state: marshal value: %w", err)
                }
                msg.Value = valueData
        }

        // Marshal old value if provided
        if oldValue != nil {
                oldData, err := json.Marshal(oldValue)
                if err != nil {
                        return nil, fmt.Errorf("state: marshal old_value: %w", err)
                }
                msg.OldValue = oldData
        }

        return msg, nil
}

// SnapshotStart creates a snapshot-start control message.
// This marks the beginning of a snapshot in the stream.
func SnapshotStart(offset string) *ControlMessage {
        return &ControlMessage{
                Headers: ControlHeaders{
                        Control: ControlSnapshotStart,
                        Offset:  offset,
                },
        }
}

// SnapshotEnd creates a snapshot-end control message.
// This marks the end of a snapshot in the stream.
func SnapshotEnd(offset string) *ControlMessage {
        return &ControlMessage{
                Headers: ControlHeaders{
                        Control: ControlSnapshotEnd,
                        Offset:  offset,
                },
        }
}

// Reset creates a reset control message.
// This signals that all state should be cleared and rebuilt from subsequent events.
func Reset(offset string) *ControlMessage {
        return &ControlMessage{
                Headers: ControlHeaders{
                        Control: ControlReset,
                        Offset:  offset,
                },
        }
}

package state

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "net/url"
        "strings"
        "sync"

        eventbus "github.com/jilio/ebu"
)

// Store is a generic interface for state storage.
// Implementations can use any backing store (memory, database, etc.).
//
// Every method can report failure: a durable backend that swallowed errors
// would let the materializer advance LastOffset past updates that were never
// applied, silently corrupting snapshots and any log compaction based on
// them. The materializer never advances the offset when a store call fails.
type Store[T any] interface {
        // Get retrieves an entity by its composite key.
        Get(compositeKey string) (T, bool, error)
        // Set stores an entity with the given composite key.
        Set(compositeKey string, value T) error
        // Delete removes an entity by its composite key.
        Delete(compositeKey string) error
        // Clear removes all entities from the store.
        Clear() error
        // All returns a copy of all entities in the store.
        All() (map[string]T, error)
}

// MemoryStore is an in-memory implementation of Store.
// It is safe for concurrent access.
type MemoryStore[T any] struct {
        data map[string]T
        mu   sync.RWMutex
}

// NewMemoryStore creates a new in-memory store.
func NewMemoryStore[T any]() *MemoryStore[T] {
        return &MemoryStore[T]{
                data: make(map[string]T),
        }
}

// Get retrieves an entity by its composite key.
func (s *MemoryStore[T]) Get(key string) (T, bool, error) {
        s.mu.RLock()
        defer s.mu.RUnlock()
        v, ok := s.data[key]
        return v, ok, nil
}

// Set stores an entity with the given composite key.
func (s *MemoryStore[T]) Set(key string, value T) error {
        s.mu.Lock()
        defer s.mu.Unlock()
        s.data[key] = value
        return nil
}

// Delete removes an entity by its composite key.
func (s *MemoryStore[T]) Delete(key string) error {
        s.mu.Lock()
        defer s.mu.Unlock()
        delete(s.data, key)
        return nil
}

// Clear removes all entities from the store.
func (s *MemoryStore[T]) Clear() error {
        s.mu.Lock()
        defer s.mu.Unlock()
        s.data = make(map[string]T)
        return nil
}

// All returns a copy of all entities in the store.
func (s *MemoryStore[T]) All() (map[string]T, error) {
        s.mu.RLock()
        defer s.mu.RUnlock()
        result := make(map[string]T, len(s.data))
        for k, v := range s.data {
                result[k] = v
        }
        return result, nil
}

// TypedCollection provides type-safe access to a collection of entities.
// It wraps a Store and associates it with a specific entity type.
type TypedCollection[T any] struct {
        store      Store[T]
        entityType string
}

// NewTypedCollection creates a collection for a specific entity type.
// The entity type name is determined from the type parameter T.
func NewTypedCollection[T any](store Store[T]) *TypedCollection[T] {
        return &TypedCollection[T]{
                store:      store,
                entityType: entityTypeFor[T](),
        }
}

// NewTypedCollectionWithType creates a collection with an explicit type name.
// Use this when the type name should differ from the Go type name.
func NewTypedCollectionWithType[T any](store Store[T], entityType string) *TypedCollection[T] {
        return &TypedCollection[T]{
                store:      store,
                entityType: entityType,
        }
}

// EntityType returns the entity type name for this collection.
func (c *TypedCollection[T]) EntityType() string {
        return c.entityType
}

// Get retrieves an entity by its key (without the type prefix).
func (c *TypedCollection[T]) Get(key string) (T, bool, error) {
        return c.store.Get(CompositeKey(c.entityType, key))
}

// keyPrefix returns the composite-key prefix that scopes this collection's
// entities within its Store.
func (c *TypedCollection[T]) keyPrefix() string {
        return CompositeKey(c.entityType, "")
}

// All returns all entities in this collection.
// The keys in the returned map are CompositeKey-encoded keys (type/key for
// components without reserved characters).
//
// Only entities under this collection's type prefix are returned: several
// collections may share one Store, and each must see (and mutate — see
// clear/restore) only its own slice of it.
func (c *TypedCollection[T]) All() (map[string]T, error) {
        all, err := c.store.All()
        if err != nil {
                return nil, err
        }
        prefix := c.keyPrefix()
        result := make(map[string]T, len(all))
        for key, value := range all {
                if strings.HasPrefix(key, prefix) {
                        result[key] = value
                }
        }
        return result, nil
}

// ErrUndecodable marks materialization failures caused by a message payload
// that cannot be decoded (malformed JSON, an incompatible schema, or an
// unknown operation). It never marks store failures. WithApplyErrorPolicy
// consults this: only decode failures are safely skippable — a store failure
// is transient and must abort so the event is retried.
var ErrUndecodable = errors.New("state: undecodable message")

// collectionApplier is an internal interface for applying changes to collections.
type collectionApplier interface {
        applyChange(msg *ChangeMessage) error
        clear() error
        // snapshot serializes every entity in the collection, keyed by composite key.
        snapshot() (map[string]json.RawMessage, error)
        // restore clears the collection and repopulates it from serialized entities.
        restore(entities map[string]json.RawMessage) error
}

// typedCollectionApplier wraps TypedCollection to implement collectionApplier.
type typedCollectionApplier[T any] struct {
        collection *TypedCollection[T]
}

func (a *typedCollectionApplier[T]) applyChange(msg *ChangeMessage) error {
        key := CompositeKey(msg.Type, msg.Key)

        switch msg.Headers.Operation {
        case OperationInsert, OperationUpdate:
                var value T
                if err := json.Unmarshal(msg.Value, &value); err != nil {
                        return fmt.Errorf("%w: unmarshal value for %s/%s: %w", ErrUndecodable, msg.Type, msg.Key, err)
                }
                if err := a.collection.store.Set(key, value); err != nil {
                        return fmt.Errorf("state: set %s: %w", key, err)
                }

        case OperationDelete:
                if err := a.collection.store.Delete(key); err != nil {
                        return fmt.Errorf("state: delete %s: %w", key, err)
                }

        default:
                return fmt.Errorf("%w: unknown operation %q for %s/%s", ErrUndecodable, msg.Headers.Operation, msg.Type, msg.Key)
        }

        return nil
}

// clear removes only this collection's entities (its type-prefixed keys):
// a shared Store may hold other collections' data, which must survive.
func (a *typedCollectionApplier[T]) clear() error {
        all, err := a.collection.All()
        if err != nil {
                return fmt.Errorf("state: clear %s: %w", a.collection.entityType, err)
        }
        for key := range all {
                if err := a.collection.store.Delete(key); err != nil {
                        return fmt.Errorf("state: clear %s: %w", key, err)
                }
        }
        return nil
}

func (a *typedCollectionApplier[T]) snapshot() (map[string]json.RawMessage, error) {
        all, err := a.collection.All()
        if err != nil {
                return nil, fmt.Errorf("state: snapshot %s: %w", a.collection.entityType, err)
        }
        entities := make(map[string]json.RawMessage, len(all))
        for key, value := range all {
                data, err := json.Marshal(value)
                if err != nil {
                        return nil, fmt.Errorf("state: marshal snapshot entity %s: %w", key, err)
                }
                entities[key] = data
        }
        return entities, nil
}

func (a *typedCollectionApplier[T]) restore(entities map[string]json.RawMessage) error {
        if err := a.clear(); err != nil {
                return err
        }
        for key, raw := range entities {
                var value T
                if err := json.Unmarshal(raw, &value); err != nil {
                        return fmt.Errorf("state: unmarshal snapshot entity %s: %w", key, err)
                }
                if err := a.collection.store.Set(key, value); err != nil {
                        return fmt.Errorf("state: restore %s: %w", key, err)
                }
        }
        return nil
}

// Materializer processes state protocol messages and maintains state.
// It applies change messages to registered collections and handles control messages.
//
// Message application is serialized: concurrent Apply, ApplyChangeMessage, and
// ApplyControlMessage calls are applied one at a time, so a reset can never be
// overwritten by a logically-earlier change, and LastOffset only advances past
// events that were applied successfully. Callbacks configured via options run
// while an application is in flight and must not call back into Apply,
// ApplyChangeMessage, ApplyControlMessage, SaveSnapshotTo, or LoadSnapshotFrom.
type Materializer struct {
        collections map[string]collectionApplier
        cfg         *materializerConfig

        // applyMu serializes message application. It is held for the entirety of
        // Apply/ApplyChangeMessage/ApplyControlMessage — application plus the
        // lastOffset update happen as one atomic step — and for snapshot
        // restore, which needs a consistent view across collections.
        applyMu sync.Mutex

        // snapshotMu serializes SaveSnapshotTo calls end to end, including the
        // SaveSnapshot I/O that runs after applyMu is released: two racing
        // savers could otherwise overwrite a newer snapshot with an older one,
        // and a truncation based on the newer offset would then lose events.
        // Lock order: snapshotMu before applyMu, never the reverse.
        snapshotMu sync.Mutex

        // mu guards the collections map and lastOffset.
        mu         sync.RWMutex
        lastOffset eventbus.Offset
}

// NewMaterializer creates a new Materializer.
func NewMaterializer(opts ...MaterializerOption) *Materializer {
        cfg := &materializerConfig{}
        for _, opt := range opts {
                opt(cfg)
        }

        return &Materializer{
                collections: make(map[string]collectionApplier),
                cfg:         cfg,
        }
}

// RegisterCollection registers a typed collection with the materializer.
// The collection's entity type determines which change messages are routed to it.
func RegisterCollection[T any](m *Materializer, collection *TypedCollection[T]) {
        m.mu.Lock()
        defer m.mu.Unlock()
        m.collections[collection.EntityType()] = &typedCollectionApplier[T]{collection: collection}
}

// changeMessageEventType and controlMessageEventType are the names state
// protocol messages are persisted under when published through ebu (see the
// EventTypeName methods in message.go).
const (
        changeMessageEventType  = "state.ChangeMessage"
        controlMessageEventType = "state.ControlMessage"
)

// Apply processes a single StoredEvent containing a state protocol message.
// This is the primary integration point with ebu's Replay functionality.
//
// Routing is by the event's Type first: events published through ebu are
// stored as "state.ChangeMessage" / "state.ControlMessage" and are decoded
// strictly — a decode failure on a typed event is a real error. Events with
// any other Type fall back to structural detection for interoperability with
// streams written by other State Protocol implementations: an event whose
// data carries a "headers" object is applied if it decodes as a control or
// change message, and skipped as a foreign event otherwise. Events without a
// "headers" field are always skipped, so state messages can share a stream
// with regular events.
//
// The offset reported by LastOffset advances only when the event was applied
// (or skipped) successfully; failed events never advance it, so a resumed
// replay picks them up again — unless WithApplyErrorPolicy(ApplySkip) is
// configured, in which case undecodable typed messages are reported and
// skipped with the offset advancing. All failures are reported to the
// WithOnError callback before being returned.
func (m *Materializer) Apply(event *eventbus.StoredEvent) error {
        m.applyMu.Lock()
        defer m.applyMu.Unlock()

        err := m.applyEvent(event)
        if err != nil && m.cfg.applyErrorPolicy == ApplySkip && errors.Is(err, ErrUndecodable) {
                // Poison message: it was reported to WithOnError by applyEvent;
                // advancing past it keeps one bad payload from wedging every
                // future replay on the same event. Store failures never take this
                // path — they are transient and the event must be retried.
                m.setLastOffset(event.Offset)
                return nil
        }
        if err != nil {
                return err
        }

        m.setLastOffset(event.Offset)
        return nil
}

// applyEvent routes and applies one event. Caller must hold applyMu.
// Every returned error has been reported to WithOnError exactly once.
func (m *Materializer) applyEvent(event *eventbus.StoredEvent) error {
        switch event.Type {
        case changeMessageEventType:
                var changeMsg ChangeMessage
                if err := json.Unmarshal(event.Data, &changeMsg); err != nil {
                        return m.reportError(fmt.Errorf("%w: unmarshal change message: %w", ErrUndecodable, err))
                }
                return m.applyChange(&changeMsg)

        case controlMessageEventType:
                var ctrlMsg ControlMessage
                if err := json.Unmarshal(event.Data, &ctrlMsg); err != nil {
                        return m.reportError(fmt.Errorf("%w: unmarshal control message: %w", ErrUndecodable, err))
                }
                return m.applyControl(&ctrlMsg)

        default:
                return m.applyUntyped(event)
        }
}

// applyUntyped structurally detects state messages persisted under other
// type names (streams written by non-ebu State Protocol implementations).
// Anything that does not positively identify as a state message is skipped
// as a foreign event — never an error: on a mixed stream a payload that
// merely resembles a state message (e.g. any event with a "headers" field)
// must not wedge or corrupt the materializer.
func (m *Materializer) applyUntyped(event *eventbus.StoredEvent) error {
        var raw struct {
                Headers json.RawMessage `json:"headers"`
        }
        if err := json.Unmarshal(event.Data, &raw); err != nil {
                // Not JSON at all: a foreign event.
                return nil
        }
        if raw.Headers == nil {
                return nil
        }

        // A headers object with a known control value is a control message.
        // Unknown control values are NOT applied here (unlike the typed route):
        // a foreign event with an unrelated "control" field must not be consumed.
        var ctrlHeaders ControlHeaders
        if json.Unmarshal(raw.Headers, &ctrlHeaders) == nil && knownControl(ctrlHeaders.Control) {
                return m.applyControl(&ControlMessage{Headers: ctrlHeaders})
        }

        // A payload that decodes as a change message with a complete
        // type/key/operation triple is applied; anything else is foreign.
        var changeMsg ChangeMessage
        if err := json.Unmarshal(event.Data, &changeMsg); err != nil {
                return nil
        }
        if changeMsg.Type == "" || changeMsg.Key == "" || changeMsg.Headers.Operation == "" {
                return nil
        }
        return m.applyChange(&changeMsg)
}

// knownControl reports whether c is a control value this package understands.
func knownControl(c Control) bool {
        switch c {
        case ControlReset, ControlSnapshotStart, ControlSnapshotEnd:
                return true
        }
        return false
}

// ApplyChangeMessage processes a ChangeMessage directly.
// Use this when you have a ChangeMessage that's not wrapped in a StoredEvent.
func (m *Materializer) ApplyChangeMessage(msg *ChangeMessage) error {
        m.applyMu.Lock()
        defer m.applyMu.Unlock()
        return m.applyChange(msg)
}

// ApplyControlMessage processes a ControlMessage directly.
// Use this when you have a ControlMessage that's not wrapped in a StoredEvent.
// It returns an error when clearing a collection fails on a reset, or — in
// strict mode — when the control value is unknown.
func (m *Materializer) ApplyControlMessage(msg *ControlMessage) error {
        m.applyMu.Lock()
        defer m.applyMu.Unlock()
        return m.applyControl(msg)
}

// applyChange applies a change message to the appropriate collection.
func (m *Materializer) applyChange(msg *ChangeMessage) error {
        m.mu.RLock()
        collection, ok := m.collections[msg.Type]
        m.mu.RUnlock()

        if !ok {
                if m.cfg.strictSchema {
                        return m.reportError(fmt.Errorf("state: unknown entity type: %s", msg.Type))
                }
                return nil // Ignore unknown types in non-strict mode
        }

        if err := collection.applyChange(msg); err != nil {
                return m.reportError(err)
        }

        return nil
}

// reportError invokes the configured error callback, if any, and returns err.
// Every materialization error passes through here exactly once.
func (m *Materializer) reportError(err error) error {
        if m.cfg.onError != nil {
                m.cfg.onError(err)
        }
        return err
}

// setLastOffset records the offset of the last successfully applied event.
// Callers must hold applyMu.
func (m *Materializer) setLastOffset(offset eventbus.Offset) {
        m.mu.Lock()
        m.lastOffset = offset
        m.mu.Unlock()
}

// applyControl applies a control message.
func (m *Materializer) applyControl(msg *ControlMessage) error {
        switch msg.Headers.Control {
        case ControlReset:
                m.mu.Lock()
                for _, c := range m.collections {
                        if err := c.clear(); err != nil {
                                m.mu.Unlock()
                                return m.reportError(fmt.Errorf("state: reset: %w", err))
                        }
                }
                m.mu.Unlock()
                if m.cfg.onReset != nil {
                        m.cfg.onReset()
                }

        case ControlSnapshotStart:
                if m.cfg.onSnapshot != nil {
                        m.cfg.onSnapshot(true)
                }

        case ControlSnapshotEnd:
                if m.cfg.onSnapshot != nil {
                        m.cfg.onSnapshot(false)
                }

        default:
                // A control this package does not understand: silently dropping it
                // in strict mode could mean missing a reset the stream demanded.
                // Mirrors unknown-entity-type handling: strict errors, lax ignores.
                if m.cfg.strictSchema {
                        return m.reportError(fmt.Errorf("%w: unknown control %q", ErrUndecodable, msg.Headers.Control))
                }
        }
        return nil
}

// LastOffset returns the offset of the last applied event.
func (m *Materializer) LastOffset() eventbus.Offset {
        m.mu.RLock()
        defer m.mu.RUnlock()
        return m.lastOffset
}

// Replay is a convenience method that replays events from an EventBus.
// It calls bus.ReplayWithUpcast and applies each event through the
// materializer, so events with registered schema migrations are upcasted
// before application.
func (m *Materializer) Replay(ctx context.Context, bus *eventbus.EventBus, from eventbus.Offset) error {
        return bus.ReplayWithUpcast(ctx, from, m.Apply)
}

const (
        materializerSnapshotVersion = 1
        snapshotKeyCodecPercentV1   = "percent-v1"
)

// materializerSnapshot versions both the overall snapshot shape and the key
// codec. Composite keys are persistent data: changing their encoding without
// a marker would let a newer materializer accept an older snapshot, restore
// unreachable keys, and then skip the compacted history covered by its offset.
type materializerSnapshot struct {
        Version     int                                   `json:"version"`
        KeyCodec    string                                `json:"key_codec"`
        Collections map[string]map[string]json.RawMessage `json:"collections"`
}

// SaveSnapshotTo serializes every registered collection and saves the result
// to s under snapshotID, tagged with the offset of the last applied event.
// The blob format is
// {"version":1,"key_codec":"percent-v1","collections":{"entityType":{"compositeKey":<entity JSON>}}}.
//
// It returns an error if no event has been applied yet (LastOffset is empty):
// such a snapshot would claim OffsetOldest, and a later TruncateBefore based
// on it could silently discard the whole log.
//
// The intended compaction sequence is:
//
//        if err := mat.SaveSnapshotTo(ctx, snapshotter, "users"); err != nil { ... }
//        // Optionally, once the snapshot is durably saved, compact the log:
//        if tr, ok := bus.GetStore().(eventbus.EventStoreTruncator); ok {
//            tr.TruncateBefore(ctx, offset) // offset the snapshot was saved at
//        }
//
// Truncation is only safe once the snapshot is durably saved AND no other
// reader or subscription still needs the truncated prefix. The offset to
// truncate at is the snapshot's offset (retrievable via the snapshotter's
// LoadSnapshot, or LastOffset when no events are applied concurrently) —
// never a later one, or events not covered by the snapshot would be lost.
//
// Message application is paused only while the collections are captured and
// serialized; the SaveSnapshot I/O itself runs without blocking Apply.
// Concurrent SaveSnapshotTo calls are serialized with each other so a slower
// older snapshot can never overwrite a newer one.
func (m *Materializer) SaveSnapshotTo(ctx context.Context, s eventbus.EventStoreSnapshotter, snapshotID string) error {
        m.snapshotMu.Lock()
        defer m.snapshotMu.Unlock()

        offset, encoded, err := m.captureSnapshot(snapshotID)
        if err != nil {
                return err
        }

        if err := s.SaveSnapshot(ctx, snapshotID, offset, encoded); err != nil {
                return fmt.Errorf("state: save snapshot %q: %w", snapshotID, err)
        }
        return nil
}

// captureSnapshot serializes all collections as one consistent view at the
// current LastOffset, holding applyMu so no application runs mid-capture.
func (m *Materializer) captureSnapshot(snapshotID string) (eventbus.Offset, json.RawMessage, error) {
        m.applyMu.Lock()
        defer m.applyMu.Unlock()

        m.mu.RLock()
        offset := m.lastOffset
        collections := make(map[string]collectionApplier, len(m.collections))
        for entityType, c := range m.collections {
                collections[entityType] = c
        }
        m.mu.RUnlock()

        if offset == eventbus.OffsetOldest {
                return "", nil, fmt.Errorf("state: refusing to save snapshot %q: no events applied yet (snapshot would claim OffsetOldest)", snapshotID)
        }

        snapshot := materializerSnapshot{
                Version:     materializerSnapshotVersion,
                KeyCodec:    snapshotKeyCodecPercentV1,
                Collections: make(map[string]map[string]json.RawMessage, len(collections)),
        }
        for entityType, c := range collections {
                entities, err := c.snapshot()
                if err != nil {
                        return "", nil, err
                }
                snapshot.Collections[entityType] = entities
        }
        if err := validatePercentV1SnapshotKeys(snapshot.Collections); err != nil {
                return "", nil, fmt.Errorf("state: refusing to save snapshot %q: %w", snapshotID, err)
        }

        // snapshot contains only string keys and json.RawMessage values produced by
        // json.Marshal, so encoding cannot fail.
        encoded, _ := json.Marshal(snapshot)
        return offset, encoded, nil
}

// decodeSnapshotCollections validates the snapshot/key-codec marker before
// returning any data to the mutating restore path. Versionless snapshots are
// the legacy type->composite-key map. They remain safe only when every entity
// type and key excludes the reserved '%' and '/' characters, because those
// composite keys are byte-for-byte identical under both codecs.
func decodeSnapshotCollections(blob json.RawMessage) (map[string]map[string]json.RawMessage, error) {
        var topLevel map[string]json.RawMessage
        if err := json.Unmarshal(blob, &topLevel); err != nil {
                return nil, fmt.Errorf("unmarshal snapshot envelope: %w", err)
        }
        if topLevel == nil {
                return nil, fmt.Errorf("snapshot envelope must be a JSON object")
        }

        // A current snapshot has a scalar version. A legacy collection may itself
        // be named "version"; its value is an object, so keep treating that shape as
        // legacy rather than reserving a previously-valid entity type name.
        if rawVersion, ok := topLevel["version"]; ok {
                var legacyVersionCollection map[string]json.RawMessage
                if err := json.Unmarshal(rawVersion, &legacyVersionCollection); err != nil {
                        var snapshot materializerSnapshot
                        if err := json.Unmarshal(blob, &snapshot); err != nil {
                                return nil, fmt.Errorf("unmarshal versioned snapshot: %w", err)
                        }
                        if snapshot.Version != materializerSnapshotVersion ||
                                snapshot.KeyCodec != snapshotKeyCodecPercentV1 || snapshot.Collections == nil {
                                return nil, fmt.Errorf("unsupported snapshot format version=%d key_codec=%q", snapshot.Version, snapshot.KeyCodec)
                        }
                        if err := validatePercentV1SnapshotKeys(snapshot.Collections); err != nil {
                                return nil, err
                        }
                        return snapshot.Collections, nil
                }
        }

        var legacy map[string]map[string]json.RawMessage
        if err := json.Unmarshal(blob, &legacy); err != nil {
                return nil, fmt.Errorf("unmarshal legacy snapshot: %w", err)
        }
        if err := validateLegacySnapshotKeys(legacy); err != nil {
                return nil, err
        }
        return legacy, nil
}

func validatePercentV1SnapshotKeys(collections map[string]map[string]json.RawMessage) error {
        for entityType, entities := range collections {
                if entities == nil {
                        return fmt.Errorf("versioned snapshot collection %q must be a JSON object, not null", entityType)
                }
                prefix := CompositeKey(entityType, "")
                for compositeKey := range entities {
                        if !strings.HasPrefix(compositeKey, prefix) || strings.Count(compositeKey, "/") != 1 {
                                return fmt.Errorf("versioned snapshot composite key %q does not match collection %q under key codec %q", compositeKey, entityType, snapshotKeyCodecPercentV1)
                        }

                        encodedKey := strings.TrimPrefix(compositeKey, prefix)
                        decodedKey, err := url.PathUnescape(encodedKey)
                        if err != nil || encodeKeyComponent(decodedKey) != encodedKey {
                                return fmt.Errorf("versioned snapshot composite key %q is not canonical under key codec %q", compositeKey, snapshotKeyCodecPercentV1)
                        }
                }
        }
        return nil
}

func validateLegacySnapshotKeys(collections map[string]map[string]json.RawMessage) error {
        for entityType, entities := range collections {
                if entities == nil {
                        return fmt.Errorf("legacy snapshot collection %q must be a JSON object, not null", entityType)
                }
                if strings.ContainsAny(entityType, "%/") {
                        return fmt.Errorf("legacy snapshot entity type %q uses a reserved key-codec character; migrate it explicitly, or discard and rebuild from OffsetOldest only if the complete source history is still available", entityType)
                }
                prefix := entityType + "/"
                for compositeKey := range entities {
                        key, ok := strings.CutPrefix(compositeKey, prefix)
                        if !ok || strings.ContainsAny(key, "%/") {
                                return fmt.Errorf("legacy snapshot composite key %q in collection %q is unsafe under key codec %q; migrate it explicitly, or discard and rebuild from OffsetOldest only if the complete source history is still available", compositeKey, entityType, snapshotKeyCodecPercentV1)
                        }
                }
        }
        return nil
}

// LoadSnapshotFrom restores the materializer from the snapshot saved under
// snapshotID: it clears all registered collections, repopulates them from the
// snapshot blob, sets LastOffset to the snapshot's offset, and returns that
// offset. The caller resumes with:
//
//        offset, err := mat.LoadSnapshotFrom(ctx, snapshotter, "users")
//        if err != nil { ... }
//        if err := mat.Replay(ctx, bus, offset); err != nil { ... }
//
// When no snapshot exists it returns OffsetOldest with the collections
// untouched, so the caller's Replay naturally rebuilds from the beginning.
// Snapshot data for entity types with no registered collection is dropped
// silently. A snapshot that omits any currently registered collection is
// rejected before mutation: accepting its offset would skip the omitted
// projection's earlier history, permanently so if that history was compacted.
// Register every collection before loading, and explicitly migrate an older
// snapshot when adding a collection. Versionless snapshots from the legacy
// composite-key codec are accepted only when all entity types and keys contain
// neither '%' nor '/'; otherwise loading fails before any collection or
// LastOffset is changed. Versioned snapshots are likewise rejected before
// mutation unless every composite key is canonical for its declared key codec
// and collection.
//
// If validation succeeds but mutating restore then fails, the materializer is
// left empty with LastOffset reset to OffsetOldest, so a full replay from
// OffsetOldest rebuilds the state.
func (m *Materializer) LoadSnapshotFrom(ctx context.Context, s eventbus.EventStoreSnapshotter, snapshotID string) (eventbus.Offset, error) {
        m.applyMu.Lock()
        defer m.applyMu.Unlock()

        offset, blob, err := s.LoadSnapshot(ctx, snapshotID)
        if err != nil {
                return eventbus.OffsetOldest, fmt.Errorf("state: load snapshot %q: %w", snapshotID, err)
        }
        if offset == eventbus.OffsetOldest && blob == nil {
                return eventbus.OffsetOldest, nil // No snapshot: replay from the beginning.
        }

        decoded, err := decodeSnapshotCollections(blob)
        if err != nil {
                return eventbus.OffsetOldest, fmt.Errorf("state: decode snapshot %q: %w", snapshotID, err)
        }

        m.mu.Lock()
        defer m.mu.Unlock()
        for entityType := range m.collections {
                if _, ok := decoded[entityType]; !ok {
                        return eventbus.OffsetOldest, fmt.Errorf("state: snapshot %q is missing registered collection %q; rebuild from OffsetOldest only if the complete source history is available, or migrate the snapshot before loading if history was compacted", snapshotID, entityType)
                }
        }

        // clearAll empties every registered collection; on restore failure it
        // leaves the materializer empty rather than partially restored, so a
        // full replay from OffsetOldest rebuilds the state.
        clearAll := func() error {
                for _, c := range m.collections {
                        if err := c.clear(); err != nil {
                                return err
                        }
                }
                return nil
        }

        resetToEmpty := func(cause error) (eventbus.Offset, error) {
                if clearErr := clearAll(); clearErr != nil {
                        cause = errors.Join(cause, clearErr)
                }
                m.lastOffset = eventbus.OffsetOldest
                return eventbus.OffsetOldest, cause
        }

        if err := clearAll(); err != nil {
                return resetToEmpty(err)
        }
        for entityType, entities := range decoded {
                c, ok := m.collections[entityType]
                if !ok {
                        continue // Entity type no longer registered: drop its snapshot data.
                }
                if err := c.restore(entities); err != nil {
                        return resetToEmpty(err)
                }
        }

        m.lastOffset = offset
        return offset, nil
}

package state

import (
        "encoding/json"
        "reflect"
        "strings"
)

// Operation represents the type of change operation per the State Protocol.
type Operation string

const (
        // OperationInsert indicates a new entity is being created.
        OperationInsert Operation = "insert"
        // OperationUpdate indicates an existing entity is being modified.
        OperationUpdate Operation = "update"
        // OperationDelete indicates an entity is being removed.
        OperationDelete Operation = "delete"
)

// Control represents the type of control message per the State Protocol.
type Control string

const (
        // ControlSnapshotStart marks the beginning of a snapshot.
        ControlSnapshotStart Control = "snapshot-start"
        // ControlSnapshotEnd marks the end of a snapshot.
        ControlSnapshotEnd Control = "snapshot-end"
        // ControlReset signals that all state should be cleared.
        ControlReset Control = "reset"
)

// Headers contains metadata for change messages per the State Protocol.
type Headers struct {
        // Operation is the type of change (insert, update, delete).
        Operation Operation `json:"operation"`
        // TxID is an optional transaction identifier for grouping related changes.
        TxID string `json:"txid,omitempty"`
        // Timestamp is an optional RFC 3339 formatted timestamp.
        Timestamp string `json:"timestamp,omitempty"`
}

// ControlHeaders contains metadata for control messages.
type ControlHeaders struct {
        // Control is the type of control message.
        Control Control `json:"control"`
        // Offset is an optional reference to a stream position.
        Offset string `json:"offset,omitempty"`
}

// ChangeMessage represents a state change event per the State Protocol.
// It contains an entity mutation (insert, update, or delete) with a composite key.
type ChangeMessage struct {
        // Type is the entity type discriminator (e.g., "user", "order").
        Type string `json:"type"`
        // Key is the unique identifier within the entity type.
        Key string `json:"key"`
        // Value contains the entity data (required for insert/update).
        Value json.RawMessage `json:"value,omitempty"`
        // OldValue contains the previous entity data (optional, for conflict detection).
        OldValue json.RawMessage `json:"old_value,omitempty"`
        // Headers contains operation metadata.
        Headers Headers `json:"headers"`
}

// ControlMessage represents a control event per the State Protocol.
// Control messages manage stream lifecycle (snapshots, resets).
type ControlMessage struct {
        // Headers contains control message metadata.
        Headers ControlHeaders `json:"headers"`
}

// TypeNamer is an optional interface that entity types can implement to provide
// their own type name. This mirrors ebu's TypeNamer pattern.
//
// StateTypeName must be a pure function of the type (never of instance
// state): the package derives names from zero values and fresh instances.
//
// Example:
//
//        type User struct { ... }
//        func (u User) StateTypeName() string { return "user" }
type TypeNamer interface {
        StateTypeName() string
}

// typeNamerType is the reflect.Type of the TypeNamer interface.
var typeNamerType = reflect.TypeOf((*TypeNamer)(nil)).Elem()

// entityTypeOf returns the entity type name for a reflect.Type, honoring
// TypeNamer without ever invoking it on a nil pointer: for pointer types a
// fresh instance is allocated, since a value-receiver method promoted to the
// pointer type would dereference nil (mirrors typeNameOf in the parent
// package).
func entityTypeOf(t reflect.Type) string {
        if t.Implements(typeNamerType) {
                if t.Kind() == reflect.Pointer {
                        return reflect.New(t.Elem()).Interface().(TypeNamer).StateTypeName()
                }
                return reflect.Zero(t).Interface().(TypeNamer).StateTypeName()
        }
        if reflect.PointerTo(t).Implements(typeNamerType) {
                return reflect.New(t).Interface().(TypeNamer).StateTypeName()
        }
        return t.String()
}

// entityTypeFor returns the entity type name for a type parameter without
// needing an instance, so pointer entity types are safe.
func entityTypeFor[T any]() string {
        return entityTypeOf(reflect.TypeOf((*T)(nil)).Elem())
}

// EntityType returns the type name for an entity.
// If the entity implements TypeNamer, returns the custom name.
// Otherwise returns the reflect-based package-qualified name.
// Returns "nil" if entity is nil.
func EntityType(entity any) string {
        if entity == nil {
                return "nil"
        }
        if namer, ok := entity.(TypeNamer); ok {
                if v := reflect.ValueOf(entity); v.Kind() == reflect.Pointer && v.IsNil() {
                        // A typed nil pointer: calling a value-receiver StateTypeName
                        // through it would dereference nil. Derive the name from the
                        // type instead.
                        return entityTypeOf(v.Type())
                }
                return namer.StateTypeName()
        }
        return reflect.TypeOf(entity).String()
}

// CompositeKey returns an unambiguous composite key from type and key
// components. Percent signs and slashes within either component are
// percent-encoded, leaving the sole unescaped slash as the separator.
// Components without those reserved characters retain the familiar type/key
// representation.
func CompositeKey(entityType, key string) string {
        return encodeKeyComponent(entityType) + "/" + encodeKeyComponent(key)
}

func encodeKeyComponent(component string) string {
        component = strings.ReplaceAll(component, "%", "%25")
        return strings.ReplaceAll(component, "/", "%2F")
}

// EventTypeName implements ebu's TypeNamer interface for ChangeMessage.
// This allows ChangeMessage to be published directly to an EventBus.
func (m ChangeMessage) EventTypeName() string {
        return "state.ChangeMessage"
}

// EventTypeName implements ebu's TypeNamer interface for ControlMessage.
// This allows ControlMessage to be published directly to an EventBus.
func (m ControlMessage) EventTypeName() string {
        return "state.ControlMessage"
}

package state

import "time"

// ChangeOption configures a change message.
type ChangeOption func(*changeConfig)

type changeConfig struct {
        txID          string
        timestamp     *time.Time
        autoTimestamp bool
        entityType    string
}

// WithTxID sets the transaction ID for grouping related changes.
// Transaction IDs allow consumers to process related changes atomically.
func WithTxID(txID string) ChangeOption {
        return func(c *changeConfig) {
                c.txID = txID
        }
}

// WithTimestamp sets an explicit timestamp for the change message.
// The timestamp will be formatted as RFC 3339.
func WithTimestamp(t time.Time) ChangeOption {
        return func(c *changeConfig) {
                c.timestamp = &t
        }
}

// WithAutoTimestamp automatically sets the timestamp to the current time.
func WithAutoTimestamp() ChangeOption {
        return func(c *changeConfig) {
                c.autoTimestamp = true
        }
}

// WithEntityType overrides the automatic entity type name.
// Use this when the type name should differ from the Go type name.
func WithEntityType(typeName string) ChangeOption {
        return func(c *changeConfig) {
                c.entityType = typeName
        }
}

// MaterializerOption configures a Materializer.
type MaterializerOption func(*materializerConfig)

// ApplyErrorPolicy determines how the Materializer handles a state message
// that cannot be decoded (see ErrUndecodable). It mirrors ebu's
// ReplayErrorPolicy for durable subscriptions.
type ApplyErrorPolicy int

const (
        // ApplyAbort makes Apply return the decode error without advancing
        // LastOffset (default). The next replay hits the same event again. Use
        // this when an undecodable message means a bug that must be fixed
        // before proceeding.
        ApplyAbort ApplyErrorPolicy = iota

        // ApplySkip reports the decode error to WithOnError and advances
        // LastOffset past the poison message, so one bad payload cannot wedge
        // every future replay. Store failures are never skipped: they are
        // transient, and the event must be retried.
        ApplySkip
)

type materializerConfig struct {
        onReset          func()
        onSnapshot       func(start bool)
        onError          func(error)
        strictSchema     bool
        applyErrorPolicy ApplyErrorPolicy
}

// WithOnReset sets a callback invoked when a reset control message is received.
// The callback is called after all collections have been cleared.
func WithOnReset(fn func()) MaterializerOption {
        return func(c *materializerConfig) {
                c.onReset = fn
        }
}

// WithOnSnapshot sets a callback invoked on snapshot-start/end messages.
// The boolean parameter is true for snapshot-start, false for snapshot-end.
func WithOnSnapshot(fn func(start bool)) MaterializerOption {
        return func(c *materializerConfig) {
                c.onSnapshot = fn
        }
}

// WithOnError sets an error handler for materialization errors.
// It is invoked exactly once for every failed application: envelope or
// change-message decode failures, unknown entity types in strict mode,
// unknown operations, and collection apply errors.
func WithOnError(fn func(error)) MaterializerOption {
        return func(c *materializerConfig) {
                c.onError = fn
        }
}

// WithApplyErrorPolicy sets how the materializer treats undecodable state
// messages during Apply. The default is ApplyAbort.
func WithApplyErrorPolicy(policy ApplyErrorPolicy) MaterializerOption {
        return func(c *materializerConfig) {
                c.applyErrorPolicy = policy
        }
}

// WithStrictSchema enables strict schema validation.
// When enabled, the materializer returns an error for unknown entity types.
// When disabled (default), unknown types are silently ignored.
func WithStrictSchema() MaterializerOption {
        return func(c *materializerConfig) {
                c.strictSchema = true
        }
}

package eventbus

import (
        "fmt"
        "reflect"
        "sync"
)

// persistedTypeRegistry owns the per-bus mapping between durable wire names
// and Go types. Reservations make multi-step operations transactional: a
// subscription can validate its identity before replay without making that
// identity permanent until durable evidence has been decoded or setup
// succeeds.
type persistedTypeRegistry struct {
        mu     sync.Mutex
        active bool
        byName map[string]*persistedTypeClaim
        byType map[reflect.Type]*persistedTypeClaim

        // Before persistence is enabled, typed registrations are retained without
        // enforcing durable wire-name uniqueness. A purely in-process bus routes by
        // reflect.Type, so equal TypeNamer names are harmless there. If persistence
        // is enabled later, activate validates the complete set atomically before a
        // store becomes observable on the bus.
        deferred        []persistedTypeSpec
        deferredSet     map[persistedTypeSpec]struct{}
        deferredPending map[*persistedTypeReservation]struct{}
}

type persistedTypeClaim struct {
        name      string
        eventType reflect.Type
        pending   int
        sticky    bool
}

type persistedTypeSpec struct {
        name      string
        eventType reflect.Type
}

type persistedTypeReservation struct {
        registry *persistedTypeRegistry
        claims   []*persistedTypeClaim
        specs    []persistedTypeSpec
        once     sync.Once
}

func newPersistedTypeRegistry() *persistedTypeRegistry {
        return &persistedTypeRegistry{
                active:          true,
                byName:          make(map[string]*persistedTypeClaim),
                byType:          make(map[reflect.Type]*persistedTypeClaim),
                deferredSet:     make(map[persistedTypeSpec]struct{}),
                deferredPending: make(map[*persistedTypeReservation]struct{}),
        }
}

// newDeferredPersistedTypeRegistry creates the registry used by EventBus.
// New applies arbitrary Option functions before it knows whether the final
// configuration is persistent, so typed registrations remain provisional
// until all options have run.
func newDeferredPersistedTypeRegistry() *persistedTypeRegistry {
        registry := newPersistedTypeRegistry()
        registry.active = false
        return registry
}

// reserve transactionally records every requested name/type pair. An active
// registry validates the complete request before changing pending counts, so a
// conflicting multi-type operation never leaves partial ownership behind. A
// deferred registry postpones that same atomic validation until activation.
func (r *persistedTypeRegistry) reserve(specs ...persistedTypeSpec) (*persistedTypeReservation, error) {
        // Deduplicate identical endpoints (for example, validation of an invalid
        // self-upcast). Conflicts are checked immediately by an active registry and
        // deferred otherwise, because they matter only if persistence is enabled.
        unique := make([]persistedTypeSpec, 0, len(specs))
        seen := make(map[persistedTypeSpec]struct{}, len(specs))
        for _, spec := range specs {
                if _, ok := seen[spec]; ok {
                        continue
                }
                seen[spec] = struct{}{}
                unique = append(unique, spec)
        }

        r.mu.Lock()
        defer r.mu.Unlock()
        reservation := &persistedTypeReservation{registry: r, specs: unique}
        if !r.active {
                r.deferredPending[reservation] = struct{}{}
                return reservation, nil
        }
        if err := r.reserveActiveLocked(reservation); err != nil {
                return nil, err
        }
        return reservation, nil
}

// reserveActiveLocked validates and installs the pending claims for one
// reservation. The caller must hold r.mu. Validation finishes before mutation,
// so a multi-type request is atomic.
func (r *persistedTypeRegistry) reserveActiveLocked(reservation *persistedTypeReservation) error {
        requestNames := make(map[string]reflect.Type, len(reservation.specs))
        requestTypes := make(map[reflect.Type]string, len(reservation.specs))
        for _, spec := range reservation.specs {
                if existing, ok := requestNames[spec.name]; ok && existing != spec.eventType {
                        return persistedTypeNameConflict(spec.name, existing, spec.eventType)
                }
                if existing, ok := requestTypes[spec.eventType]; ok && existing != spec.name {
                        return persistedTypeIdentityConflict(spec.eventType, existing, spec.name)
                }
                requestNames[spec.name] = spec.eventType
                requestTypes[spec.eventType] = spec.name

                if claim, ok := r.byName[spec.name]; ok && claim.eventType != spec.eventType {
                        return persistedTypeNameConflict(spec.name, claim.eventType, spec.eventType)
                }
                if claim, ok := r.byType[spec.eventType]; ok && claim.name != spec.name {
                        return persistedTypeIdentityConflict(spec.eventType, claim.name, spec.name)
                }
        }

        reservation.claims = make([]*persistedTypeClaim, 0, len(reservation.specs))
        for _, spec := range reservation.specs {
                claim := r.byName[spec.name]
                if claim == nil {
                        claim = &persistedTypeClaim{name: spec.name, eventType: spec.eventType}
                        r.byName[spec.name] = claim
                        r.byType[spec.eventType] = claim
                }
                claim.pending++
                reservation.claims = append(reservation.claims, claim)
        }
        return nil
}

func persistedTypeNameConflict(name string, existing, requested reflect.Type) error {
        return fmt.Errorf("eventbus: persisted event type name %q is already registered for Go type %v; type %v must use a distinct TypeNamer.EventTypeName", name, existing, requested)
}

func persistedTypeIdentityConflict(eventType reflect.Type, existing, requested string) error {
        return fmt.Errorf("eventbus: Go event type %s is already associated with persisted event type name %q; EventTypeName must be a pure, immutable function of the type (got %q)", eventType, existing, requested)
}

// activate atomically audits every successful provisional registration and
// turns durable-name enforcement on. Pending reservations are included, so a
// concurrent registration cannot slip between the audit and activation.
func (r *persistedTypeRegistry) activate() error {
        r.mu.Lock()
        defer r.mu.Unlock()
        if r.active {
                return nil
        }

        all := make([]persistedTypeSpec, 0, len(r.deferred))
        all = append(all, r.deferred...)
        for reservation := range r.deferredPending {
                all = append(all, reservation.specs...)
        }
        validation := &persistedTypeReservation{registry: r, specs: all}
        if err := r.reserveActiveLocked(validation); err != nil {
                return err
        }

        // reserveActiveLocked represented the entire provisional set as pending.
        // Convert committed registrations to sticky claims, then replace the
        // aggregate pending counts with the original in-flight reservations.
        for _, claim := range validation.claims {
                claim.pending--
        }
        for _, spec := range r.deferred {
                claim := r.byName[spec.name]
                claim.sticky = true
        }
        for reservation := range r.deferredPending {
                reservation.claims = make([]*persistedTypeClaim, 0, len(reservation.specs))
                for _, spec := range reservation.specs {
                        claim := r.byName[spec.name]
                        claim.pending++
                        reservation.claims = append(reservation.claims, claim)
                }
        }

        r.deferred = nil
        r.deferredSet = nil
        r.deferredPending = nil
        r.active = true
        return nil
}

func (r *persistedTypeRegistry) isActive() bool {
        r.mu.Lock()
        defer r.mu.Unlock()
        return r.active
}

func (bus *EventBus) activatePersistenceTypes() {
        if err := bus.persistedTypes.activate(); err != nil {
                panic(fmt.Sprintf("eventbus: cannot enable persistence: %v", err))
        }
}

// Commit records a successful registration; on an active registry it makes
// every mapping sticky for the lifetime of the bus. Rollback releases only
// this operation's pending references. Both methods are terminal and
// idempotent, which keeps deferred rollback safe after an early proof-of-type
// commit.
func (r *persistedTypeReservation) Commit() {
        r.finish(true)
}

func (r *persistedTypeReservation) Rollback() {
        r.finish(false)
}

func (r *persistedTypeReservation) finish(commit bool) {
        if r == nil {
                return
        }
        r.once.Do(func() {
                r.registry.mu.Lock()
                defer r.registry.mu.Unlock()
                if !r.registry.active {
                        delete(r.registry.deferredPending, r)
                        if commit {
                                for _, spec := range r.specs {
                                        if _, exists := r.registry.deferredSet[spec]; exists {
                                                continue
                                        }
                                        r.registry.deferredSet[spec] = struct{}{}
                                        r.registry.deferred = append(r.registry.deferred, spec)
                                }
                        }
                        return
                }
                for _, claim := range r.claims {
                        claim.pending--
                        if commit {
                                claim.sticky = true
                        }
                        if claim.pending == 0 && !claim.sticky {
                                delete(r.registry.byName, claim.name)
                                delete(r.registry.byType, claim.eventType)
                        }
                }
        })
}

// reservePersistedTypes retains successful typed registrations provisionally
// on a purely in-process bus. They do not constrain reflect.Type routing, but
// are audited atomically if persistence is enabled later.
func (bus *EventBus) reservePersistedTypes(specs ...persistedTypeSpec) (*persistedTypeReservation, error) {
        return bus.persistedTypes.reserve(specs...)
}

func (bus *EventBus) reserveReplayID(subscriptionID string) error {
        if subscriptionID == "" {
                return fmt.Errorf("eventbus: subscription ID cannot be empty")
        }
        bus.replayMu.Lock()
        defer bus.replayMu.Unlock()
        if _, exists := bus.replayIDs[subscriptionID]; exists {
                return fmt.Errorf("eventbus: durable subscription ID %q was already registered or is actively owned on this bus; it cannot be reused while that ownership remains", subscriptionID)
        }
        bus.replayIDs[subscriptionID] = struct{}{}
        return nil
}

func (bus *EventBus) releaseReplayID(subscriptionID string) {
        bus.replayMu.Lock()
        delete(bus.replayIDs, subscriptionID)
        bus.replayMu.Unlock()
}

func (bus *EventBus) registerReplayMarker(marker *internalHandler) {
        bus.replayMu.Lock()
        if _, exists := bus.replayMarkers[marker]; !exists {
                bus.replayMarkers[marker] = struct{}{}
                bus.replayMarkerCount.Add(1)
        }
        bus.replayMu.Unlock()
}

func (bus *EventBus) unregisterReplayMarker(marker *internalHandler) {
        bus.replayMu.Lock()
        if _, exists := bus.replayMarkers[marker]; exists {
                delete(bus.replayMarkers, marker)
                bus.replayMarkerCount.Add(-1)
        }
        bus.replayMu.Unlock()
}

package eventbus

import (
        "crypto/rand"
        "sync"
        "time"
)

// crockford32 is the Crockford base32 alphabet used by ULIDs: no I, L, O, U,
// so identifiers are unambiguous when read or transcribed.
const crockford32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"

// ulidEntropy buffers crypto/rand reads. Reading 10 bytes per ID directly
// from crypto/rand costs a syscall each time; a buffered reader amortizes it
// across ~100 IDs while keeping the same entropy source.
var ulidEntropy = struct {
        sync.Mutex
        buf []byte
        pos int
}{buf: make([]byte, 1024), pos: 1024}

// NewEventID returns a new ULID: a 26-character, lexicographically sortable,
// globally unique identifier (48-bit millisecond timestamp + 80 bits of
// crypto/rand entropy, Crockford base32).
//
// The bus assigns one to every event it persists (see Event.ID). It is
// exported so custom stores and tests can mint compatible IDs.
//
// IDs sort by creation time to millisecond precision; ties are ordered
// randomly. Uniqueness does not depend on the clock: the 80 random bits alone
// make collisions vanishingly unlikely even at the same millisecond.
func NewEventID() string {
        var bin [16]byte // 6 timestamp bytes + 10 entropy bytes

        ms := uint64(time.Now().UnixMilli())
        bin[0] = byte(ms >> 40)
        bin[1] = byte(ms >> 32)
        bin[2] = byte(ms >> 24)
        bin[3] = byte(ms >> 16)
        bin[4] = byte(ms >> 8)
        bin[5] = byte(ms)

        ulidEntropy.Lock()
        if ulidEntropy.pos+10 > len(ulidEntropy.buf) {
                // crypto/rand.Read never fails on supported platforms (it panics
                // internally on the truly unrecoverable ones), so the error is
                // impossible to surface meaningfully here.
                rand.Read(ulidEntropy.buf)
                ulidEntropy.pos = 0
        }
        copy(bin[6:], ulidEntropy.buf[ulidEntropy.pos:ulidEntropy.pos+10])
        ulidEntropy.pos += 10
        ulidEntropy.Unlock()

        // 128 bits -> 26 base32 characters (the top bit pair of the first
        // character is always zero: 26*5 = 130 bits of space for 128 bits).
        var out [26]byte
        out[0] = crockford32[bin[0]>>5]
        bitPos := 3 // bits already consumed from bin
        for i := 1; i < 26; i++ {
                byteIdx := bitPos / 8
                shift := bitPos % 8
                v := bin[byteIdx] << shift >> 3
                if shift > 3 && byteIdx+1 < len(bin) {
                        v |= bin[byteIdx+1] >> (11 - shift)
                }
                out[i] = crockford32[v&0x1f]
                bitPos += 5
        }
        return string(out[:])
}

package eventbus

import (
        "encoding/json"
        "fmt"
        "reflect"
        "sync"
)

// UpcastFunc transforms event data from one version to another. It receives
// raw JSON and must return transformed data plus exactly the toType declared
// when it is registered; another or empty type produces UpcastContractError.
type UpcastFunc func(data json.RawMessage) (json.RawMessage, string, error)

// UpcastErrorHandler is called when an upcast operation fails
type UpcastErrorHandler func(eventType string, data json.RawMessage, err error)

// Upcaster represents a transformation from one event type to another
type Upcaster struct {
        FromType string     // Source event type
        ToType   string     // Target event type
        Upcast   UpcastFunc // Transformation function
}

// UpcastContractError reports a raw UpcastFunc that returned a type name other
// than the successor declared at registration. The registry, not user output,
// owns chain topology; accepting a different type could bypass cycle checks or
// repeatedly invoke the same upcaster forever.
type UpcastContractError struct {
        FromType     string
        DeclaredType string
        ReturnedType string
}

func (e *UpcastContractError) Error() string {
        return fmt.Sprintf("eventbus: upcast from %s declared successor %q but returned type %q", e.FromType, e.DeclaredType, e.ReturnedType)
}

// upcastRegistry manages all registered upcasters.
// Each source type has at most one upcaster: apply follows a single chain
// (v1 -> v2 -> v3), so a second upcaster for the same source type could
// never run and is rejected at registration instead of silently ignored.
type upcastRegistry struct {
        upcasters    map[string]Upcaster // Map from source type to its upcaster
        mu           sync.RWMutex
        errorHandler UpcastErrorHandler
}

// newUpcastRegistry creates a new upcast registry
func newUpcastRegistry() *upcastRegistry {
        return &upcastRegistry{
                upcasters: make(map[string]Upcaster),
        }
}

// register adds an upcaster to the registry
func (r *upcastRegistry) register(fromType, toType string, upcast UpcastFunc) error {
        if fromType == "" || toType == "" {
                return fmt.Errorf("eventbus: upcast types cannot be empty")
        }
        if fromType == toType {
                return fmt.Errorf("eventbus: cannot upcast type to itself")
        }
        if upcast == nil {
                return fmt.Errorf("eventbus: upcast function cannot be nil")
        }

        r.mu.Lock()
        defer r.mu.Unlock()

        if existing, ok := r.upcasters[fromType]; ok {
                return fmt.Errorf("eventbus: upcaster already registered for type %s (-> %s): each source type has exactly one successor; chain versions (v1 -> v2 -> v3) instead", fromType, existing.ToType)
        }

        // Check for circular dependencies
        if r.wouldCreateCycle(fromType, toType) {
                return fmt.Errorf("eventbus: upcast would create circular dependency")
        }

        r.upcasters[fromType] = Upcaster{
                FromType: fromType,
                ToType:   toType,
                Upcast:   upcast,
        }
        return nil
}

// wouldCreateCycle checks if adding an upcast would create a circular dependency
func (r *upcastRegistry) wouldCreateCycle(fromType, toType string) bool {
        // Check if there's already a path from toType back to fromType
        visited := make(map[string]bool)
        return r.hasCycleDFS(toType, fromType, visited)
}

// hasCycleDFS performs depth-first search to detect cycles
func (r *upcastRegistry) hasCycleDFS(current, target string, visited map[string]bool) bool {
        if current == target {
                return true
        }
        if visited[current] {
                return false
        }
        visited[current] = true

        if upcaster, ok := r.upcasters[current]; ok {
                return r.hasCycleDFS(upcaster.ToType, target, visited)
        }
        return false
}

// apply attempts to apply upcasts to transform data to the latest version.
//
// The registry lock is only held for map lookups, never while a user upcast
// function runs: upcast functions may therefore safely call back into the
// registry (e.g. RegisterUpcast) without deadlocking. A registration that
// races with apply may or may not be observed by the in-flight chain.
func (r *upcastRegistry) apply(data json.RawMessage, eventType string) (json.RawMessage, string, error) {
        currentData := data
        currentType := eventType
        appliedTypes := make(map[string]bool) // Prevent infinite loops

        for {
                // Mark this type as processed
                appliedTypes[currentType] = true

                // Find the next upcaster and snapshot the error handler under the
                // read lock; user code runs after the lock is released.
                r.mu.RLock()
                upcaster, found := r.upcasters[currentType]
                errorHandler := r.errorHandler
                r.mu.RUnlock()

                if !found {
                        break // No more upcasts available
                }

                // Check for loops
                if appliedTypes[upcaster.ToType] {
                        return data, eventType, fmt.Errorf("eventbus: upcast loop detected")
                }

                // Apply user code behind a panic-to-error boundary. Durable Follow and
                // resumable replay can then retry the unchanged offset instead of losing
                // their goroutine/process to an unhandled upcaster panic.
                newData, newType, err := callUpcast(upcaster.Upcast, currentData)
                if err != nil {
                        if errorHandler != nil {
                                errorHandler(currentType, currentData, err)
                        }
                        return data, eventType, fmt.Errorf("eventbus: upcast failed from %s to %s: %w",
                                upcaster.FromType, upcaster.ToType, err)
                }
                if newType != upcaster.ToType {
                        contractErr := &UpcastContractError{
                                FromType:     upcaster.FromType,
                                DeclaredType: upcaster.ToType,
                                ReturnedType: newType,
                        }
                        if errorHandler != nil {
                                errorHandler(currentType, currentData, contractErr)
                        }
                        return data, eventType, contractErr
                }

                currentData = newData
                currentType = newType
        }

        return currentData, currentType, nil
}

func callUpcast(upcast UpcastFunc, data json.RawMessage) (newData json.RawMessage, newType string, err error) {
        defer func() {
                if recovered := recover(); recovered != nil {
                        err = fmt.Errorf("eventbus: upcast panic: %v", recovered)
                }
        }()
        return upcast(data)
}

// clear removes all registered upcasters
func (r *upcastRegistry) clear() {
        r.mu.Lock()
        defer r.mu.Unlock()
        r.upcasters = make(map[string]Upcaster)
}

// clearType removes all upcasters for a specific source type
func (r *upcastRegistry) clearType(eventType string) {
        r.mu.Lock()
        defer r.mu.Unlock()
        delete(r.upcasters, eventType)
}

// RegisterUpcast registers a type-safe upcast function.
//
// From and To must be concrete types. Interface types are rejected with an
// error: upcasts are keyed by type name, and events are stored under their
// concrete type's name (or TypeNamer name), so an upcast keyed by an
// interface type name could never match a stored event — a silently dead
// registration.
func RegisterUpcast[From any, To any](bus *EventBus, upcast func(From) To) error {
        if bus == nil {
                return fmt.Errorf("eventbus: bus cannot be nil")
        }
        if upcast == nil {
                return fmt.Errorf("eventbus: upcast function cannot be nil")
        }

        from := reflect.TypeOf((*From)(nil)).Elem()
        to := reflect.TypeOf((*To)(nil)).Elem()
        if from.Kind() == reflect.Interface || to.Kind() == reflect.Interface {
                return fmt.Errorf("eventbus: cannot register upcast between interface types (%s -> %s): events are stored under concrete type names, so an interface-keyed upcast would never match", from, to)
        }

        // Use the same TypeNamer-aware derivation as persistence, replay, and
        // Follow. Reflection-only names silently miss events that use stable
        // EventTypeName values on the wire.
        fromType := typeNameOf(from)
        toType := typeNameOf(to)
        typeReservation, err := bus.reservePersistedTypes(
                persistedTypeSpec{name: fromType, eventType: from},
                persistedTypeSpec{name: toType, eventType: to},
        )
        if err != nil {
                return err
        }
        defer typeReservation.Rollback()

        upcastFunc := func(data json.RawMessage) (json.RawMessage, string, error) {
                var from From
                if err := json.Unmarshal(data, &from); err != nil {
                        return nil, "", fmt.Errorf("unmarshal source: %w", err)
                }

                to := upcast(from)

                newData, err := json.Marshal(to)
                if err != nil {
                        return nil, "", fmt.Errorf("marshal target: %w", err)
                }

                return newData, toType, nil
        }

        if err := bus.upcastRegistry.register(fromType, toType, upcastFunc); err != nil {
                return err
        }
        typeReservation.Commit()
        return nil
}

// RegisterUpcastFunc registers a raw upcast function for complex
// transformations. The function must return exactly toType as its new type;
// a mismatch is an UpcastContractError and the original event remains
// unadvanced. Because raw string endpoints carry no Go reflect.Type,
// they cannot participate in the typed persisted-name collision registry;
// callers are responsible for choosing globally unique, compatible names.
func RegisterUpcastFunc(bus *EventBus, fromType, toType string, upcast UpcastFunc) error {
        if bus == nil {
                return fmt.Errorf("eventbus: bus cannot be nil")
        }
        return bus.upcastRegistry.register(fromType, toType, upcast)
}

// WithUpcast adds a raw, string-keyed upcast function during bus creation.
// Like RegisterUpcastFunc, it cannot participate in typed name-collision
// checks because no Go endpoint types are available.
//
// It panics if the registration is invalid (empty types, self-upcast, nil
// function, or a circular dependency): these are programming errors that
// would otherwise be silently ignored at startup. Use RegisterUpcastFunc if
// you need an error value instead.
func WithUpcast(fromType, toType string, upcast UpcastFunc) Option {
        return func(bus *EventBus) {
                if err := bus.upcastRegistry.register(fromType, toType, upcast); err != nil {
                        panic(fmt.Sprintf("eventbus: WithUpcast(%q, %q): %v", fromType, toType, err))
                }
        }
}

// setErrorHandler sets the error handler under the registry lock so it is
// safe to call concurrently with apply.
func (r *upcastRegistry) setErrorHandler(handler UpcastErrorHandler) {
        r.mu.Lock()
        defer r.mu.Unlock()
        r.errorHandler = handler
}

// WithUpcastErrorHandler sets the error handler for upcast failures
func WithUpcastErrorHandler(handler UpcastErrorHandler) Option {
        return func(bus *EventBus) {
                bus.upcastRegistry.setErrorHandler(handler)
        }
}

// SetUpcastErrorHandler sets the upcast error handler at runtime.
// Safe to call concurrently with replay/publish.
func (bus *EventBus) SetUpcastErrorHandler(handler UpcastErrorHandler) {
        bus.upcastRegistry.setErrorHandler(handler)
}

// ClearUpcasts removes all registered upcasters
func (bus *EventBus) ClearUpcasts() {
        bus.upcastRegistry.clear()
}

// ClearUpcastsForType removes all upcasters for a specific source type
func (bus *EventBus) ClearUpcastsForType(eventType string) {
        bus.upcastRegistry.clearType(eventType)
}