realtime

import "github.com/nathabonfim59/pbvex/backend/internal/realtime"

Package realtime implements the PBVex SSE realtime endpoint.

Index

• Variables
• func DeriveSubscriptionID(version, path string, args any) string
• func ProtocolError(e *core.RequestEvent, status int, code deploy.ErrorCode, message string, cause error) error
• func RequestID(e *core.RequestEvent) string
• type Broadcaster
  • func NewBroadcaster(service *deploy.Service, config Config) *Broadcaster
  • func (b *Broadcaster) Handle(e *core.RequestEvent) error
  • func (b *Broadcaster) InvalidateAll()
  • func (b *Broadcaster) ReconnectAll()
• type Config
  • func DefaultConfig() Config
• type Subscription

Variables

var (
    ErrArgsNotJSON      = errors.New("args is not valid JSON")
    ErrArgsNotCanonical = errors.New("args is not canonical JSON")
    ErrArgsInvalid      = errors.New("invalid args")
)

func DeriveSubscriptionID

func DeriveSubscriptionID(version, path string, args any) string

DeriveSubscriptionID is a test/export helper that canonicalizes args before deriving the subscription ID.

func ProtocolError

func ProtocolError(e *core.RequestEvent, status int, code deploy.ErrorCode, message string, cause error) error

ProtocolError writes a structured PBVex error response.

func RequestID

func RequestID(e *core.RequestEvent) string

RequestID returns a bounded request ID, preferring the X-Request-Id header and falling back to a generated UUID. It strips or rejects unsafe values.

type Broadcaster

Broadcaster manages active realtime subscriptions and broadcasts invalidation events to them using a bounded, coalescing design.

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

func NewBroadcaster

func NewBroadcaster(service *deploy.Service, config Config) *Broadcaster

NewBroadcaster creates a new realtime broadcaster.

func (*Broadcaster) Handle

func (b *Broadcaster) Handle(e *core.RequestEvent) error

Handle is the POST /api/pbvex/realtime handler (GET is retained as a strictly bounded compatibility fallback).

func (*Broadcaster) InvalidateAll

func (b *Broadcaster) InvalidateAll()

InvalidateAll notifies every active subscription to re-run its query. It is non-blocking and coalesces: slow subscriptions receive a single pending notification that will be processed after the current run.

func (*Broadcaster) ReconnectAll

func (b *Broadcaster) ReconnectAll()

ReconnectAll closes all active subscription connections so that clients reconnect and re-negotiate event-size limits with the newly active deployment. Called on activation/rollback where the pinned deployment snapshot (and its maxReturnValueBytes) may differ from the new one. The generation is incremented under the write lock so that subscriptions admitted before this call but not yet registered are rejected by subscribeWithFence.

type Config

Config controls realtime behavior.

type Config struct {
    PingInterval         time.Duration
    MaxConnections       int
    MaxConnectionsPerIP  int
    MaxConcurrentQueries int
    MaxBodyBytes         int64
    MaxGETArgsBytes      int64
}

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns the default realtime configuration.

type Subscription

Subscription is a single realtime SSE subscription.

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

Generated by gomarkdoc