Documentation
¶
Overview ¶
Package events implements the in-process pub/sub hub behind the per-resource SSE subscription endpoints (GET /api/{resource}/events, S45).
ADR-015: single-node by design. The engine runs as one process today, so the hub is an in-memory map with no external broker; when HA/multi-node ships, a broker (or Postgres LISTEN/NOTIFY) replaces the fan-out — the Publish call sites stay the same. Delivery is at-most-once with no replay/Last-Event-ID: a subscriber that connects after an event, or that is closed as a slow consumer, misses it (documented limitation; replay is future work).
Index ¶
Constants ¶
const DefaultMaxPerTenant = 1000
DefaultMaxPerTenant bounds concurrent subscribers per tenant when no explicit limit is configured (APPXIMO_MAX_SSE_PER_TENANT).
Variables ¶
var ErrTenantLimit = errors.New("tenant SSE subscriber limit reached")
ErrTenantLimit is returned by Subscribe when the tenant has reached its concurrent-subscriber cap. The HTTP layer maps it to 429.
Functions ¶
This section is empty.
Types ¶
type Event ¶
type Event struct {
Type string // "create" | "update" | "delete"
Resource string
ID string
Record map[string]any // shared across subscribers — receivers must not mutate
}
Event is one resource change, broadcast to every subscriber of (tenant, resource). Record is the post-write row (RETURNING *) for create/update, and nil for delete (the row is gone; only the id is known — the DELETE path does not pay for a RETURNING just in case someone listens).
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub is the process-wide registry of subscribers, keyed tenant → resource → subscriber set. A nil *Hub is valid and inert: Publish/Count no-op and Subscribe fails, so tests and callers that don't wire SSE pass nil.
func (*Hub) Publish ¶
Publish broadcasts ev to every subscriber of (tenant, ev.Resource). It NEVER blocks the caller (the request path): the per-subscriber send is a non-blocking select, and a subscriber whose buffer is full is marked slow (its Slow channel is closed once) so its connection terminates. With zero subscribers the cost is one RLock + two map lookups.
func (*Hub) Subscribe ¶
func (h *Hub) Subscribe(tenant, resource string, allowedFields []string, condField string, condValue any) (*Subscriber, error)
Subscribe registers a new subscriber for (tenant, resource) and returns it. Returns ErrTenantLimit when the tenant is at its concurrent cap.
func (*Hub) Unsubscribe ¶
func (h *Hub) Unsubscribe(sub *Subscriber)
Unsubscribe removes sub from the hub. Idempotent; safe on a sub that was never registered. Every connection MUST call this on exit (defer) — it is what prevents subscriber/goroutine leaks.
type Subscriber ¶
type Subscriber struct {
// C delivers events. Never closed (avoids send-on-closed races); the
// handler exits via Slow or its request context instead.
C chan Event
// Slow is closed (once) when the subscriber's buffer overflows. The handler
// must send a final error event and terminate the connection.
Slow chan struct{}
// AllowedFields is the RBAC field allowlist captured at subscribe time
// (empty = all fields). The SSE writer applies the same FilterFields the
// GET handlers use, so a role never sees a field in an event that it
// cannot see in a GET.
AllowedFields []string
// CondField/CondValue carry the row-level RBAC condition captured at
// subscribe time ("" = none). Only eq is supported — the only operator the
// policy engine emits today; events for non-matching rows are not
// delivered (same row scoping as the list endpoint).
CondField string
CondValue any
// contains filtered or unexported fields
}
Subscriber is one open SSE connection's mailbox. The HTTP handler owns the receive side: it selects on C (events), Slow (closed by the hub when this subscriber's buffer overflows), and the request context.
Source Files