extensions

package
v1.0.28-beta.1 Latest Latest
Warning

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

 Go to latest Published: Jul 31, 2026 License: MIT Imports: 30 Imported by: 0

Documentation

Overview

Package extensions provides opt-in, bounded live transport adapters for CRDT replication groups.

Its zero-value feature set exposes no endpoint and never starts an HTTP listener. Applications remain responsible for TLS, identity lifecycle, durable state and outbox transactions, bootstrap/anti-entropy, membership, rate limits, and operations. The adapters are deliberately not a production replication service.

Index

Constants

View Source 
const (
	// Subprotocol identifies this package's control and binary change envelope.
	// It is independent from CRDT frame-format versions.
	Subprotocol = "crdt-sync-v1"
	// BatchSubprotocol carries a bounded list of complete v1 envelopes. It is
	// opt-in and does not change the v1 envelope or CRDT frame contract.
	BatchSubprotocol = "crdt-sync-v2"
)
View Source 
const (
	Relay_Sync_FullMethodName = "/darkinno.crdt.extensions.v1.Relay/Sync"
)

Variables

View Source 
var (
	// ErrInvalidConfig reports a missing or unsafe extensions configuration.
	ErrInvalidConfig = errors.New("crdt extensions: invalid configuration")
	// ErrUnauthorized reports authentication or authorization failure.
	ErrUnauthorized = errors.New("crdt extensions: unauthorized")
	// ErrClosed reports use of a closed transport client.
	ErrClosed = errors.New("crdt extensions: client is closed")
	// ErrBatchUnsupported reports an unavailable batch subprotocol.
	ErrBatchUnsupported = errors.New("crdt extensions: websocket batch subprotocol is not enabled")
	// ErrBatchLimit reports a local batch that exceeds a configured boundary.
	ErrBatchLimit = errors.New("crdt extensions: websocket batch limit exceeded")
)
View Source 
var File_extensions_relay_proto protoreflect.FileDescriptor
View Source 
var Relay_ServiceDesc = grpc.ServiceDesc{
	ServiceName: "darkinno.crdt.extensions.v1.Relay",
	HandlerType: (*RelayServer)(nil),
	Methods:     []grpc.MethodDesc{},
	Streams: []grpc.StreamDesc{
		{
			StreamName:    "Sync",
			Handler:       _Relay_Sync_Handler,
			ServerStreams: true,
			ClientStreams: true,
		},
	},
	Metadata: "extensions/relay.proto",
}

Relay_ServiceDesc is the grpc.ServiceDesc for Relay service. It's only intended for direct use with grpc.RegisterService, and not to be introspected or modified (even as a copy)

Functions

func RegisterRelayServer added in v1.0.25 

func RegisterRelayServer(s grpc.ServiceRegistrar, srv RelayServer)

Types

type Authenticate 

type Authenticate func(*http.Request) (Peer, error)

Authenticate authenticates one transport request before it is upgraded or its body is read. Returning an error rejects the request.

type Authorize 

type Authorize func(Peer, replica.Manifest, replica.Dot) error

Authorize binds one proposed CRDT change to an authenticated peer and its negotiated manifest. At minimum it should prevent a peer from publishing as another logical replica actor.

type AuthorizeSubscription 

type AuthorizeSubscription func(Peer, replica.Manifest) error

AuthorizeSubscription decides whether peer may receive live changes for the manifest. Read permission is deliberately separate from write authorization.

type ClientConfig 

type ClientConfig struct {
	Header          http.Header
	HTTPClient      *http.Client
	Policy          crdt.ProtocolPolicy
	MaxMessageBytes int
	MaxActorBytes   int
	// EnableBatches offers BatchSubprotocol in addition to Subprotocol. It
	// remains false by default so an existing client keeps its v1 contract.
	EnableBatches bool
	// MaxBatchChanges is a local publication and receive bound when
	// EnableBatches is true. It must not exceed the relay's configured bound;
	// both sides default to 16. It is ignored for a v1-only connection.
	MaxBatchChanges  int
	HandshakeTimeout time.Duration
	WriteTimeout     time.Duration
	OnChange         func(replica.Change) error
}

ClientConfig configures either optional transport client. OnChange must pass received changes to an application-owned, manifest-compatible replica.Inbox or an equivalently durable boundary. HTTPClient is used for HTTP/SSE and, if non-nil, as the underlying client for WebSocket dialing.

type Config 

type Config struct {
	Features              Feature
	Groups                []*Group
	Authenticate          Authenticate
	Authorize             Authorize
	AuthorizeSubscription AuthorizeSubscription
	// OriginPatterns lists case-insensitive host patterns permitted to make
	// browser-originated cross-origin requests, for example "app.example" or
	// "*.example.internal". It uses path.Match syntax, so "*" is rejected.
	// The request host is always permitted. These are host patterns rather than
	// URL strings so HTTP and WebSocket apply the same rule.
	OriginPatterns    []string
	MaxMessageBytes   int
	MaxActorBytes     int
	MaxQueuedMessages int
	MaxQueuedBytes    int
	// MaxBatchChanges bounds independently identified changes in one
	// crdt-sync-v2 WebSocket message. It is relevant only when
	// FeatureWebSocketBatch is enabled and cannot exceed MaxQueuedMessages,
	// so v1 and SSE peers can be queued atomically or disconnected.
	MaxBatchChanges  int
	HandshakeTimeout time.Duration
	WriteTimeout     time.Duration
	// Telemetry receives bounded, payload-free handshake and publication
	// outcomes. A nil Reporter is the default and leaves relay paths unchanged.
	Telemetry *telemetry.Reporter
}

Config configures one optional transport handler. Features is default-off; no endpoint is active unless FeatureWebSocket or FeatureHTTP is selected. Authentication and separate read/write authorization are required whenever a feature is active.

type Feature 

type Feature uint8

Feature selects an optional transport surface. The zero value intentionally enables no transport. 

const (
	// FeatureWebSocket enables the manifest-bound WebSocket endpoint.
	FeatureWebSocket Feature = 1 << iota
	// FeatureHTTP enables the HTTP publication and SSE live-event endpoints.
	FeatureHTTP
	// FeatureWebSocketBatch enables the opt-in crdt-sync-v2 WebSocket
	// subprotocol. It requires FeatureWebSocket and leaves HTTP/SSE on their
	// single-change v1 envelope.
	FeatureWebSocketBatch
)

func (Feature) Enabled 

func (f Feature) Enabled(feature Feature) bool

Enabled reports whether f includes feature.

type GRPCAuthenticate added in v1.0.25 

type GRPCAuthenticate func(context.Context) (Peer, error)

GRPCAuthenticate authenticates a gRPC stream before its manifest or change payload is read. It should derive the identity from transport credentials or trusted metadata, never from the CRDT actor supplied in a change.

type GRPCClient added in v1.0.25 

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

GRPCClient maintains one manifest-bound gRPC live subscription. It neither owns the ClientConn nor persists an outbox, and it does not reconnect automatically. The application owns credentials, TLS, connection reuse, durable recovery, and reconnect policy.

func OpenGRPC added in v1.0.25 

func OpenGRPC(ctx context.Context, relay RelayClient, manifest replica.Manifest, config GRPCClientConfig) (*GRPCClient, error)

OpenGRPC opens Relay.Sync, sends the local exact manifest, and verifies the exact manifest returned by the relay before it starts receiving changes. A successful return means the relay registered the live subscription before its manifest confirmation. relay is normally NewRelayClient(connection).

func (*GRPCClient) Close added in v1.0.25 

func (client *GRPCClient) Close() error

Close cancels the live stream but does not close the application-owned gRPC ClientConn or claim durable delivery.

func (*GRPCClient) Done added in v1.0.25 

func (client *GRPCClient) Done() <-chan struct{}

Done closes when the receive loop ends.

func (*GRPCClient) Err added in v1.0.25 

func (client *GRPCClient) Err() error

Err returns the first non-local stream or callback failure.

func (*GRPCClient) Publish added in v1.0.25 

func (client *GRPCClient) Publish(ctx context.Context, change replica.Change) error

Publish validates a change against the negotiated manifest and sends its canonical envelope. A successful return means only that gRPC accepted the message for transport; callers retain durable outbox and retry decisions.

type GRPCClientConfig added in v1.0.25 

type GRPCClientConfig struct {
	Policy          crdt.ProtocolPolicy
	MaxMessageBytes int
	MaxActorBytes   int
	OnChange        func(replica.Change) error
}

GRPCClientConfig configures a managed, manifest-bound Relay.Sync client. OnChange must hand every received change to an application-owned, manifest-compatible replica.Inbox or an equivalently durable boundary.

The context passed to OpenGRPC owns the lifetime of the live stream. It must have a realistic stream deadline (or be cancelled during shutdown); do not use a short handshake-only deadline because gRPC applies it to the entire RPC. Publish checks its context before waiting to send, while a blocked gRPC Send is interrupted by the stream context.

type GRPCConfig added in v1.0.25 

type GRPCConfig struct {
	Groups                []*Group
	Authenticate          GRPCAuthenticate
	Authorize             Authorize
	AuthorizeSubscription AuthorizeSubscription
	MaxMessageBytes       int
	MaxActorBytes         int
	MaxQueuedMessages     int
	MaxQueuedBytes        int
	// Telemetry receives bounded, payload-free handshake and publication
	// outcomes. A nil Reporter is the default and leaves relay paths unchanged.
	Telemetry *telemetry.Reporter
}

GRPCConfig configures the manifest-bound gRPC Relay service. It deliberately mirrors the live-relay security and capacity boundary without turning gRPC flow control into an unbounded application queue.

type GRPCRelay added in v1.0.25 

type GRPCRelay struct {
	UnimplementedRelayServer
	// contains filtered or unexported fields
}

GRPCRelay implements Relay over one bidirectional gRPC stream per live subscription. The first client message and first server response are exact encoded replica manifests. Later messages carry the existing canonical change envelope, so gRPC introduces no second CRDT frame format.

A gRPC stream is live-only. It has no replay cursor, operation log, persistent outbox, or automatic reconnection. Applications must persist state/frontier and recover missed changes independently.

func NewGRPCRelay added in v1.0.25 

func NewGRPCRelay(config GRPCConfig) (*GRPCRelay, error)

NewGRPCRelay validates and constructs a disabled-by-default gRPC transport surface. Authentication plus independent read/write authorization are mandatory because TLS authenticates a channel but does not decide a tenant's CRDT group permissions.

func NewGRPCServer added in v1.0.25 

func NewGRPCServer(config GRPCConfig) (*grpc.Server, *GRPCRelay, error)

NewGRPCServer constructs and registers an application-ready gRPC server. Hosts that share a grpc.Server may instead call NewGRPCRelay, install its ServerOptions during server construction, then RegisterRelayServer.

func (*GRPCRelay) ServerOptions added in v1.0.25 

func (relay *GRPCRelay) ServerOptions() []grpc.ServerOption

ServerOptions returns the message-size boundary required by the Relay protocol. Supply these options when mounting Relay in an application-owned grpc.Server; NewGRPCServer does so automatically.

func (*GRPCRelay) Sync added in v1.0.25 

func (relay *GRPCRelay) Sync(stream grpc.BidiStreamingServer[SyncMessage, SyncMessage]) error

Sync accepts one manifest-bound live stream. gRPC's transport flow control applies underneath, while the per-peer queue remains the bounded relay boundary: a slow stream is disconnected rather than retaining arbitrary application state in memory.

type Group 

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

Group owns a manifest-bound replica inbox and its currently connected live peers. It has no operation log, snapshot store, or replay history.

func NewGroup 

func NewGroup(config GroupConfig) (*Group, error)

NewGroup creates one manifest-bound in-memory receiver. Apply runs while delivery ordering is held, so it must use the concrete CRDT decoder with application limits, leave the CRDT unchanged on an error, and not re-enter Group or block on a transport callback.

func (*Group) Frontier 

func (g *Group) Frontier() replica.Frontier

Frontier returns a copy of g's installed contiguous delivery frontier.

func (*Group) Manifest 

func (g *Group) Manifest() replica.Manifest

Manifest returns the immutable-by-convention manifest negotiated by g.

func (*Group) Pending 

func (g *Group) Pending() (changes, bytes int)

Pending reports the number and bytes of out-of-order changes retained by g.

type GroupConfig 

type GroupConfig struct {
	Manifest          replica.Manifest
	Frontier          replica.Frontier
	Policy            crdt.ProtocolPolicy
	MaxPendingChanges int
	MaxPendingBytes   int
	Apply             replica.ApplyDelta
}

GroupConfig describes one bounded, manifest-bound receiver. Frontier must come from the same durable transaction as the application CRDT state when a production application restores a group.

type HTTPClient 

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

HTTPClient maintains one HTTP/SSE live subscription for a manifest. It publishes changes with POST and receives change events over SSE. It has no replay log and does not reconnect automatically.

func ConnectHTTP 

func ConnectHTTP(ctx context.Context, endpoint string, manifest replica.Manifest, config ClientConfig) (*HTTPClient, error)

ConnectHTTP starts an authenticated SSE subscription, verifies the exact manifest supplied by the relay, and then receives live changes. A successful return means the relay registered the subscription before it sent the stream manifest. endpoint is the base mount URL, for example "https://sync.example.com/crdt". The configured HTTPClient must not have a global Timeout because SSE is a long-lived response; use request contexts and WriteTimeout instead.

func (*HTTPClient) Close 

func (client *HTTPClient) Close() error

Close stops the SSE stream without claiming durable delivery.

func (*HTTPClient) Done 

func (client *HTTPClient) Done() <-chan struct{}

Done closes when the SSE receive loop stops.

func (*HTTPClient) Err 

func (client *HTTPClient) Err() error

Err returns the first non-local stream or callback error.

func (*HTTPClient) Publish 

func (client *HTTPClient) Publish(ctx context.Context, change replica.Change) error

Publish validates and posts a canonical change envelope. A POST failure does not close the independent SSE subscription; callers decide whether and how to retry from their durable outbox.

type Handler 

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

Handler exposes enabled optional endpoints. It is safe to mount into an application-owned http.ServeMux and never starts an HTTP listener itself.

func NewHandler 

func NewHandler(config Config) (*Handler, error)

NewHandler validates config and constructs a disabled handler for the zero feature set. A disabled handler returns 404 for every request and does not invoke authentication or start background work.

func (*Handler) Mount 

func (h *Handler) Mount(mux *http.ServeMux, prefix string) (err error)

Mount mounts h below prefix and strips that prefix before routing its endpoints. For example, Mount(mux, "/crdt/") exposes "/crdt/ws" and the HTTP endpoints under "/crdt/http/".

func (*Handler) ServeHTTP 

func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request)

ServeHTTP routes only enabled transport endpoints. It does not serve an index page or start a listener.

type Peer 

type Peer struct {
	ID string
}

Peer is an authenticated application identity. ID must be stable and must not be taken from a client-supplied CRDT actor identifier.

type RelayClient added in v1.0.25 

type RelayClient interface {
	Sync(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[SyncMessage, SyncMessage], error)
}

RelayClient is the client API for Relay service.

For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.

Relay is a manifest-bound, live-only CRDT transport. The first message from each side is a Hello. All later messages are independently identified, canonical CRDT change envelopes.

func NewRelayClient added in v1.0.25 

func NewRelayClient(cc grpc.ClientConnInterface) RelayClient

type RelayServer added in v1.0.25 

type RelayServer interface {
	Sync(grpc.BidiStreamingServer[SyncMessage, SyncMessage]) error
	// contains filtered or unexported methods
}

RelayServer is the server API for Relay service. All implementations must embed UnimplementedRelayServer for forward compatibility.

Relay is a manifest-bound, live-only CRDT transport. The first message from each side is a Hello. All later messages are independently identified, canonical CRDT change envelopes.

type Relay_SyncClient added in v1.0.25 

type Relay_SyncClient = grpc.BidiStreamingClient[SyncMessage, SyncMessage]

This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.

type Relay_SyncServer added in v1.0.25 

type Relay_SyncServer = grpc.BidiStreamingServer[SyncMessage, SyncMessage]

This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.

type SyncMessage added in v1.0.25 

type SyncMessage struct {

	// Types that are valid to be assigned to Payload:
	//
	//	*SyncMessage_Hello
	//	*SyncMessage_Change
	Payload isSyncMessage_Payload `protobuf_oneof:"payload"`
	// contains filtered or unexported fields
}

func (*SyncMessage) Descriptor deprecated added in v1.0.25 

func (*SyncMessage) Descriptor() ([]byte, []int)

Deprecated: Use SyncMessage.ProtoReflect.Descriptor instead.

func (*SyncMessage) GetChange added in v1.0.25 

func (x *SyncMessage) GetChange() []byte

func (*SyncMessage) GetHello added in v1.0.25 

func (x *SyncMessage) GetHello() []byte

func (*SyncMessage) GetPayload added in v1.0.25 

func (x *SyncMessage) GetPayload() isSyncMessage_Payload

func (*SyncMessage) ProtoMessage added in v1.0.25 

func (*SyncMessage) ProtoMessage()

func (*SyncMessage) ProtoReflect added in v1.0.25 

func (x *SyncMessage) ProtoReflect() protoreflect.Message

func (*SyncMessage) Reset added in v1.0.25 

func (x *SyncMessage) Reset()

func (*SyncMessage) String added in v1.0.25 

func (x *SyncMessage) String() string

type SyncMessage_Change added in v1.0.25 

type SyncMessage_Change struct {
	Change []byte `protobuf:"bytes,2,opt,name=change,proto3,oneof"`
}

type SyncMessage_Hello added in v1.0.25 

type SyncMessage_Hello struct {
	Hello []byte `protobuf:"bytes,1,opt,name=hello,proto3,oneof"`
}

type UnimplementedRelayServer added in v1.0.25 

type UnimplementedRelayServer struct{}

UnimplementedRelayServer must be embedded to have forward compatible implementations.

NOTE: this should be embedded by value instead of pointer to avoid a nil pointer dereference when methods are called.

func (UnimplementedRelayServer) Sync added in v1.0.25 

func (UnimplementedRelayServer) Sync(grpc.BidiStreamingServer[SyncMessage, SyncMessage]) error

type UnsafeRelayServer added in v1.0.25 

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

UnsafeRelayServer may be embedded to opt out of forward compatibility for this service. Use of this interface is not recommended, as added methods to RelayServer will result in compilation errors.

type WebSocketClient 

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

WebSocketClient maintains one manifest-bound WebSocket live connection. It has no persistent outbox and does not reconnect automatically.

func DialWebSocket 

func DialWebSocket(ctx context.Context, endpoint string, manifest replica.Manifest, config ClientConfig) (*WebSocketClient, error)

DialWebSocket authenticates the WebSocket handshake through config.Header, verifies the exact manifest returned by the relay, then receives live binary changes. A successful return means the relay registered the live subscription before it sent the confirmation. Use wss after the host application has configured TLS.

func (*WebSocketClient) Close 

func (client *WebSocketClient) Close() error

Close terminates the live connection. It does not make a durability claim; persist an application checkpoint before a graceful recovery boundary.

func (*WebSocketClient) Done 

func (client *WebSocketClient) Done() <-chan struct{}

Done closes when the receive loop stops.

func (*WebSocketClient) Err 

func (client *WebSocketClient) Err() error

Err returns the first non-local connection or callback failure.

func (*WebSocketClient) Publish 

func (client *WebSocketClient) Publish(ctx context.Context, change replica.Change) error

Publish validates change against the connection manifest, then transmits its canonical envelope. A caller may retry after an ambiguous network result; durable outbox and recovery behavior remain application-owned.

func (*WebSocketClient) PublishBatch added in v1.0.19 

func (client *WebSocketClient) PublishBatch(ctx context.Context, changes []replica.Change) error

PublishBatch sends independently identified changes in one explicitly negotiated WebSocket message. It is a transport coalescing operation, not an atomic application transaction. The relay can accept an earlier item and reject a later one, so callers must retain every original change in their durable outbox and retry them independently after any ambiguous connection result.

type YJSAuthorize added in v1.0.26 

type YJSAuthorize func(Peer, string, YJSMessageKind) error

YJSAuthorize authorizes publication to one configured room. Yjs update bytes are opaque to this relay: authorization belongs to the room and authenticated identity, never to a client-selected Yjs client ID.

type YJSAuthorizeSubscription added in v1.0.26 

type YJSAuthorizeSubscription func(Peer, string) error

YJSAuthorizeSubscription authorizes access to a configured room separately from publication.

type YJSConfig added in v1.0.26 

type YJSConfig struct {
	Rooms                 []*YJSRoom
	Authenticate          Authenticate
	Authorize             YJSAuthorize
	AuthorizeSubscription YJSAuthorizeSubscription
	OriginPatterns        []string
	MaxMessageBytes       int
	MaxQueuedMessages     int
	MaxQueuedBytes        int
	MaxAwarenessClients   int
	HandshakeTimeout      time.Duration
	WriteTimeout          time.Duration
}

YJSConfig configures a y-websocket-compatible y-protocols relay. It has no default rooms and requires authentication plus independent read/write authorization. The transport is live-only except for each room's explicitly bounded in-memory update history.

type YJSHandler added in v1.0.26 

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

YJSHandler serves y-websocket-compatible paths. Mount it below a prefix; the final single path element selects a configured room, e.g. /yjs/notes. It accepts the standard y-protocols sync (0), awareness (1), and awareness query (3) messages. Authentication and permissions are application-owned.

func NewYJSHandler added in v1.0.26 

func NewYJSHandler(config YJSConfig) (*YJSHandler, error)

NewYJSHandler validates and constructs the opt-in compatibility relay.

func (*YJSHandler) ServeHTTP added in v1.0.26 

func (handler *YJSHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request)

ServeHTTP serves one configured room selected by a single escaped path segment. Dynamic room creation is intentionally not supported: accepting an untrusted room name must not allocate retained server state.

type YJSMessageKind added in v1.0.26 

type YJSMessageKind uint8

YJSMessageKind identifies the message class presented to YJSAuthorize. It intentionally does not expose document or awareness payloads to an authorization callback. 

const (
	// YJSUpdate is a Yjs sync-step-2 or update payload.
	YJSUpdate YJSMessageKind = iota + 1
	// YJSAwareness is an ephemeral y-protocols awareness payload.
	YJSAwareness
)

type YJSRoom added in v1.0.26 

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

YJSRoom is a bounded, opaque Yjs update cache and live subscriber set. It is intentionally isolated from Group: y-protocols and this module's framed CRDT protocols have incompatible state and recovery semantics.

func NewYJSRoom added in v1.0.26 

func NewYJSRoom(config YJSRoomConfig) (*YJSRoom, error)

NewYJSRoom creates one room. The zero value is deliberately not usable: room names and every retained-resource boundary must be selected by the embedding application.

func (*YJSRoom) Name added in v1.0.26 

func (room *YJSRoom) Name() string

Name returns the configured immutable room name.

type YJSRoomConfig added in v1.0.26 

type YJSRoomConfig struct {
	Name            string
	MaxUpdateBytes  int
	MaxHistoryBytes int
	MaxUpdates      int
}

YJSRoomConfig configures one explicitly named, in-memory y-protocols room. A room retains complete Yjs update messages only to bootstrap later live peers. It cannot compact or validate Yjs document semantics; production hosts must replace it with a Yjs-aware durable store before its bounded history is exhausted.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.