sse

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

type BrokerConfig struct {
	// SubscriberBuffer bounds each subscriber's outbound queue inside the
	// broker. Fast subscribers never see backpressure; slow ones get the
	// OnSlowSubscriber policy applied. Default 64.
	SubscriberBuffer int

	// OnSlowSubscriber is consulted when a subscriber's queue is full at
	// Publish time. When nil, [BrokerPolicyDrop] is used. The Drop
	// default is deliberate for SSE: dropping one event from one slow
	// subscriber leaves the stream coherent (the next event with a
	// fresh id replaces it, and replay via Last-Event-ID can recover).
	// websocket.HubConfig.OnSlowConn defaults to Close instead — a
	// dropped WS frame would corrupt the message-boundary contract,
	// so the safer default is to evict the bad peer.
	OnSlowSubscriber func(c *Client, pe *PreparedEvent) BrokerPolicy

	// SlowSubscriberConcurrency caps the number of in-flight per-
	// subscriber slow-path goroutines spawned by Publish/PublishPrepared
	// when one or more subscribers' queues are full. Zero means
	// [DefaultBrokerSlowConcurrency] — runtime.GOMAXPROCS(0)*4. Negative
	// opts out (true unbounded) — useful for benchmarks; production
	// callers should keep the cap on so a misbehaving OnSlowSubscriber
	// callback cannot fan out into thousands of goroutines.
	SlowSubscriberConcurrency int
}

type BrokerPolicy uint8

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

type ClientPolicy uint8

type Config struct {
	// Handler is the SSE handler function called for each connected client.
	// Required; panics at init if nil.
	Handler Handler

	// HeartbeatInterval is the interval between heartbeat comments sent to
	// detect client disconnects. Set to a negative value to disable.
	// Default: 15s.
	HeartbeatInterval time.Duration

	// RetryInterval is the reconnection time (in milliseconds) sent to the
	// client in the initial "retry:" field. Zero means no retry field is sent
	// (client uses its default, typically ~3s).
	RetryInterval int

	// MaxQueueDepth bounds the per-client outbound queue. Zero means
	// unbounded — the legacy blocking Send semantics are preserved exactly.
	// When set, Send enqueues and returns immediately; a per-client drain
	// goroutine writes to the wire. If the queue is full at Send time,
	// OnSlowClient is invoked; if OnSlowClient is nil, [ClientPolicyDrop]
	// is the default.
	MaxQueueDepth int

	// OnSlowClient decides what to do when [MaxQueueDepth] is exceeded.
	// Only consulted when MaxQueueDepth > 0. The hook may inspect c (via
	// DroppedEvents/QueueDepth) and the dropped Event to drive
	// observability or escalating policies. Default: [ClientPolicyDrop].
	OnSlowClient func(c *Client, e Event) ClientPolicy

	// ReplayStore persists events for Last-Event-ID resume. When nil
	// (default), Client.LastEventID() returns the header but replay is
	// the user's problem — matches today's behavior. When set, the
	// middleware:
	//   - on connect with a Last-Event-ID, reads Since(lastID) and
	//     writes the missed events to the wire BEFORE invoking Handler;
	//   - wraps Send so each call also Appends to the store, rewriting
	//     the wire id: field with the canonical store-assigned ID;
	//   - on Since returning [ErrLastIDUnknown], silently falls
	//     through and still invokes Handler — the original header
	//     value remains visible via [Client.LastEventID] so the
	//     user's resumption logic can react.
	//
	// Performance note: synchronous Append runs INSIDE the per-Client
	// write lock so wire order matches store order. A slow store
	// (e.g. Redis over a degraded network) directly stalls Send and
	// any concurrent heartbeat or SendComment. For low-latency
	// publishing against a remote KV, prefer
	// [NewKVReplayStore] with [KVReplayStoreConfig.AsyncAppend] = true
	// so Append returns immediately and the actual KV.Set fires in
	// the background.
	ReplayStore ReplayStore

	// OnConnect is called when a new SSE client connects, before Handler.
	// The celeris.Context is available for extracting request metadata.
	// Return a non-nil error to reject the connection.
	OnConnect func(c *celeris.Context, client *Client) error

	// OnDisconnect is called after the SSE stream closes.
	OnDisconnect func(c *celeris.Context, client *Client)

	// Skip defines a function to skip this middleware for certain requests.
	Skip func(c *celeris.Context) bool

	// SkipPaths lists paths to skip from SSE handling (exact match).
	SkipPaths []string
}

type Event struct {
	// ID is the event ID. If non-empty, sent as "id: <ID>\n".
	// The client stores this and sends it back as Last-Event-ID on reconnect.
	ID string

	// Event is the event type. If non-empty, sent as "event: <Event>\n".
	// Defaults to "message" on the client side when omitted.
	Event string

	// Data is the event payload. Sent as "data: <line>\n" for each line.
	// Multi-line data is split on \n and each line gets its own "data:" prefix.
	Data string

	// Retry is the reconnection time in milliseconds. If > 0, sent as
	// "retry: <Retry>\n". Per the SSE specification, this value is in
	// milliseconds (not time.Duration) to match the wire format directly.
	Retry int
}

type Handler func(client *Client)

type KVReplayStoreConfig struct {
	// KV is the backing store. Required.
	KV store.KV

	// Prefix is the namespace under which event blobs and the
	// shared counter live. Defaults to no prefix when empty.
	Prefix string

	// TTL bounds the lifetime of stored event blobs. Zero means no
	// expiry — same convention as [store.KV.Set].
	TTL time.Duration

	// MaxIndex bounds the in-memory ID index. Zero or negative ⇒
	// [MaxKVReplayIndex].
	MaxIndex int

	// AsyncAppend, when true, makes [ReplayStore.Append] return as
	// soon as the local ID has been allocated — the actual KV.Set
	// fires in a background goroutine. Trades durability for latency:
	//   - An Append() that returns successfully is NOT guaranteed to
	//     be in the KV when the next reconnect happens.
	//   - A Since() that runs concurrently with a still-flight Append
	//     may observe ErrNotFound on that id and skip it. Documented
	//     visibility window for the AsyncAppend mode.
	// Use only when wire-write latency dominates and replay can
	// tolerate eventual consistency.
	//
	// Backpressure: at the [KVReplayStoreConfig.AsyncAppendConcurrency]
	// cap, Append blocks the caller until a goroutine slot frees. This
	// is intentional — under heavy publish load against a stalled KV,
	// blocking the publisher is preferable to letting goroutine count
	// explode. Visible-latency backpressure beats unbounded memory
	// growth.
	AsyncAppend bool

	// AsyncAppendConcurrency caps the number of in-flight goroutines
	// spawned by AsyncAppend. Zero means [DefaultAsyncAppendConcurrency].
	// A slow / stalled KV (e.g. a Redis instance under degraded
	// network) without this cap could pile up unbounded goroutines.
	// At the cap, Append blocks (see [KVReplayStoreConfig.AsyncAppend]).
	AsyncAppendConcurrency int

	// CounterKey is the KV key under which the cross-instance counter
	// lives. Zero value ⇒ "<prefix>seq". Only consulted when the
	// supplied [store.KV] also implements [store.Counter]; otherwise
	// the constructor falls back to a per-process counter.
	CounterKey string
}

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

type ReplayStore interface {
	// Append records the event and returns the canonical ID the
	// middleware will emit on the wire. Implementations SHOULD ignore
	// e.ID and assign their own monotonically increasing ID.
	Append(ctx context.Context, e Event) (id string, err error)

	// Since returns every event appended strictly after lastID, in
	// append order. lastID == "" means "no checkpoint" — implementations
	// may return the empty slice or a recent-N tail (default: empty).
	// Returns ErrLastIDUnknown if lastID is non-empty but cannot be
	// interpreted.
	Since(ctx context.Context, lastID string) ([]Event, error)
}