gala

type AttachOption func(*Gala) error

type Config struct {
	// DispatchMode controls whether events are dispatched durably (River) or in-memory.
	DispatchMode DispatchMode
	// ConnectionURI is the database connection URI used for the dedicated gala river client
	ConnectionURI string
	// QueueName is the gala queue used for durable dispatch jobs
	QueueName string
	// WorkerCount is the max worker concurrency for the gala queue
	WorkerCount int
	// MaxRetries sets max attempts for gala dispatch jobs when greater than zero
	MaxRetries int
	// RunMigrations enables River schema migrations on startup (use for tests only)
	RunMigrations bool
	// FetchCooldown is the minimum time between job fetches per worker (default 100ms, min 1ms)
	// Lower values increase throughput but also database load. River enforces 1ms minimum.
	FetchCooldown time.Duration
	// FetchPollInterval is the fallback polling interval when LISTEN/NOTIFY misses events (default 1s)
	// This is only used when LISTEN/NOTIFY fails to deliver notifications.
	FetchPollInterval time.Duration
	// Kinds are the envelope kind namespaces this runtime registers alongside the default
	// dispatch kind; each kind gets a dedicated queue and shares the envelope worker
	Kinds []Namespace
	// TopicRenames maps retired topic names to their designated replacements, applied to
	// queued envelopes at dispatch and during job migration
	TopicRenames map[TopicName]TopicName
	// OperationRenames maps retired payload operation strings to their designated replacements
	OperationRenames map[string]string
	// JobTimeout is the maximum run time for one dispatch job; long-running batch
	// operations need far more than River's one-minute default
	JobTimeout time.Duration
	// SoftStopTimeout is how long in-flight jobs may keep running after a stop begins
	// before their contexts are cancelled
	SoftStopTimeout time.Duration
}

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

type ContextKey = string

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

type ContextSnapshot struct {
	// Values contains codec-managed context values
	Values map[ContextKey]json.RawMessage `json:"values,omitempty"`
}

type Definition[T any] struct {
	// Topic is the topic handled by this listener
	Topic Topic[T]
	// Name is the stable listener name; empty defaults to the topic name
	Name string
	// Operations optionally scopes listener interest to specific mutation operations
	// Empty means the listener accepts all operations for the topic
	Operations []string
	// Gate optionally drops the event silently when it returns false; it receives the
	// restored context before any caller or context-key mutation
	Gate func(context.Context, T) bool
	// Caller optionally replaces or augments the restored caller (never nil)
	Caller func(restored *auth.Caller, payload T) *auth.Caller
	// ContextKeys are applied to the context in order before the handler runs
	ContextKeys []func(context.Context) context.Context
	// LogFields is merged over the automatic dispatch log fields
	LogFields func(T) map[string]any
	// Cancel optionally classifies a handler error as terminal, converting it to river.JobCancel
	Cancel func(context.Context, T, error) bool
	// OnExhausted runs when a scheduled loop stops on its error-streak budget
	OnExhausted func(context.Context, T, error)
	// Schedule makes this listener a self-sustaining adaptive re-emit loop when non-nil;
	// exactly one of Handle and Schedule.Handle must be set
	Schedule *ScheduleSpec[T]
	// Handle is the callback invoked for this listener
	Handle Handler[T]
}

type DispatchMode string

type EmitOption func(*Envelope)

type Envelope struct {
	// ID is the unique event identifier
	ID EventID `json:"id"`
	// Topic is the destination topic
	Topic TopicName `json:"topic"`
	// OccurredAt is the emit timestamp in UTC
	OccurredAt time.Time `json:"occurred_at"`
	// Headers holds operational metadata
	Headers Headers `json:"headers"`
	// Payload is encoded topic payload data
	Payload json.RawMessage `json:"payload"`
	// ContextSnapshot holds restorable context metadata
	ContextSnapshot ContextSnapshot `json:"context_snapshot"`
}

type EnvelopeArgs struct {
	// Envelope is the encoded gala envelope payload
	Envelope []byte `json:"envelope"`
	// UniqueKey scopes the ByArgs uniqueness hash to this field alone via the river tag
	UniqueKey string `json:"unique_key,omitempty" river:"unique"`
}

type EventID string

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

type Handler[T any] func(HandlerContext, T) error

type HandlerContext struct {
	// Context is the restored event context used for listener execution
	Context context.Context
	// Envelope is the envelope being processed
	Envelope Envelope
	// Injector provides typed dependency lookup via samber/do
	Injector do.Injector
	// Caller is the pre-resolved caller for this dispatch
	Caller *auth.Caller
}

type Headers struct {
	// Properties stores additional metadata for UI visibility
	Properties map[string]string `json:"properties,omitempty"`
	// Tags are low-cardinality labels forwarded to the transport layer (e.g. River job tags)
	Tags []string `json:"tags,omitempty"`
	// Listeners are the registered listener names for the topic, populated at dispatch time
	Listeners []string `json:"listeners,omitempty"`
	// Kind optionally routes dispatch to a registered job kind
	Kind string `json:"kind,omitempty"`
	// ScheduledAt defers execution until the specified time; nil means immediate
	ScheduledAt *time.Time `json:"scheduled_at,omitempty"`
	// UniqueKey enforces at most one live job per key at insert time
	UniqueKey string `json:"unique_key,omitempty"`
	// SkipUniqueKey suppresses the topic's UniqueKey derivation
	SkipUniqueKey bool `json:"skip_unique_key,omitempty"`
	// UniqueOnce extends UniqueKey matching to terminal job states
	UniqueOnce bool `json:"unique_once,omitempty"`
	// Metadata carries structured operation context as opaque JSON
	Metadata json.RawMessage `json:"metadata,omitempty"`
}

type ListenerError struct {
	// ListenerName is the name of the listener that failed
	ListenerName string
	// Cause is the underlying error from the listener
	Cause error
	// Panicked indicates whether the listener panicked
	Panicked bool
}

type ListenerID string

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

type OperationContext struct {
	// OwnerID is the owning organization for the operation
	OwnerID string `json:"ownerId,omitempty" jsonschema:"description=Owning organization identifier"`
	// Operation is the mutation type when applicable: CREATE, UPDATE, DELETE
	Operation string `json:"operation,omitempty" jsonschema:"description=Mutation type when applicable: CREATE, UPDATE, DELETE"`
	// EntityID is the identifier of the entity the operation targets
	EntityID string `json:"entityId,omitempty" jsonschema:"description=Identifier of the target entity"`
	// EntityType is the schema type of the target entity
	EntityType string `json:"entityType,omitempty" jsonschema:"description=Schema type of the target entity"`
	// Attributes is the source-specific provenance payload, decoded on demand
	Attributes json.RawMessage `json:"attributes,omitempty" jsonschema:"description=Source-specific provenance payload"`
}

type PayloadOperation interface {
	PayloadOperation() string
}

type PayloadOperationRenamer interface {
	WithPayloadOperation(operation string) any
}

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

type PoolOption func(*Pool)

type Registration interface {
	// Attach registers the listener and its topic contract on the runtime
	Attach(g *Gala) (ListenerID, error)
}

type Schedule struct {
	// MinInterval is the shortest allowed interval between runs
	MinInterval time.Duration `json:"min_interval"`
	// MaxInterval is the longest allowed interval between runs
	MaxInterval time.Duration `json:"max_interval"`
	// BackoffFactor is the multiplier applied when backing off (idle or error)
	BackoffFactor float64 `json:"backoff_factor"`
	// HighDriftThreshold is the delta count above which the interval resets to MinInterval
	HighDriftThreshold int `json:"high_drift_threshold"`
	// MaxErrorStreak stops the loop after this many consecutive failed cycles; UnlimitedErrorStreak disables exhaustion
	MaxErrorStreak int `json:"max_error_streak"`
}

type ScheduleSpec[T any] struct {
	// Schedule controls adaptive interval computation
	Schedule Schedule
	// Handle is the handler invoked each cycle
	Handle func(context.Context, T) (int, error)
	// State extracts the ScheduleState from the envelope
	State func(T) ScheduleState
	// Wrap builds a new envelope carrying the updated ScheduleState
	Wrap func(T, ScheduleState) T
	// PrepareEmit optionally enriches the context and headers before re-emitting
	PrepareEmit func(context.Context, T) (context.Context, Headers)
	// Override optionally returns a per-envelope schedule that overrides Schedule
	Override func(T) *Schedule
}

type ScheduleState struct {
	// Incarnation identifies one logical schedule chain across every cycle
	Incarnation string `json:"incarnation,omitempty"`
	// Interval is the current scheduling interval
	Interval time.Duration `json:"interval"`
	// IdleStreak is the number of consecutive runs with zero delta
	IdleStreak int `json:"idle_streak"`
	// ErrorStreak is the number of consecutive runs that returned an error
	ErrorStreak int `json:"error_streak"`
	// Cycle is the monotonic cycle counter
	Cycle int `json:"cycle"`
}

type Topic[T any] struct {
	// Name is the stable topic identifier
	Name TopicName
	// Kind is the job kind emissions on this topic dispatch under
	Kind string
	// UniqueKey optionally derives Headers.UniqueKey from the payload for every emission on the topic
	UniqueKey func(T) string
}

type TopicName string

type TopicOption[T any] func(*Topic[T])

type WorkflowFlags struct {
	// Bypass skips workflow approval interceptors for system operations
	Bypass bool `json:"bypass,omitempty"`
	// AllowEventEmission keeps workflow listener execution enabled while Bypass is set
	AllowEventEmission bool `json:"allow_event_emission,omitempty"`
}