Documentation
¶
Overview ¶
Package kvevents coalesces persisted document events (KV changes and live document, comment, and presence events) and fans them out to every SSE subscriber in a server process. It also relays ephemeral events, such as cursor moves, that are never stored.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func IsProbePayload ¶
IsProbePayload reports whether a payload belongs to listener verification rather than a document event.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker owns one process-wide notification listener and one worker for each document that currently has subscribers.
func New ¶
func New(ctx context.Context, cfg Config, source Source, listenerFactory ListenerFactory) (*Broker, error)
New constructs a broker and verifies notification support before returning. In auto mode a failed verification selects polling; forced notify mode returns the verification error instead.
func (*Broker) ActiveMode ¶
ActiveMode returns the selected runtime mode.
func (*Broker) Close ¶
func (b *Broker) Close()
Close stops listener and document workers, then closes every subscriber. It is safe to call more than once.
func (*Broker) Relay ¶ added in v0.2.1
Relay sends an event that is not stored to the subscribers of its document. In notify mode the event goes through PostgreSQL to every server replica. In poll mode only the subscribers of this process get it.
func (*Broker) RequestedMode ¶
RequestedMode returns the configured mode before automatic detection.
type Config ¶
Config controls event delivery. PollInterval is primarily useful to callers that need a shorter interval in a controlled environment; zero uses 2s.
type Listener ¶
type Listener interface {
Wait(ctx context.Context) (string, error)
Close(ctx context.Context) error
}
Listener waits for payloads published on the notification channel: the ID of a document with new stored events, or a relayed event.
type ListenerFactory ¶
type ListenerFactory interface {
Open(ctx context.Context) (Listener, error)
Notify(ctx context.Context, payload string) error
}
ListenerFactory opens a verified notification connection and sends notifications to every listener.
type Mode ¶
type Mode string
Mode selects how a Broker learns that new persisted events are available.
const ( // ModeAuto uses PostgreSQL notifications when they can be verified and // otherwise falls back to coalesced polling. ModeAuto Mode = "auto" // ModeNotify requires PostgreSQL LISTEN/NOTIFY support. ModeNotify Mode = "notify" // ModePoll uses one poll loop per document with active subscribers. ModePoll Mode = "poll" )
type NotificationSender ¶
type NotificationSender interface {
Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
}
NotificationSender emits the verification probe through the application's normal database pool, independently of the dedicated LISTEN connection.
type PostgresListenerFactory ¶
type PostgresListenerFactory struct {
// contains filtered or unexported fields
}
PostgresListenerFactory opens verified PostgreSQL LISTEN connections.
func NewPostgresListenerFactory ¶
func NewPostgresListenerFactory(databaseURL string, sender NotificationSender) *PostgresListenerFactory
NewPostgresListenerFactory returns a factory that owns a dedicated pgx connection outside the application's pool. databaseURL must reach the same database directly or through a session-pooling proxy.
type Source ¶
type Source interface {
CurrentDocumentKVRevision(ctx context.Context, documentID string) (int64, error)
ListDocumentKVEvents(ctx context.Context, documentID string, afterRevision int64, limit int32) ([]repo.DocumentKVEvent, error)
}
Source is the narrow persistence surface the event broker needs.
type Subscriber ¶
type Subscriber interface {
Subscribe(ctx context.Context, documentID string) (<-chan repo.DocumentKVEvent, func(), error)
Relay(ctx context.Context, event repo.DocumentKVEvent) error
}
Subscriber is the event-stream surface consumed by the HTTP server.
Source Files