Documentation
¶
Overview ¶
Package events implements the in-process publish/subscribe broker behind dbbat's live event stream.
The broker is deliberately transport-agnostic: it knows about topics, sequence numbers and backpressure, and nothing about WebSockets, SSE, gin or HTTP. That separation is the mitigation for the transport trade-off recorded in docs/approvals.md — the stream currently ships over a WebSocket, but the envelope below would ride an SSE response body unchanged, so a later swap is confined to internal/api/stream.go.
Two invariants matter more than anything else here:
- Publishing is non-blocking and never returns an error. Events originate on the proxy hot path, and a slow or broken stream subscriber must never be able to stall — let alone break — a live database connection.
- Authorization is *not* the broker's job. Subscribers carry an authorizer supplied by the API layer, re-evaluated on every send, because sql_text routinely carries PII and the stream must never be a wider read path than GET /api/v1/queries. That re-check runs in the *transport's* write loop, never on the publish path: the API's authorizer does store round-trips, and calling it inline from Publish would let a slow database stall a proxy session holding a query.
Index ¶
- Constants
- func ConnectionQueriesTopic(connectionUID string) string
- func ConnectionUIDFromTopic(topic string) (string, bool)
- func ValidTopic(topic string) bool
- type Authorizer
- type Broker
- func (b *Broker) Publish(topic, eventType string, data map[string]any)
- func (b *Broker) PublishLocal(topic, eventType string, data map[string]any)
- func (b *Broker) SetForwarder(f Forwarder)
- func (b *Broker) Subscribe(authorize Authorizer, buffer int) *Subscriber
- func (b *Broker) SubscriberCount() int
- type Event
- type Forwarder
- type Subscriber
- func (s *Subscriber) Authorized(topic string) bool
- func (s *Subscriber) Close()
- func (s *Subscriber) Dropped() int64
- func (s *Subscriber) Events() <-chan Event
- func (s *Subscriber) Subscribe(topic string) bool
- func (s *Subscriber) TakePriority() []Event
- func (s *Subscriber) Topics() []string
- func (s *Subscriber) Unsubscribe(topic string)
Constants ¶
const ( // TopicApprovalsPending carries every approval-pending query across all // connections. Low volume by construction, and exempt from drop-on-overflow. TopicApprovalsPending = "approvals/pending" // TopicConnections carries connection open/close events. Admin only. TopicConnections = "connections" // TopicConnectionPrefix is the prefix of the per-connection query topic: // connection/<uid>/queries. TopicConnectionPrefix = "connection/" // TopicConnectionSuffix closes the per-connection query topic. TopicConnectionSuffix = "/queries" )
Topic names and prefixes. The namespace is the extension point; nothing else about the transport is query-specific.
const ( // EventQuery is a query observed on a connection. EventQuery = "query" // EventApprovalPending is a query parked awaiting a human decision. EventApprovalPending = "approval_pending" // EventApprovalResolved is the resolution of a hold — approved, denied or // abandoned — carrying who resolved it. EventApprovalResolved = "approval_resolved" // EventConnection is a connection open/close. EventConnection = "connection" )
Event types carried in the envelope's `type` field.
const DefaultBuffer = 256
DefaultBuffer is the per-subscriber send-buffer depth. Deep enough to ride out a GC pause or a slow browser tab, shallow enough that a genuinely stuck subscriber is detected in bounded memory.
const MaxTopicLength = 128
MaxTopicLength bounds a topic name. The longest legitimate one is connection/<uuid>/queries; anything materially longer is a client trying to make the server remember something.
Variables ¶
This section is empty.
Functions ¶
func ConnectionQueriesTopic ¶
ConnectionQueriesTopic builds the per-connection query topic name.
func ConnectionUIDFromTopic ¶
ConnectionUIDFromTopic extracts the connection uid from a connection/<uid>/queries topic, reporting whether the topic had that shape.
func ValidTopic ¶
ValidTopic reports whether a topic name is one this server can ever serve. It is a pure string check with no I/O, so the transport can reject junk before it reaches anything that allocates or hits the database.
Types ¶
type Authorizer ¶
Authorizer decides whether a subscriber may currently receive a topic. It is evaluated at subscribe time *and* again immediately before each send, so a user whose role or ownership changed mid-stream stops receiving at once rather than at the next reconnect.
It may be expensive (the API implementation reads the store), so the broker never calls it while publishing — see Authorized.
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker is the in-process fan-out hub.
func (*Broker) Publish ¶
Publish delivers an event to every local subscriber of the topic and hands it to the forwarder for other replicas. Never blocks, never errors: a nil broker is a no-op so proxy code can call it unconditionally.
func (*Broker) PublishLocal ¶
PublishLocal delivers an event to local subscribers only, without re-forwarding. This is the entry point for events arriving *from* another replica — forwarding them again would loop.
func (*Broker) SetForwarder ¶
SetForwarder installs (or clears, with nil) the cross-replica forwarder.
func (*Broker) Subscribe ¶
func (b *Broker) Subscribe(authorize Authorizer, buffer int) *Subscriber
Subscribe registers a new subscriber. The returned *Subscriber must be closed by the caller (typically with defer) to release broker resources.
func (*Broker) SubscriberCount ¶
SubscriberCount reports how many subscribers are attached. Test/telemetry helper.
type Event ¶
type Event struct {
Topic string `json:"topic"`
Seq uint64 `json:"seq"`
Type string `json:"type"`
Data map[string]any `json:"data"`
At time.Time `json:"at"`
}
Event is one published message. Seq is assigned by the broker at publish time and is monotonic per broker instance (not across replicas — clients use it to detect local gaps, not to order globally).
type Forwarder ¶
type Forwarder func(ev Event)
Forwarder is called for every locally-originated publish so the caller can fan the event out to other replicas (dbbat runs multiple pods; the session holding a query is on replica A while the admin's socket is on replica B). It must not block; the broker calls it inline on the publish path.
type Subscriber ¶
type Subscriber struct {
// contains filtered or unexported fields
}
Subscriber is one client's view of the broker: a topic set plus a bounded delivery channel.
func (*Subscriber) Authorized ¶
func (s *Subscriber) Authorized(topic string) bool
Authorized re-evaluates the subscriber's access to a topic. The transport calls it immediately before writing each event — that is the actual moment of send, and it is a goroutine the session does not wait on, so an expensive authorizer costs latency on one client's stream and nothing else.
func (*Subscriber) Close ¶
func (s *Subscriber) Close()
Close detaches the subscriber from the broker and closes its channel. Idempotent.
func (*Subscriber) Dropped ¶
func (s *Subscriber) Dropped() int64
Dropped returns and resets the number of events dropped due to overflow. The transport turns a non-zero value into {"type":"lagged","dropped":N} so the client knows to refetch from REST rather than assume continuity.
func (*Subscriber) Events ¶
func (s *Subscriber) Events() <-chan Event
Events is the delivery channel. It is closed when the subscriber is closed.
func (*Subscriber) Subscribe ¶
func (s *Subscriber) Subscribe(topic string) bool
Subscribe adds a topic after re-checking authorization. Returns false when the subscriber may not read that topic.
func (*Subscriber) TakePriority ¶
func (s *Subscriber) TakePriority() []Event
TakePriority drains any drop-exempt events that overflowed the main channel. The transport calls this alongside each channel read so pending-approval events are delivered even when the client is behind on everything else.
func (*Subscriber) Topics ¶
func (s *Subscriber) Topics() []string
Topics returns the currently subscribed topic names.
func (*Subscriber) Unsubscribe ¶
func (s *Subscriber) Unsubscribe(topic string)
Unsubscribe removes a topic.
Source Files