websocket

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 30, 2026 License: MIT Imports: 47 Imported by: 0

Documentation

Overview

Package websocket provides typed channels, room publication and the versioned Foundry WebSocket protocol. Applications own domain handlers and DTOs.

Index

Constants

View Source
const KeyFeature keyspace.Feature = "websocket"

KeyFeature is the physical address family owned by this adapter.

View Source
const MaxChannelEvents = 256
View Source
const MaxChannels = 256
View Source
const MaxReplayMessages = 1024
View Source
const MaxRoomBytes = 512
View Source
const MaxTicketRedeemers = auth.MaxCredentials

MaxTicketRedeemers bounds the guards one hub accepts tickets for.

View Source
const ProtocolVersion = 1
View Source
const Subprotocol = "foundry.v1"
View Source
const TicketSubprotocolPrefix = "foundry.ticket."

TicketSubprotocolPrefix marks a single-use handshake ticket offered as one extra Sec-WebSocket-Protocol entry next to Subprotocol. The hub never selects or echoes it; only Subprotocol is negotiated.

Variables

View Source
var MembershipConflict = safeerror.Sentinel(fault.Conflict, "WebSocket connection holds a different membership for this scope")

MembershipConflict reports that the authority still holds a different membership of this connection for the scope, typically left behind by a leave that failed transiently. The hub releases that record and joins again; it is a per-operation failure, never a namespace policy conflict.

View Source
var PolicyConflict = safeerror.Sentinel(fault.Conflict, "live WebSocket cluster policy differs")

PolicyConflict reports live cluster state written under a different channel registry or authority limits in the same namespace. It is the unrecoverable configuration fault that terminates a running hub; adapters return it (or wrap it) for such state. Other operation failures are treated as transient.

Functions

func DeclaredModule

func DeclaredModule(name foundation.ProviderID, key foundation.Key[*Hub], config ServerConfig, requires []foundation.ProviderID, construct func(foundation.Resolver) (*Hub, []foundryhttp.Middleware, error)) foundation.Module

DeclaredModule allows domain declarations to provide upgrade middleware during pure construction while preserving the same single-server kernel ownership.

func DisconnectConnection

func DisconnectConnection(ctx context.Context, source PublisherSource, id ConnectionID) error

DisconnectConnection is trusted server-side management. Distributed delivery uses live pub/sub; an error never proves that no instance acted on the command.

func DisconnectSubject

func DisconnectSubject[M model.Identifiable, K any](ctx context.Context, source PublisherSource, guard auth.Guard[M], reference model.Reference[M, K]) error

DisconnectSubject preserves the model/key type plus guard/provider namespace. Supply the generated reference containing the stored key, not a presentation getter. Call after committed revocation; periodic fresh authorization remains the backstop if a live disconnect message is lost.

func Module

func Module(name foundation.ProviderID, key foundation.Key[*Hub], config ServerConfig, requires []foundation.ProviderID, construct func(foundation.Resolver) (*Hub, error)) foundation.Module

Module registers the WebSocket kernel and an injectable Hub. HTTP middleware wraps the exact upgrade route; trusted proxy/public-origin policy can therefore run before origin validation. List providers owning borrowed services in requires.

func NotPublished

func NotPublished(err error) bool

NotPublished reports that a Publish or Broadcast error happened before any live publication, fan-out or replay history was attempted (validation, local admission, a closing hub or payload encoding), so retrying cannot produce a duplicate. Other errors may follow partial distributed publication.

func Route

func Route(hub *Hub, path string, middleware ...foundryhttp.Middleware) foundryhttp.RouteRegistration

Route serves this Hub's upgrades on an application router at an exact static path. Middleware wraps only the upgrade; router-wide middleware also applies.

func SharedModule

func SharedModule(name foundation.ProviderID, key foundation.Key[*Hub], server foundation.Key[*foundryhttp.Server], requires []foundation.ProviderID, construct func(foundation.Resolver) (*Hub, error)) foundation.Module

SharedModule constructs a Hub served by an existing HTTP kernel instead of a dedicated listener: mount Route on that kernel's router; Ready reports that server's address. Boot confirms the distributed subscription before the listener starts; the hub drains its sockets when the application lifetime ends (after any stop delay), inside the HTTP shutdown grace. Upgrade sockets hold HTTP request and connection capacity for their lifetime, so size those limits for realtime load.

func ValidateClusterConnection

func ValidateClusterConnection(ctx context.Context, key ClusterKey, instance InstanceID, connection ConnectionID) error

func ValidateClusterOperation

func ValidateClusterOperation(ctx context.Context, key ClusterKey) error

Types

type Action

type Action string
const (
	Subscribe   Action = "subscribe"
	Unsubscribe Action = "unsubscribe"
	Message     Action = "message"
)

type Anonymous

type Anonymous struct{}

type BackendConnection

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

BackendConnection owns immutable cluster selection, borrowing its authority adapter. It registers no kernel or socket. A local connection is explicitly process-local.

func NewClusterConnection

func NewClusterConnection(backend ClusterBackend, cluster ClusterConfig) (*BackendConnection, error)

func NewLocalConnection

func NewLocalConnection() *BackendConnection

func (*BackendConnection) Local

func (c *BackendConnection) Local() bool

Local reports a process-local connection that cannot publish across processes.

func (*BackendConnection) NewHub

func (c *BackendConnection) NewHub(registry *Registry, authentication *foundryhttp.Authentication, config Config, opts ...Option) (*Hub, error)

func (*BackendConnection) NewPublisher

func (c *BackendConnection) NewPublisher(registry *Registry, config Config, opts ...Option) (*Publisher, error)

type Channel

type Channel[C, R, S any] struct {
	// contains filtered or unexported fields
}

Channel binds a nominal owner, room key and concrete authenticated subject. Use a distinct owner type per channel. Runtime declaration identity also rejects separately declared channels sharing the same owner or string name.

func OwnedRooms

func OwnedRooms[C any, R comparable, S model.Identifiable](id ChannelID, rooms Rooms[R], guard auth.Guard[S], reference model.Reference[S, R]) Channel[C, R, S]

OwnedRooms permits only a concrete room equal to the authenticated model's stored key. Reuse the generated zero model's reference metadata; presentation getters never decide ownership. Channel-wide subscription is forbidden.

func Private

func Private[C, R any, S model.Identifiable](id ChannelID, rooms Rooms[R], guard auth.Guard[S], authorize func(context.Context, S, Target[R]) error) Channel[C, R, S]

Private always requires explicit room/channel authorization after the fresh typed guard check. Matching a guard alone does not authorize arbitrary rooms.

func Public

func Public[C, R any](id ChannelID, rooms Rooms[R]) Channel[C, R, Anonymous]

func (Channel[C, R, S]) ID

func (c Channel[C, R, S]) ID() ChannelID

func (Channel[C, R, S]) Validate

func (c Channel[C, R, S]) Validate() error

func (Channel[C, R, S]) WithHooks

func (c Channel[C, R, S]) WithHooks(hooks Hooks[R, S]) Channel[C, R, S]

func (Channel[C, R, S]) WithReplay

func (c Channel[C, R, S]) WithReplay(config ReplayConfig) Channel[C, R, S]

type ChannelID

type ChannelID string

type ChannelInfo

type ChannelInfo struct {
	ID         ChannelID                 `json:"id"`
	Room       foundryhttp.URLScalarInfo `json:"room"`
	Private    bool                      `json:"private"`
	OwnedRooms bool                      `json:"owned_rooms"`
	Guard      auth.GuardName            `json:"guard,omitempty"`
	Provider   auth.ProviderName         `json:"provider,omitempty"`
	Presence   *contract.Schema          `json:"presence,omitempty"`
	Replay     ReplayConfig              `json:"replay,omitzero"`
	Events     []EventInfo               `json:"events"`
}

type ChannelSnapshot

type ChannelSnapshot struct {
	ID                                                                      ChannelID
	Subscriptions                                                           int
	Incoming, Accepted, Completed, Failures, Published, Delivered, Replayed uint64
}

type ClientDescription

type ClientDescription struct {
	Protocol ProtocolInfo  `json:"protocol"`
	Limits   ClientLimits  `json:"limits"`
	Channels []ChannelInfo `json:"channels"`
}

func DescribeClient

func DescribeClient(registry *Registry, config Config) (ClientDescription, error)

DescribeClient uses the same registry/config supplied to New or NewDistributed. It starts no hub, opens no sockets and invokes no handlers.

type ClientLimits

type ClientLimits struct {
	Subscriptions         int   `json:"subscriptions"`
	FrameBytes            int   `json:"frame_bytes"`
	PresenceMembers       int   `json:"presence_members"`
	MemberBytes           int   `json:"member_bytes"`
	DeduplicationEntries  int   `json:"deduplication_entries"`
	OperationMilliseconds int64 `json:"operation_ms"`
	// InboundQueue bounds a client's in-flight operations; MessageRate is the
	// per-connection frame budget. Generated clients respect both.
	InboundQueue int                 `json:"inbound_queue"`
	MessageRate  ClientRate          `json:"message_rate"`
	Payload      contract.JSONLimits `json:"payload"`
}

ClientLimits exports the actual configured public budgets. Time is expressed in milliseconds rounded up; server policy/credential material remains private.

func (ClientLimits) Validate

func (limits ClientLimits) Validate() error

Validate reuses runtime configuration validation with the smallest private capacity scope. Public metadata cannot infer a server's private connection counts, origins or credential refresh policy.

type ClientRate

type ClientRate struct {
	Requests           uint32 `json:"requests"`
	WindowMilliseconds int64  `json:"window_ms"`
}

ClientRate is a frame budget of Requests per WindowMilliseconds.

type ClusterBackend

ClusterBackend borrows the existing live pub/sub transport and adds bounded atomic lease/membership/history capabilities. Mutations are attempted once; errors may represent an unknown outcome. Expired connections never resurrect. Members excludes expired leases and counts each subject's distinct connections. Close/Leave are idempotent and never delete another instance's live ownership.

type ClusterConfig

type ClusterConfig struct {
	Namespace                keyspace.Namespace
	MaxConnections           int
	MaxConnectionsPerSubject int
	ConnectionTTL            time.Duration
	OperationTimeout         time.Duration
	Buffer                   pubsub.Limits
	// RetainConnectionsOnGap keeps local sockets open across a lost fan-out
	// stream; events published during the gap are then missed silently. The
	// default closes them with a retryable status so clients reconnect and
	// request replay.
	RetainConnectionsOnGap bool
	// ExcludeRemoteConnections lets ExceptConnection name a connection hosted
	// by another instance by carrying the exclusion in the fan-out envelope.
	// Instances before this field reject such envelopes, so enable it only
	// once every instance in the namespace runs a release that has it.
	// Excluding a connection of this instance (RelayToOthers) never needs it.
	ExcludeRemoteConnections bool
}

ClusterConfig distinguishes authority-wide quotas from Config's local limits. Transient authority failures fail only the affected operation. A lost fan-out stream is resubscribed with backoff; no mutation retry is inferred from the Redis client's ability to reconnect. Only a policy conflict terminates a Hub.

func DefaultClusterConfig

func DefaultClusterConfig(namespace keyspace.Namespace) ClusterConfig

func (ClusterConfig) Validate

func (c ClusterConfig) Validate() error

type ClusterKey

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

func NewClusterKey

func NewClusterKey(namespace keyspace.Namespace, policy string, limits ClusterLimits) (ClusterKey, error)

func (ClusterKey) Limits

func (k ClusterKey) Limits() ClusterLimits

func (ClusterKey) Namespace

func (k ClusterKey) Namespace() keyspace.Namespace

func (ClusterKey) Policy

func (k ClusterKey) Policy() string

func (ClusterKey) String

func (k ClusterKey) String() string

func (ClusterKey) Validate

func (k ClusterKey) Validate() error

type ClusterLimits

type ClusterLimits struct {
	Connections           int           `json:"connections"`
	ConnectionsPerSubject int           `json:"connections_per_subject"`
	Subscriptions         int           `json:"subscriptions"`
	PresenceMembers       int           `json:"presence_members"`
	MemberBytes           int           `json:"member_bytes"`
	FrameBytes            int           `json:"frame_bytes"`
	ConnectionTTL         time.Duration `json:"connection_ttl"`
	Retention             time.Duration `json:"retention"`
}

ClusterLimits are immutable authority policy, shared by every process in one namespace. Metadata lives as long as connections or replay can remain live.

func (ClusterLimits) Validate

func (l ClusterLimits) Validate() error

type ClusterMembership

type ClusterMembership struct {
	Scope    Scope
	Subject  MemberID
	Presence bool
	Data     json.RawMessage
}

ClusterMembership is an adapter boundary. Data contains only the channel's safe member DTO. Subject is an opaque guard/provider/stored-identity digest.

func (ClusterMembership) Validate

func (m ClusterMembership) Validate(limits ClusterLimits) error

type Code

type Code string
const (
	Malformed          Code = "malformed"
	UnsupportedVersion Code = "unsupported_version"
	UnknownChannel     Code = "unknown_channel"
	UnknownEvent       Code = "unknown_event"
	WrongDirection     Code = "wrong_direction"
	Unauthenticated    Code = "unauthenticated"
	Forbidden          Code = "forbidden"
	NotSubscribed      Code = "not_subscribed"
	AlreadySubscribed  Code = "already_subscribed"
	InvalidPayload     Code = "invalid_payload"
	CapacityExceeded   Code = "capacity_exceeded"
	OperationFailed    Code = "operation_failed"
	OperationTimedOut  Code = "operation_timed_out"
	Stopping           Code = "stopping"
	RateLimited        Code = "rate_limited"
	// Unavailable is a retryable transient failure, such as a cluster
	// authority timeout or exhausted management capacity.
	Unavailable Code = "unavailable"
)

func (Code) Error

func (c Code) Error() string

type Config

type Config struct {
	MaxPresenceScopes int
	MaxConnections    int
	MaxSubscriptions  int
	MaxFrameBytes     int
	// InboundQueue bounds frames read but not yet processed. Generated clients
	// limit their in-flight operations to it; overflowing it closes the socket.
	InboundQueue int
	// OutboundQueue bounds queued frames per connection. MaxQueuedBytes bounds
	// their actual bytes, including inbound frames and pending-admission
	// buffers; MaxTotalQueuedBytes bounds all connections together. Exceeding
	// either disconnects the slow connection instead of blocking publishers.
	OutboundQueue       int
	MaxQueuedBytes      int
	MaxTotalQueuedBytes int64
	// MaxOperations bounds concurrent trusted publication, presence and
	// disconnect operations. Callers wait briefly, then receive fault.Overloaded.
	MaxOperations            int
	MaxPresenceMembers       int
	MaxMemberBytes           int
	OperationTimeout         time.Duration
	WriteTimeout             time.Duration
	HeartbeatInterval        time.Duration
	PongTimeout              time.Duration
	AuthRefreshInterval      time.Duration
	DrainTimeout             time.Duration
	MaxConnectionsPerIP      int
	MaxConnectionsPerSubject int
	DeduplicationEntries     int
	MessageRate              ratelimit.Limit
	Payload                  contract.JSONLimits
	// AdditionalOrigins are exact HTTP(S) origins, never wildcard patterns.
	// Same-origin is accepted after trusted-proxy handling. Originless clients
	// must be explicitly allowed; serialized opaque/null origins never are.
	AdditionalOrigins []foundryhttp.Origin `config:",json"`
	AllowOriginless   bool
}

func DefaultConfig

func DefaultConfig() Config

func (Config) Validate

func (c Config) Validate() error

type Connection

type Connection struct{}

type ConnectionID

type ConnectionID = model.ID[Connection]

type ConnectionName

type ConnectionName string

ConnectionName selects an instance, distinct from names of other service families.

func (ConnectionName) Validate

func (n ConnectionName) Validate() error

type Connections

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

Connections is immutable; default and named selection alias the same instance.

func NewConnections

func NewConnections(selected ConnectionName, entries ...NamedConnection) (*Connections, error)

func (*Connections) Connection

func (r *Connections) Connection(name ConnectionName) (*BackendConnection, error)

func (*Connections) Default

func (r *Connections) Default() (*BackendConnection, error)

func (*Connections) DefaultName

func (r *Connections) DefaultName() ConnectionName

func (*Connections) Names

func (r *Connections) Names() []ConnectionName

type Diagnostics

type Diagnostics struct {
	Runtime                                                         Snapshot
	QueuedFrames, QueuedBytes                                       int
	HeartbeatFailures, RateRejected, Revocations, ForcedDisconnects uint64
	// CleanupFailures counts cluster membership releases or stream closes that
	// failed; leases expire by TTL and presence is reconciled.
	CleanupFailures uint64
	Channels        []ChannelSnapshot
}

func Diagnose

func Diagnose[M any](ctx context.Context, hub *Hub, guard auth.Guard[M], authorize func(context.Context, M) error) (Diagnostics, error)

Diagnose requires both a typed authenticated guard and explicit management policy. No public HTTP endpoint is installed. Compose with the existing HTTP authentication adapter; its scope determines the administrative subject.

type Direction

type Direction string
const (
	ClientToServer Direction = "client_to_server"
	ServerToClient Direction = "server_to_client"
)

type EventID

type EventID string

type EventInfo

type EventInfo struct {
	ID                      EventID         `json:"id"`
	Direction               Direction       `json:"direction"`
	Dynamic                 bool            `json:"dynamic"`
	AcceptedAcknowledgement bool            `json:"accepted_acknowledgement"`
	Payload                 contract.Schema `json:"payload,omitzero"`
}

type EventRegistration

type EventRegistration[C any] struct {
	// contains filtered or unexported fields
}

type Hooks

type Hooks[R, S any] struct {
	Joined func(context.Context, MessageContext[R, S]) error
	Left   func(context.Context, LeaveContext[R]) error
}

type Hub

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

Hub owns connections, bounded queues and active callbacks. Construction starts no I/O or goroutines. Keep borrowed authentication/services alive until Done. A Hub must not be copied. Stop is permanent; construct a new Hub to restart.

func New

func New(registry *Registry, authentication *foundryhttp.Authentication, config Config, opts ...Option) (*Hub, error)

func NewDistributed

func NewDistributed(registry *Registry, authentication *foundryhttp.Authentication, config Config, backend ClusterBackend, cluster ClusterConfig, opts ...Option) (*Hub, error)

func (*Hub) Done

func (h *Hub) Done() <-chan struct{}

func (*Hub) Probe

func (h *Hub) Probe(ctx context.Context) error

Probe reports local serving health for readiness checks without I/O: nil while the hub admits connections and, when distributed, its fan-out stream is subscribed. It fails while stopping, after a terminal fault or during a stream gap/resubscription. Transient per-operation failures do not fail it.

func (*Hub) Ready

func (h *Hub) Ready(ctx context.Context) (string, error)

Ready waits for this Hub's Module-selected listener and returns its bound address. Manually embedded HTTP handlers use their host's readiness API.

func (*Hub) Registry

func (h *Hub) Registry() *Registry

func (*Hub) ServeHTTP

func (h *Hub) ServeHTTP(w stdhttp.ResponseWriter, r *stdhttp.Request)

ServeHTTP remains active until all socket loops and callbacks actually exit. Native HTTP middleware can wrap this handler. Route only the intended upgrade path to it. A manual host must call Stop; Module owns both listener and Hub.

func (*Hub) Snapshot

func (h *Hub) Snapshot() Snapshot

Snapshot is local operational metadata. It contains no credentials or models.

func (*Hub) Start

func (h *Hub) Start(ctx context.Context) error

Start confirms the distributed subscription before upgrades may be accepted. A local Hub needs no separate startup. The wait context bounds waiting; Stop owns any setup still in progress. A failed initial subscription is terminal; later stream loss is recovered by resubscribing without stopping the hub.

func (*Hub) Stop

func (h *Hub) Stop(ctx context.Context) error

Stop seals admission, cancels sockets and waits for actual handler/hook exit. Its context bounds waiting, not callback ownership. Calling it from one of this Hub's active callbacks is rejected instead of waiting on itself. It returns the unrecoverable cluster fault that terminated the hub, if any.

type Incoming

type Incoming[C, R, S, P any] struct {
	// contains filtered or unexported fields
}

func DefineIncoming

func DefineIncoming[C, R, S, P any](channel Channel[C, R, S], id EventID, payload contract.JSON[P]) Incoming[C, R, S, P]

func RawIncoming

func RawIncoming[C, R, S any](channel Channel[C, R, S], id EventID) Incoming[C, R, S, json.RawMessage]

RawIncoming/RawOutgoing deliberately expose dynamic JSON. Metadata marks this escape hatch; it is excluded from claims of typed payload schemas.

func (Incoming[C, R, S, P]) AcknowledgeAccepted

func (e Incoming[C, R, S, P]) AcknowledgeAccepted() Incoming[C, R, S, P]

AcknowledgeAccepted emits an accepted frame after decoding and authorization, before invoking the handler. Completion still emits ack or a stable error.

func (Incoming[C, R, S, P]) Authorize

func (e Incoming[C, R, S, P]) Authorize(check func(context.Context, MessageContext[R, S], P) error) Incoming[C, R, S, P]

Authorize adds a payload-aware policy after channel authorization and decoding. Return an authorization error on denial. A nil policy invalidates the binding.

func (Incoming[C, R, S, P]) Handle

func (e Incoming[C, R, S, P]) Handle(handler func(context.Context, MessageContext[R, S], P) error) EventRegistration[C]

func (Incoming[C, R, S, P]) Relay

func (e Incoming[C, R, S, P]) Relay(outgoing Outgoing[C, P]) EventRegistration[C]

Relay is an ordinary typed handler; private room and payload policies still run before the corresponding outgoing event is published.

func (Incoming[C, R, S, P]) RelayToOthers

func (e Incoming[C, R, S, P]) RelayToOthers(outgoing Outgoing[C, P]) EventRegistration[C]

RelayToOthers is Relay without live delivery back to the sending connection.

type Instance

type Instance struct{}

type InstanceID

type InstanceID = model.ID[Instance]

type LeaveContext

type LeaveContext[R any] struct {
	Publisher  *Hub
	Connection ConnectionID
	Target     Target[R]
	Subject    SubjectReference
}

LeaveContext deliberately contains no retained model or authentication scope.

type Member

type Member[D any] struct {
	ID          MemberID
	Data        D
	Connections int
}

type MemberFrame

type MemberFrame struct {
	ID          MemberID        `json:"id"`
	Data        json.RawMessage `json:"data"`
	Connections int             `json:"connections"`
}

type MemberID

type MemberID string

type MessageContext

type MessageContext[R, S any] struct {
	Publisher  *Hub
	Connection ConnectionID
	Target     Target[R]
	Subject    S
}

type MessageID

type MessageID = model.ID[Publication]

func Broadcast

func Broadcast[C, R, S, P any](ctx context.Context, source PublisherSource, channel Channel[C, R, S], event Outgoing[C, P], payload P, options ...PublishOption) (MessageID, error)

Broadcast reaches every connection subscribed anywhere in this channel, once per connection even when it has overlapping whole-channel/room subscriptions.

func Publish

func Publish[C, R, S, P any](ctx context.Context, source PublisherSource, channel Channel[C, R, S], room R, event Outgoing[C, P], payload P, options ...PublishOption) (MessageID, error)

Publish reaches subscriptions for exactly this room. Whole-channel subscribers do not implicitly subscribe to every private room. Success means local queue admission or distributed live publication was attempted, not recipient delivery. Slow peers disconnect. Distributed errors may follow partial publication/history.

type NamedConnection

type NamedConnection struct {
	Name  ConnectionName
	Value *BackendConnection
}

NamedConnection borrows one existing instance. Selection never creates another owner.

type Option

type Option func(*options) error

Option configures optional hub collaborators at construction.

func WithLogger

func WithLogger(logger *slog.Logger) Option

WithLogger reports cluster degradation, stream loss/recovery and terminal faults with safe structured diagnostics. Without it the hub does not log.

func WithTickets

func WithTickets(redeemers ...TicketRedeemer) Option

WithTickets lets browsers, which cannot send an Authorization header on a WebSocket, authenticate with a ticket from TicketSubprotocolPrefix. The hub redeems it after the origin check, trying redeemers in order; the first that knows it authenticates that guard's channels for the connection. Every later auth scope re-verifies the bound credential, so revocation still disconnects.

type Outgoing

type Outgoing[C, P any] struct {
	// contains filtered or unexported fields
}

func DefineOutgoing

func DefineOutgoing[C, R, S, P any](channel Channel[C, R, S], id EventID, payload contract.JSON[P]) Outgoing[C, P]

func RawOutgoing

func RawOutgoing[C, R, S any](channel Channel[C, R, S], id EventID) Outgoing[C, json.RawMessage]

func (Outgoing[C, P]) Registration

func (e Outgoing[C, P]) Registration() EventRegistration[C]

type PresenceChannel

type PresenceChannel[C, R, S, D any] struct {
	// contains filtered or unexported fields
}

func WithPresence

func WithPresence[C, R any, S model.Identifiable, D any](channel Channel[C, R, S], payload contract.JSON[D], member func(context.Context, S) (D, error)) PresenceChannel[C, R, S, D]

WithPresence requires an explicit safe DTO descriptor and authenticated model. contract.JSON rejects persistence models. Only the DTO bytes and immutable subject reference survive the operation; full models are never stored.

func (PresenceChannel[C, R, S, D]) Channel

func (p PresenceChannel[C, R, S, D]) Channel() Channel[C, R, S]

func (PresenceChannel[C, R, S, D]) Count

func (p PresenceChannel[C, R, S, D]) Count(ctx context.Context, hub *Hub, target Target[R]) (int, error)

Count uses the same authoritative safe-member snapshot as Members. It counts subjects, not tabs, and reports errors instead of a fallback local cluster count.

func (PresenceChannel[C, R, S, D]) Members

func (p PresenceChannel[C, R, S, D]) Members(ctx context.Context, hub *Hub, target Target[R]) ([]Member[D], error)

Members is a trusted server-side inspection API, not an authorization check. Local results describe this Hub only; distributed results use its authority.

type PresenceSnapshot

type PresenceSnapshot struct {
	Revision uint64        `json:"revision"`
	Members  []MemberFrame `json:"members"`
}

PresenceSnapshot carries authority order so concurrent refreshes cannot replace newer counts with an older result. Revisions are bounded positive Redis integers.

type ProtocolActions

type ProtocolActions struct {
	Subscribe   Action `json:"subscribe"`
	Unsubscribe Action `json:"unsubscribe"`
	Message     Action `json:"message"`
}

type ProtocolInfo

type ProtocolInfo struct {
	Version           int               `json:"version"`
	Subprotocol       string            `json:"subprotocol"`
	TicketPrefix      string            `json:"ticket_subprotocol_prefix"`
	MaxRoomBytes      int               `json:"max_room_bytes"`
	MaxReplayMessages int               `json:"max_replay_messages"`
	Actions           ProtocolActions   `json:"actions"`
	Responses         ProtocolResponses `json:"responses"`
	Codes             []Code            `json:"codes"`
}

ProtocolInfo is the versioned transport vocabulary used by runtime and client exporters. It contains no live connection state or credentials.

func ProtocolDescription

func ProtocolDescription() ProtocolInfo

type ProtocolResponses

type ProtocolResponses struct {
	Subscribed      ResponseType `json:"subscribed"`
	Unsubscribed    ResponseType `json:"unsubscribed"`
	Acknowledged    ResponseType `json:"acknowledged"`
	Accepted        ResponseType `json:"accepted"`
	Error           ResponseType `json:"error"`
	Event           ResponseType `json:"event"`
	PresenceJoined  ResponseType `json:"presence_joined"`
	PresenceLeft    ResponseType `json:"presence_left"`
	PresenceUpdated ResponseType `json:"presence_updated"`
}

type Publication

type Publication struct{}

type PublishOption

type PublishOption func(*publishOptions)

PublishOption adjusts one publication's live delivery.

func ExceptConnection

func ExceptConnection(id ConnectionID) PublishOption

ExceptConnection skips live delivery to one connection, typically the sender of a relayed message ("to others"). Recent replay still includes the event.

type Publisher

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

Publisher owns publication operations without a socket listener or auth scopes. Its constructor binds the distributed backend and immutable registry.

func NewPublisher

func NewPublisher(registry *Registry, config Config, backend ClusterBackend, cluster ClusterConfig, opts ...Option) (*Publisher, error)

NewPublisher constructs trusted server-side publication without authentication scopes or socket ownership. Use the exact same channel and cluster policies as the socket servers. Stop publication before closing the borrowed Redis client.

func (*Publisher) Close

func (p *Publisher) Close(ctx context.Context) error

Close stops publication and waits, bounded by ctx, for admitted operations and the cluster subscription to finish. A timeout ends only the wait; keep the borrowed adapter open until Done closes.

func (*Publisher) Done

func (p *Publisher) Done() <-chan struct{}

Done closes once every publication operation and background task exited.

type PublisherSource

type PublisherSource interface {
	// contains filtered or unexported methods
}

PublisherSource is implemented by a socket Hub and a server-side Publisher. The same typed declaration/room/payload boundary serves HTTP, jobs and sockets.

type Registration

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

func Register

func Register[C, R, S any](channel Channel[C, R, S], events ...EventRegistration[C]) Registration

Register is the only erasure boundary. Event ownership is checked by the Go compiler and by nominal declaration identity when the registry is built.

type Registry

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

func NewRegistry

func NewRegistry(registrations ...Registration) (*Registry, error)

func (*Registry) Channels

func (r *Registry) Channels() []ChannelInfo

Channels returns deterministic owned contract snapshots. Contract exporters consume these descriptors; no second payload or room schema is maintained.

type ReplayConfig

type ReplayConfig struct {
	Messages int           `json:"messages"`
	Bytes    int           `json:"bytes"`
	TTL      time.Duration `json:"ttl"`
}

ReplayConfig retains a recent per-channel bucket. Room replay filters that bucket; traffic in other rooms can evict older messages. Zero disables history. TTL is a duration in nanoseconds in Go/JSON metadata, not a reconnect cursor.

func (ReplayConfig) Validate

func (c ReplayConfig) Validate() error

type Request

type Request struct {
	Version int             `json:"v"`
	Action  Action          `json:"action"`
	ID      RequestID       `json:"id"`
	Channel ChannelID       `json:"channel"`
	Room    *string         `json:"room,omitempty"`
	Event   EventID         `json:"event,omitempty"`
	Payload json.RawMessage `json:"payload,omitempty"`
	Replay  *int            `json:"replay,omitempty"`
}

Request is the wire boundary only. Normal domain handlers receive typed MessageContext and payload values; no JSON map inspection is necessary.

func DecodeRequest

func DecodeRequest(data []byte, maxBytes int) (Request, error)

DecodeRequest enforces the exact versioned envelope in one strict pass: duplicate/unknown members, invalid UTF-8 and unpaired surrogates are rejected. Callers must bound frame buffering first. A malformed request still reports its ID when one was decoded and valid, so errors can correlate.

type RequestID

type RequestID string

type Response

type Response struct {
	Version   int             `json:"v"`
	Type      ResponseType    `json:"type"`
	ID        RequestID       `json:"id,omitempty"`
	Channel   ChannelID       `json:"channel,omitempty"`
	Room      *string         `json:"room,omitempty"`
	Event     EventID         `json:"event,omitempty"`
	MessageID MessageID       `json:"message_id,omitzero"`
	Payload   json.RawMessage `json:"payload,omitempty"`
	Code      Code            `json:"code,omitempty"`
	Members   []MemberFrame   `json:"members,omitempty"`
	Member    *MemberFrame    `json:"member,omitempty"`
	Replayed  bool            `json:"replayed,omitempty"`
}

func DecodePublication

func DecodePublication(data []byte, maxBytes int) (Response, error)

DecodePublication validates the stable event envelope for adapters and history in one strict pass. Runtime registries additionally validate the registered room and payload types.

type ResponseType

type ResponseType string
const (
	Subscribed      ResponseType = "subscribed"
	Unsubscribed    ResponseType = "unsubscribed"
	Acknowledged    ResponseType = "ack"
	Accepted        ResponseType = "accepted"
	ErrorResponse   ResponseType = "error"
	EventResponse   ResponseType = "event"
	PresenceJoined  ResponseType = "presence_joined"
	PresenceLeft    ResponseType = "presence_left"
	PresenceUpdated ResponseType = "presence_updated"
)

type Rooms

type Rooms[R any] struct {
	// contains filtered or unexported fields
}

func DefineRooms

func DefineRooms[R any](codec foundryhttp.PathCodec[R]) Rooms[R]

DefineRooms reuses the existing text codec and its same-value metadata.

func (Rooms[R]) Description

func (r Rooms[R]) Description() (foundryhttp.URLScalarInfo, error)

func (Rooms[R]) Validate

func (r Rooms[R]) Validate() error

type Scope

type Scope struct {
	Channel ChannelID `json:"channel"`
	Room    string    `json:"room"`
	HasRoom bool      `json:"has_room"`
}

Scope distinguishes whole-channel membership from an exact room. Adapters use its opaque Hash; applications retain Channel/Target rather than constructing it.

func (Scope) Hash

func (s Scope) Hash() string

func (Scope) Validate

func (s Scope) Validate() error

type ServerConfig

type ServerConfig struct {
	HTTP       foundryhttp.ServerConfig
	Path       string
	Middleware []foundryhttp.Middleware
}

func DefaultServerConfig

func DefaultServerConfig() ServerConfig

func (ServerConfig) Validate

func (c ServerConfig) Validate() error

type Snapshot

type Snapshot struct {
	Distributed, Degraded                                       bool
	Streaming                                                   bool
	BackgroundTasks                                             int
	Stopping                                                    bool
	Connections                                                 int
	Subscriptions                                               int
	ActiveOperations                                            int
	QueuedBytes                                                 int64
	Accepted, Rejected, Publications, SlowConsumers, Failures   uint64
	Gaps, Resubscriptions, DroppedEnvelopes, OperationOverloads uint64
}

type SubjectReference

type SubjectReference struct {
	Guard    auth.GuardName
	Provider auth.ProviderName
	Identity model.Identity
}

SubjectReference is immutable authenticated attribution for cleanup and management. It is not a credential and cannot authenticate another operation.

type Target

type Target[R any] struct{ Room value.Optional[R] }

Target distinguishes a whole-channel subscription from a concrete room. The same value type is retained through authorization and incoming handlers.

func Room

func Room[R any](room R) Target[R]

func WholeChannel

func WholeChannel[R any]() Target[R]

type TicketRedeemer

type TicketRedeemer interface {
	TicketSource() auth.CredentialName
	RedeemTicket(context.Context, secret.String) (value.Optional[auth.BoundCredential], error)
}

TicketRedeemer exchanges a single-use handshake ticket for a bound credential of the guard whose credential source it names. token.Tokens implements it. An unknown, expired or already redeemed ticket is omitted, not an error.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL