core

type Backfill struct {
	MessageType string        `json:"messageType"`
	Input       BackfillInput `json:"input"`
	ConsumerID  string        `json:"consumerID"`
	Stream      string        `json:"stream"`
	Provider    string        `json:"provider"`
	State       string        `json:"state"`
	Next        uint64        `json:"next"`
	Published   uint64        `json:"published"`
	Skipped     uint64        `json:"skipped"`
	UpdatedAt   time.Time     `json:"updatedAt"`
	Error       string        `json:"error,omitempty"`
	Persisted   bool          `json:"persisted"`
}

type BackfillInput struct {
	ID           string `json:"id"`
	Subscription string `json:"subscription"`
	Start        uint64 `json:"start"`
	End          uint64 `json:"end"`
}

type Binding struct {
	Identity     Identity
	Subscription SubscriptionConfig
	Stream       StreamConfig
}

type Capabilities struct {
	ConsumerControls bool `json:"consumerControls"`
	Backfill         bool `json:"backfill"`
	RPC              bool `json:"rpc"`
	Durable          bool `json:"durable"`
	Replay           bool `json:"replay"`
	DeadLetters      bool `json:"deadLetters"`
	KeyOrdering      bool `json:"keyOrdering"`
}

type Config struct {
	RPC             RPCConfig                     `json:"rpc"             yaml:"rpc"`
	Identity        Identity                      `json:"identity"        yaml:"identity"`
	Streams         map[string]StreamConfig       `json:"streams"         yaml:"streams"`
	Subscriptions   map[string]SubscriptionConfig `json:"subscriptions"   yaml:"subscriptions"`
	ObserverBuffer  int                           `json:"observerBuffer"  yaml:"observer_buffer"`
	MaxPayloadBytes int                           `json:"maxPayloadBytes" yaml:"max_payload_bytes"`
	Providers       map[string]ConnectionConfig   `json:"providers"       yaml:"providers"`
	Discovery       string                        `json:"discovery"       yaml:"discovery"`
	Version         string                        `json:"version"         yaml:"version"`
	Endpoints       []Endpoint                    `json:"endpoints"       yaml:"endpoints"`
}

type ConnectionConfig struct {
	Type               string `json:"type"               yaml:"type"`
	URL                string `json:"url"                yaml:"url"`
	DeadLetterReplicas int    `json:"deadLetterReplicas" yaml:"dead_letter_replicas"`
}

type ConsumerInfo struct {
	Subscription SubscriptionConfig `json:"subscription"`
	Provider     string             `json:"provider"`
	ConsumerID   string             `json:"consumerID"`
	Pending      uint64             `json:"pending"`
	AckPending   uint64             `json:"ackPending"`
	Redelivered  uint64             `json:"redelivered"`
	Paused       bool               `json:"paused"`
	Processing   Latency            `json:"processing"`
	Delivery     Latency            `json:"delivery"`
}

type DeadLetter struct {
	ID       string       `json:"id"`
	Message  Envelope     `json:"message"`
	Delivery DeliveryInfo `json:"delivery"`
	FailedAt time.Time    `json:"failedAt"`
	Reason   string       `json:"reason"`
	Replayed bool         `json:"replayed"`
}

type Delivery interface {
	Message() Envelope
	Info() DeliveryInfo
	Ack(ctx context.Context) error
	Retry(ctx context.Context, delay time.Duration) error
	Reject(ctx context.Context) error
}

type DeliveryInfo struct {
	Stream         string       `json:"stream"`
	ConsumerID     string       `json:"consumerID"`
	SubscriptionID string       `json:"subscriptionID"`
	Destination    Identity     `json:"destination"`
	Mode           DeliveryMode `json:"mode"`
	Attempt        uint64       `json:"attempt"`
	Sequence       uint64       `json:"sequence"`
}

type DeliveryMode string

type Endpoint struct {
	Protocol string `json:"protocol" yaml:"protocol"`
	URL      string `json:"url"      yaml:"url"`
}

type Envelope struct {
	ID            string            `json:"id"`
	Type          string            `json:"type"`
	Source        Identity          `json:"source"`
	ContentType   string            `json:"contentType"`
	Data          json.RawMessage   `json:"data"`
	CreatedAt     time.Time         `json:"createdAt"`
	Key           string            `json:"key,omitempty"`
	CorrelationID string            `json:"correlationID,omitempty"`
	CausationID   string            `json:"causationID,omitempty"`
	Headers       map[string]string `json:"headers,omitempty"`
	// TargetConsumer directs recovery to the original subscription.
	TargetConsumer string `json:"targetConsumer,omitempty"`
}

type HandleHook interface {
	Hook
	BeforeHandle(ctx context.Context, message Envelope, info DeliveryInfo) error
}

type Handler func(context.Context, Envelope, DeliveryInfo) error

type Hook interface{ Name() string }

type HookEvent struct {
	Duration time.Duration `json:"duration,omitempty"`
	Stage    Stage         `json:"stage"`
	Identity Identity      `json:"identity"`
	Provider string        `json:"provider,omitempty"`
	Message  *Envelope     `json:"message,omitempty"`
	Delivery *DeliveryInfo `json:"delivery,omitempty"`
	Receipt  *Receipt      `json:"receipt,omitempty"`
	Error    string        `json:"error,omitempty"`
	At       time.Time     `json:"at"`
}

type HookFuncs struct {
	HookName  string
	OnEvent   func(context.Context, HookEvent)
	OnPublish func(context.Context, *Envelope) error
	OnHandle  func(context.Context, Envelope, DeliveryInfo) error
}

type Identity struct {
	Namespace  string `json:"namespace"  yaml:"namespace"`
	ServiceID  string `json:"serviceID"  yaml:"service_id"`
	InstanceID string `json:"instanceID" yaml:"instance_id"`
}

type Instance struct {
	Identity  Identity   `json:"identity"`
	Version   string     `json:"version"`
	Endpoints []Endpoint `json:"endpoints"`
	Ready     bool       `json:"ready"`
}

type InstanceLister interface {
	List(ctx context.Context, namespace string) ([]Instance, error)
}

type Latency struct {
	Count   uint64        `json:"count"`
	Total   time.Duration `json:"total"`
	Max     time.Duration `json:"max"`
	Average time.Duration `json:"average"`
}

type Management interface {
	Inspect(ctx context.Context, namespace string, config StreamConfig) (StreamInfo, error)
	StoreDeadLetter(ctx context.Context, namespace string, letter DeadLetter) error
	ListDeadLetters(ctx context.Context, identity Identity, subscription string, cursor string, limit int) ([]DeadLetter, string, error)
	ReplayDeadLetter(ctx context.Context, identity Identity, subscription string, id string) (Receipt, error)
}

type Middleware func(Handler) Handler

type Observer interface {
	Hook
	Observe(ctx context.Context, event HookEvent)
}

type Operations interface {
	InspectConsumer(ctx context.Context, binding Binding) (ConsumerInfo, error)
	PauseConsumer(ctx context.Context, binding Binding, paused bool) error
	RunBackfill(ctx context.Context, binding Binding, input BackfillInput) (Backfill, error)
	ListBackfills(ctx context.Context, identity Identity, cursor string, limit int) ([]Backfill, string, error)
}

type Option func(*Runtime) error

type Provider interface {
	Name() string
	Capabilities() Capabilities
	Connect(ctx context.Context) error
	Close(ctx context.Context) error
	Health(ctx context.Context) error
	EnsureStream(ctx context.Context, namespace string, config StreamConfig) error
	Publish(ctx context.Context, namespace string, config StreamConfig, message Envelope) (Receipt, error)
	Subscribe(ctx context.Context, binding Binding) (Subscription, error)
}

type ProviderInfo struct {
	Name         string       `json:"name"`
	Type         string       `json:"type"`
	Capabilities Capabilities `json:"capabilities"`
	Healthy      bool         `json:"healthy"`
}

type PublishHook interface {
	Hook
	BeforePublish(ctx context.Context, message *Envelope) error
}

type RPCBinding struct {
	Identity        Identity
	Method          string
	Config          RPCConfig
	MaxPayloadBytes int
}

type RPCCode string

type RPCConfig struct {
	Provider    string        `json:"provider"    yaml:"provider"`
	Timeout     time.Duration `json:"timeout"     yaml:"timeout"`
	Concurrency int           `json:"concurrency" yaml:"concurrency"`
	MaxInFlight int           `json:"maxInFlight" yaml:"max_in_flight"`
}

type RPCError struct {
	Code    RPCCode `json:"code"`
	Message string  `json:"message"`
}

type RPCHandler func(context.Context, Envelope) (json.RawMessage, error)

type RPCProvider interface {
	RequestRPC(ctx context.Context, identity Identity, request RPCRequest) (RPCResponse, error)
	ServeRPC(ctx context.Context, binding RPCBinding, handler func(context.Context, RPCRequest) RPCResponse) (RPCServer, error)
}

type RPCRequest struct {
	Service  string    `json:"service"`
	Method   string    `json:"method"`
	Envelope Envelope  `json:"envelope"`
	Deadline time.Time `json:"deadline"`
}

type RPCResponse struct {
	ID     string          `json:"id"`
	Source Identity        `json:"source"`
	Data   json.RawMessage `json:"data,omitempty"`
	Error  *RPCError       `json:"error,omitempty"`
}

type RPCServer interface {
	Close(ctx context.Context) error
}

type Receipt struct {
	MessageID string `json:"messageID"`
	Sequence  uint64 `json:"sequence"`
	Persisted bool   `json:"persisted"`
	Duplicate bool   `json:"duplicate"`
}

type Registry interface {
	Resolver
	Register(ctx context.Context, instance Instance) error
	Deregister(ctx context.Context, identity Identity) error
}

type Resolver interface {
	Resolve(ctx context.Context, namespace string, service string) ([]Instance, error)
}

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

type Snapshot struct {
	Identity      Identity             `json:"identity"`
	Running       bool                 `json:"running"`
	StartedAt     time.Time            `json:"startedAt"`
	Providers     []ProviderInfo       `json:"providers"`
	Streams       []StreamInfo         `json:"streams"`
	Subscriptions []SubscriptionConfig `json:"subscriptions"`
	Published     uint64               `json:"published"`
	Handled       uint64               `json:"handled"`
	Acknowledged  uint64               `json:"acknowledged"`
	Failed        uint64               `json:"failed"`
	Retried       uint64               `json:"retried"`
	DeadLettered  uint64               `json:"deadLettered"`
	ObserverDrops uint64               `json:"observerDrops"`
	RPCCalls      uint64               `json:"rpcCalls"`
	RPCHandled    uint64               `json:"rpcHandled"`
	RPCFailed     uint64               `json:"rpcFailed"`
	RPCTimedOut   uint64               `json:"rpcTimedOut"`
}

type Stage string

type StreamConfig struct {
	Name        string        `json:"name"        yaml:"name"`
	Provider    string        `json:"provider"    yaml:"provider"`
	Subjects    []string      `json:"subjects"    yaml:"subjects"`
	MaxAge      time.Duration `json:"maxAge"      yaml:"max_age"`
	MaxMessages int64         `json:"maxMessages" yaml:"max_messages"`
	Replicas    int           `json:"replicas"    yaml:"replicas"`
}

type StreamInfo struct {
	Config    StreamConfig `json:"config"`
	Messages  uint64       `json:"messages"`
	Bytes     uint64       `json:"bytes"`
	Consumers int          `json:"consumers"`
}

type Subscription interface {
	Next(ctx context.Context) (Delivery, error)
	Close(ctx context.Context) error
}

type SubscriptionConfig struct {
	ID            string        `json:"id"                      yaml:"id"`
	Stream        string        `json:"stream"                  yaml:"stream"`
	MessageType   string        `json:"messageType"             yaml:"message_type"`
	Mode          DeliveryMode  `json:"mode"                    yaml:"delivery"`
	Durable       bool          `json:"durable"                 yaml:"durable"`
	Concurrency   int           `json:"concurrency"             yaml:"concurrency"`
	MaxInFlight   int           `json:"maxInFlight"             yaml:"max_in_flight"`
	Timeout       time.Duration `json:"timeout"                 yaml:"timeout"`
	MaxAttempts   int           `json:"maxAttempts"             yaml:"max_attempts"`
	RetryDelay    time.Duration `json:"retryDelay"              yaml:"retry_delay"`
	StartAt       string        `json:"startAt"                 yaml:"start_at"`
	StartSequence uint64        `json:"startSequence,omitempty" yaml:"start_sequence"`
	// BroadcastID pins a durable broadcast subscriber across process restarts.
	BroadcastID string `json:"broadcastID,omitempty" yaml:"broadcast_id"`
}