subscription

type BatchHandler[E any] func(ctx context.Context, events []GlobalEvent[E]) error

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

type Checkpoint interface {
	// Load returns the last processed sequence for a consumer. Returns 0 if new.
	Load(ctx context.Context, consumerID string) (uint64, error)

	// Save persists the consumer's position. Must be atomic.
	Save(ctx context.Context, consumerID string, sequence uint64) error
}

type CheckpointInfo struct {
	// ConsumerID is the subscription's consumer identifier.
	ConsumerID string

	// Sequence is the global sequence number of the event being processed.
	Sequence uint64
}

type Config[E any] struct {
	// ConsumerID uniquely identifies this consumer. Required.
	ConsumerID string

	// Reader provides access to the global event stream. Required.
	Reader GlobalReader[E]

	// Checkpoint tracks consumed position. Required.
	Checkpoint Checkpoint

	// Handler processes individual events. Either Handler or BatchHandler required.
	Handler Handler[E]

	// BatchHandler processes events in batches. Mutually exclusive with Handler.
	BatchHandler BatchHandler[E]

	// Notifier provides low-latency event notifications. Optional.
	// Without it, the subscription relies on polling only.
	Notifier StoreNotifier

	// PollInterval is the fallback polling interval. Default: 500ms.
	// Belt and suspenders — ensures delivery even if notifier drops signals.
	PollInterval time.Duration

	// BatchSize is the max events to read per poll. Default: 100.
	BatchSize int

	// MaxRetries before sending to DLQ. Default: 5.
	MaxRetries int

	// RetryBaseDelay is the base delay for exponential backoff. Default: 100ms.
	RetryBaseDelay time.Duration

	// RetryMaxDelay caps the backoff. Default: 10s.
	RetryMaxDelay time.Duration

	// OnDLQ is called when an event exhausts retries. Optional.
	// If nil, the error is logged and the event is skipped.
	OnDLQ func(ctx context.Context, event GlobalEvent[E], err error)

	// Logger for operational logging. Optional.
	Logger *slog.Logger

	// GapTimeout is how long to wait for a sequence gap to fill before
	// assuming it was a rollback. Default: 500ms.
	// Set to 0 to disable gap detection (single-writer mode).
	GapTimeout time.Duration

	// MaxGapWait is the total time willing to wait for gaps before moving on.
	// Default: 5s.
	MaxGapWait time.Duration

	// OnGapTimeout controls what happens when a gap times out (GapTimeout elapsed).
	// Default (GapActionError): stop processing at the gap, retry on next poll.
	// GapActionSkip: skip the gap immediately (old behavior, risks missing events).
	OnGapTimeout GapAction

	// SequenceChecker verifies whether a specific global sequence exists in the
	// store. Used as a self-healing check before skipping a gap after MaxGapWait.
	// If nil, gaps are skipped without verification after MaxGapWait.
	SequenceChecker SequenceChecker

	// EventTypes filters events by their EventType field. Only events whose
	// EventType is in this list will be delivered to the handler.
	// If empty or nil, all events are delivered (no filtering).
	// Checkpoint still advances past filtered-out events to ensure progress.
	EventTypes []string

	// CheckpointEvery saves the checkpoint every N events instead of every event.
	// Default: 1 (save after every event — current behavior).
	// Set to 100 to save every 100th event. On crash, at most N-1 events
	// may be replayed (idempotent handlers recommended for N > 1).
	CheckpointEvery int

	// CheckpointMaxAge is the maximum time between checkpoint saves.
	// When set, a checkpoint is saved when either CheckpointEvery OR
	// CheckpointMaxAge is reached, whichever comes first.
	// Default: 0 (disabled — only count-based batching).
	CheckpointMaxAge time.Duration

	// AtomicCheckpoint indicates that the Handler saves the checkpoint within
	// its own transaction (e.g., via CheckpointFromContext). When true, the
	// subscription injects CheckpointInfo into the context and skips the
	// external Checkpoint.Save call after each event.
	//
	// This prevents double processing on crash recovery for non-idempotent
	// side effects. See pgview and sqlview CheckpointInTx option.
	AtomicCheckpoint bool

	// LagMonitor configures optional lag monitoring. When set, a background
	// goroutine periodically checks how far behind the subscription is and
	// fires the callback when lag exceeds the threshold. Nil means no monitoring.
	LagMonitor *LagMonitorConfig
}

type EventSubscription[E any] struct {
	// contains filtered or unexported fields
}

type GapAction int

type GlobalEvent[E any] struct {
	// GlobalSequence is the store-wide monotonic position. Never reused.
	GlobalSequence uint64

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

	// EventType is the string name of the event (e.g., "OrderCreated").
	// Available when the store has an EventRegistry configured.
	EventType string

	// Version is the per-stream version number.
	Version int

	// Data is the domain event payload.
	Data E

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

type GlobalReader[E any] interface {
	// ReadFrom returns events starting from the given global sequence (inclusive),
	// up to limit events. Returns events in global sequence order.
	ReadFrom(ctx context.Context, fromSequence uint64, limit int) ([]GlobalEvent[E], error)

	// LatestSequence returns the highest global sequence in the store, or 0 if empty.
	LatestSequence(ctx context.Context) (uint64, error)
}

type Handler[E any] func(ctx context.Context, event GlobalEvent[E]) error

type LagCallback func(name string, lag uint64, position uint64, latest uint64)

type LagMonitorConfig struct {
	// Callback is invoked when lag exceeds Threshold.
	Callback LagCallback

	// Threshold is the minimum lag (in events) before the callback fires.
	// For example, 1000 means the callback fires only when the subscription
	// is more than 1000 events behind.
	Threshold uint64

	// CheckInterval is how often the monitor checks the lag. Default: 10s.
	CheckInterval time.Duration
}

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

type MemoryGlobalReader[E any] struct {
	// contains filtered or unexported fields
}

type Option[E any] func(*Config[E])

type SequenceChecker interface {
	SequenceExists(ctx context.Context, seq uint64) (bool, error)
}

type StoreAdapter[E any] struct {
	// contains filtered or unexported fields
}

type StoreNotifier interface {
	Notify(ctx context.Context) <-chan uint64
	Close() error
}

type StoreReader[E any] interface {
	ReadFrom(ctx context.Context, fromSequence uint64, limit int) ([]eskit.Event[E], error)
	LatestSequence(ctx context.Context) (uint64, error)
}

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