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.
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) 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 document IDs published on the notification channel.
type ListenerFactory ¶
ListenerFactory opens a verified notification connection.
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)
}
Subscriber is the event-stream surface consumed by the HTTP server.
Source Files