sse

Documentation

Overview

Package sse hosts a tiny in-process event bus + HTTP handler that browsers consume via the EventSource API. Replaces the 30-second client polling loop on /problems and /anomalies for state-change events: a Problem opens / resolves, an anomaly fires / clears, and the page updates immediately instead of waiting up to 30s for the next poll.

v0.6.3 — optional Redis pub/sub bridge. In a distributed deployment (COREMETRY_MODE=worker on one pod, mode=api on another) the worker pod fires problem.open via Publish, but browsers are connected to api pods' SSE endpoints. Without a cross-pod fan-out the event vanishes locally. Set a Bridge via SetBridge before any Publish call — every event then also rides a Redis PUBLISH so api pods' Subscribe loops re-deliver them to local subscribers.

Loops: each pod stamps its own podID into the bridged envelope; an inbound event with a matching podID is discarded before local fanout. Single-pod deployments without a bridge keep the pre-v0.6.3 zero-cost behaviour.

Wire format: standard SSE (text/event-stream). Each message is a JSON object { kind, payload } so the client can tell "problem opened" from "anomaly cleared" without one endpoint per event type.

Index

• func Handler(b *Broker) http.Handler
• type Bridge
• type Broker
  • func NewBroker() *Broker
  • func (b *Broker) Publish(kind string, payload any)
  • func (b *Broker) SetBridge(br Bridge)
  • func (b *Broker) StartBridge(ctx context.Context)
  • func (b *Broker) Subscribe(ch chan<- Event) func()
• type Event

Functions

func Handler

func Handler(b *Broker) http.Handler

Handler returns an http.Handler that streams events to the client over text/event-stream. Sends a comment heartbeat every 15s so intermediate proxies (NGINX, Cloudflare) don't time out the connection — many default to 60s idle.

Auth flow: the auth.Middleware in front of the mux already enforces JWT/cookie. By the time we hit Handler the user is authenticated; we don't need to re-check.

Types

type Bridge

type Bridge interface {
	Publish(ctx context.Context, channel string, msg []byte) error
	Subscribe(ctx context.Context, channel string) (<-chan []byte, error)
}

Bridge is the small slice of cache.Cache the SSE broker needs for cross-pod fan-out. Defined here as an interface so the sse package doesn't take a dependency on internal/cache (which already imports nothing from sse — keeps the layering one-way).

cache.Cache satisfies this naturally because it already implements Publish/Subscribe for the L1 invalidation channel.

type Broker

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

Broker is the in-process pub/sub bus. Producers (evaluator, anomaly detector) call Publish; consumers (HTTP handler) call Subscribe to get a channel that receives every future event until the request context cancels.

Channels are buffered (32) so a slow client doesn't block the producer. If the buffer fills (operator opens 100 tabs and pauses one) we drop events for that subscriber rather than stalling — the client's React Query polling will pick up the state on its next refetch.

func NewBroker

func NewBroker() *Broker

func (*Broker) Publish

func (b *Broker) Publish(kind string, payload any)

Publish fans the event out to local subscribers, and (when a bridge is attached) PUBLISHes it on Redis so peer pods can do the same. Non-blocking on both paths; a slow consumer drops the event rather than back-pressuring the producer (which would block the entire alert evaluator tick).

func (*Broker) SetBridge

func (b *Broker) SetBridge(br Bridge)

SetBridge attaches a cross-pod fan-out transport. Pass nil to disable (default — single-pod behaviour). Call before StartBridge; safe to call once at boot.

func (*Broker) StartBridge

func (b *Broker) StartBridge(ctx context.Context)

StartBridge spins the inbound Redis pub/sub goroutine. Inbound messages are decoded and locally fanned out, skipping our own pod's re-delivered events.

Outbound publishing is handled inside Publish itself (so the caller's single Publish() call still reaches every pod). The inbound side is the only thing that needs its own goroutine.

Subscribe failure is non-fatal: we log and stay single-pod. The local fanout path keeps working; only cross-pod delivery is degraded, and the operator sees the failure in the boot log.

func (*Broker) Subscribe

func (b *Broker) Subscribe(ch chan<- Event) func()

Subscribe registers a channel for events. Returns a function that removes the subscription — caller defers it.

type Event

type Event struct {
	Kind    string          `json:"kind"`
	Payload json.RawMessage `json:"payload,omitempty"`
}

Event is the wire envelope. Kind is short ("problem.open", "problem.resolve", "anomaly.open", "anomaly.clear") so the client switch is one comparison. Payload is opaque JSON the receiver decodes if it cares about the details (e.g. the problem's service for badge counts).