crdt

package
v1.6.0 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: MIT Imports: 16 Imported by: 2

Documentation

Overview

Package crdt provides an optional CRDT (Conflict-Free Replicated Data Type) layer for Grove. It enables offline-first, multi-node, and eventually-consistent use cases by tracking field-level changes with Hybrid Logical Clocks and merging them automatically during sync.

Tag model fields with crdt:lww, crdt:counter, or crdt:set to opt into CRDT behavior. Fields without CRDT tags are unaffected. 

type Document struct {
    grove.BaseModel `grove:"table:documents,alias:d"`
    ID        string   `grove:"id,pk"`
    Title     string   `grove:"title,crdt:lww"`
    ViewCount int64    `grove:"view_count,crdt:counter"`
    Tags      []string `grove:"tags,type:jsonb,crdt:set"`
}

Package crdt provides conflict-free replicated data types for Grove.

The crdt package enables offline-first, multi-node, and eventually-consistent use cases by adding an optional CRDT layer on top of Grove's existing ORM. It supports three CRDT types (LWW-Register, PN-Counter, OR-Set), multiple sync topologies (edge-to-cloud, peer-to-peer, hub-and-spoke), and integrates with the Forge ecosystem for routing, streaming, and middleware.

CRDT Types

Tag struct fields with crdt:"type" to enable CRDT merge semantics:

  • crdt:lww — Last-Writer-Wins Register. Higher HLC timestamp wins.
  • crdt:counter — PN-Counter. Per-node increment/decrement, merged by max.
  • crdt:set — OR-Set. Add-wins observed-remove set.

Fields without crdt: tags work normally with zero overhead.

Quick Start 

// 1. Define a CRDT-enabled model
type Document struct {
    grove.BaseModel `grove:"table:documents,alias:d"`
    ID        string   `grove:"id,pk"`
    Title     string   `grove:"title,crdt:lww"`
    ViewCount int64    `grove:"view_count,crdt:counter"`
    Tags      []string `grove:"tags,type:jsonb,crdt:set"`
}

// 2. Create the CRDT plugin
plugin := crdt.New(crdt.WithNodeID("node-1"))
db.Hooks().AddHook(plugin, hook.Scope{Tables: []string{"documents"}})

// 3. Sync between nodes
syncer := crdt.NewSyncer(plugin,
    crdt.WithTransport(crdt.HTTPTransport("https://cloud.example.com/sync")),
    crdt.WithSyncTables("documents"),
)
report, err := syncer.Sync(ctx)

Forge Integration

When running inside a Forge app, use the grove/extension package with WithCRDT to get automatic route registration, SSE streaming, and middleware support: 

ext := extension.New(
    extension.WithDriver(pgdb),
    extension.WithCRDT(crdtPlugin, hook.Scope{Tables: []string{"documents"}}),
    extension.WithSyncer(syncer),
    extension.WithMigrations(crdt.Migrations),
)
app.RegisterExtension(ext)

// Routes registered automatically:
//   POST /sync/pull    — pull changes from this node
//   POST /sync/push    — push changes to this node
//   GET  /sync/stream  — SSE real-time change stream

Sync Hooks

Implement SyncHook to intercept data during sync operations for validation, transformation, filtering, or auditing: 

type MyHook struct { crdt.BaseSyncHook }

func (h *MyHook) BeforeInboundChange(ctx context.Context, c *crdt.ChangeRecord) (*crdt.ChangeRecord, error) {
    // validate, transform, or reject incoming changes
    return c, nil
}

func (h *MyHook) BeforeOutboundRead(ctx context.Context, changes []crdt.ChangeRecord) ([]crdt.ChangeRecord, error) {
    // filter changes before sending to remote peers
    return changes, nil
}

plugin := crdt.New(
    crdt.WithNodeID("node-1"),
    crdt.WithSyncHook(&MyHook{}),
)

Streaming Transport

Use StreamingTransport for real-time SSE-based change propagation alongside the periodic poll-based sync: 

syncer := crdt.NewSyncer(crdtPlugin,
    crdt.WithTransport(crdt.NewStreamingTransport("https://cloud.example.com/sync",
        crdt.WithStreamTables("documents"),
    )),
    crdt.WithSyncInterval(30*time.Second), // Fallback poll interval
)

// Start both: SSE streaming for real-time + periodic poll as fallback
go syncer.Run(ctx)         // Periodic pull/push
go syncer.StreamSync(ctx)  // SSE real-time (if transport supports it)

Hybrid Logical Clock

All CRDT operations use a Hybrid Logical Clock (HLC) that combines physical wall-clock time with a logical counter and node ID. This provides total ordering without coordination between nodes.

Storage

CRDT metadata is stored in shadow tables (_<table>_crdt) alongside the primary table. The primary table schema is never modified. Shadow tables are created automatically via migrations or Plugin.EnsureShadowTable.

Standalone Usage (No Forge)

The crdt package works without Forge. Use NewHTTPHandler to create a standard http.Handler for sync endpoints: 

handler := crdt.NewHTTPHandler(crdtPlugin)
mux.Handle("/sync/", handler)

Collaborative Text

crdt:text fields are character-level collaborative sequences with formatting attributes (the YText analog): TextState with origin-span addressing, stable cursor positions (TextRef, TextState.RefAt, TextState.IndexOf), Quill-style deltas (TextState.Delta) and run-coalesced sequential typing. ORM writes reconcile whole strings via TextState.SetString; transports carry TextOp payloads.

Op Application And Compaction

ApplyChange is the canonical ChangeRecord → FieldState fold every transport and downstream store uses: it honors all type-specific payloads (counter deltas, set/list/text ops, document path writes, and full-state carriers via ChangeRecord.State). Long-lived states GC their tombstones with the horizon-based Compact methods (State.Compact, RGAListState.Compact, ORSetState.Compact, TextState.Compact).

Architecture 

grove/crdt/
├── crdt.go          — Core types: CRDTType, State, FieldState, ChangeRecord
├── clock.go         — HLC implementation
├── register.go      — LWW-Register merge
├── counter.go       — PN-Counter merge
├── set.go           — OR-Set merge (add-wins)
├── list.go          — RGA list merge (cached document order)
├── text.go          — Text CRDT (origin-span fragments, formatting)
├── document.go      — Nested path-keyed document merge
├── merge.go         — MergeEngine: field + state merging
├── apply.go         — ApplyChange: canonical op application
├── compact.go       — Horizon-based tombstone compaction (GC)
├── plugin.go        — Plugin (Grove hook integration)
├── hooks.go         — PostMutation/PreQuery hooks for CRDT
├── metadata.go      — Shadow table read/write operations
├── sync.go          — Syncer: push/pull orchestration + StreamSync
├── sync_hooks.go    — SyncHook interface for intercepting sync data
├── transport.go     — Transport interface, HTTPClient, StreamingTransport
├── server.go        — SyncController + HTTPHandler (standalone)
├── options.go       — Functional options
├── inspect.go       — Debug/inspect utilities
└── migrations.go    — Shadow table DDL

Index

Constants

View Source 
const (
	PresenceJoin      = "join"
	PresenceUpdateEvt = "update"
	PresenceLeave     = "leave"
)

Presence event type constants.

View Source 
const DefaultChangesLimit = 10000

DefaultChangesLimit is the maximum number of change records returned by ReadChangesSince when no explicit limit is provided. This prevents unbounded result sets on large shadow tables.

Variables

This section is empty.

Functions

func DocumentRoomID 

func DocumentRoomID(table, pk string) string

DocumentRoomID returns the standard room ID for a table + primary key. Use this for per-document collaboration rooms.

func DropShadowTableDDL 

func DropShadowTableDDL(table string) string

DropShadowTableDDL generates the DROP TABLE statement for a shadow table.

func MarshalPresenceEvent 

func MarshalPresenceEvent(event PresenceEvent) ([]byte, error)

MarshalEvent serializes a PresenceEvent to JSON bytes for SSE transport.

func NewHTTPHandler 

func NewHTTPHandler(plugin *Plugin, opts ...SyncControllerOption) http.Handler

NewHTTPHandler creates a standard http.Handler for sync endpoints. Use this when not running inside a Forge app. For Forge apps, use grove/extension.WithCRDT() which auto-registers routes.

Endpoints:

  • POST /pull — remote nodes pull changes from this node
  • POST /push — remote nodes push changes to this node

func RoomHTTPHandler 

func RoomHTTPHandler(rm *RoomManager) http.Handler

RoomHTTPHandler provides HTTP endpoints for room management. Mount under the sync server path (e.g., /sync/rooms).

func ShadowTableDDL 

func ShadowTableDDL(table string) string

ShadowTableDDL generates the CREATE TABLE statement for a CRDT shadow table. The DDL uses $1-style placeholders that are compatible with PostgreSQL. For SQLite, the caller should substitute the appropriate syntax.

func ShadowTableName 

func ShadowTableName(table string) string

ShadowTableName returns the shadow table name for a given table.

func ShadowTableSyncIndex 

func ShadowTableSyncIndex(table string) string

ShadowTableSyncIndex generates the CREATE INDEX statement for efficient sync queries. The index covers (hlc_ts, hlc_counter) to match the ReadChangesSince query which filters on both columns.

func ValidCRDTType 

func ValidCRDTType(t string) bool

ValidCRDTType returns true if t is a recognized CRDT type.

Types

type AttrState 

type AttrState struct {
	Value  json.RawMessage `json:"value"`
	HLC    HLC             `json:"hlc"`
	NodeID string          `json:"node_id"`
}

AttrState is one formatting attribute's LWW register on a fragment. A JSON null Value clears the attribute (kept as an LWW tombstone).

type BaseCRDTPlugin 

type BaseCRDTPlugin struct{}

BaseCRDTPlugin provides no-op defaults for all plugin interfaces. Embed this in your plugin struct and override only what you need.

func (BaseCRDTPlugin) AfterHistoryRead 

func (BaseCRDTPlugin) AfterHistoryRead(_ context.Context, _, _ string, s *State) (*State, error)

func (BaseCRDTPlugin) AfterMerge 

func (BaseCRDTPlugin) AfterMerge(_ context.Context, _ *MergeEvent) error

func (BaseCRDTPlugin) AfterMetadataRead 

func (BaseCRDTPlugin) AfterMetadataRead(_ context.Context, _ *MetadataReadEvent, s *State) (*State, error)

func (BaseCRDTPlugin) AfterMetadataWrite 

func (BaseCRDTPlugin) AfterMetadataWrite(_ context.Context, _ *MetadataWriteEvent) error

func (BaseCRDTPlugin) AfterPresenceEvent 

func (BaseCRDTPlugin) AfterPresenceEvent(_ context.Context, _ *PresenceEvent) error

func (BaseCRDTPlugin) AfterRoomJoin 

func (BaseCRDTPlugin) AfterRoomJoin(_ context.Context, _, _ string) error

func (BaseCRDTPlugin) AfterRoomLeave 

func (BaseCRDTPlugin) AfterRoomLeave(_ context.Context, _, _ string) error

func (BaseCRDTPlugin) BeforeHistoryRead 

func (BaseCRDTPlugin) BeforeHistoryRead(_ context.Context, _, _ string, _ HLC) error

TimeTravelInterceptor no-ops.

func (BaseCRDTPlugin) BeforeMerge 

func (BaseCRDTPlugin) BeforeMerge(_ context.Context, ev *MergeEvent) (*FieldState, error)

MergeInterceptor no-ops.

func (BaseCRDTPlugin) BeforeMetadataRead 

func (BaseCRDTPlugin) BeforeMetadataRead(_ context.Context, _ *MetadataReadEvent) error

func (BaseCRDTPlugin) BeforeMetadataWrite 

func (BaseCRDTPlugin) BeforeMetadataWrite(_ context.Context, ev *MetadataWriteEvent) (*FieldState, error)

MetadataInterceptor no-ops.

func (BaseCRDTPlugin) BeforePresenceUpdate 

func (BaseCRDTPlugin) BeforePresenceUpdate(_ context.Context, u *PresenceUpdate) (*PresenceUpdate, error)

PresenceInterceptor no-ops.

func (BaseCRDTPlugin) BeforeRoomJoin 

func (BaseCRDTPlugin) BeforeRoomJoin(_ context.Context, _, _ string) error

RoomInterceptor no-ops.

func (BaseCRDTPlugin) BeforeRoomLeave 

func (BaseCRDTPlugin) BeforeRoomLeave(_ context.Context, _, _ string) error

func (BaseCRDTPlugin) Name 

func (BaseCRDTPlugin) Name() string

func (BaseCRDTPlugin) OnClientConnect 

func (BaseCRDTPlugin) OnClientConnect(_ context.Context, _, _ string) error

ConnectionInterceptor no-ops.

func (BaseCRDTPlugin) OnClientDisconnect 

func (BaseCRDTPlugin) OnClientDisconnect(_ context.Context, _, _ string) error

func (BaseCRDTPlugin) OnRoomCreated 

func (BaseCRDTPlugin) OnRoomCreated(_ context.Context, _ *Room) error

func (BaseCRDTPlugin) OnRoomDestroyed 

func (BaseCRDTPlugin) OnRoomDestroyed(_ context.Context, _ *Room) error

type BaseSyncHook 

type BaseSyncHook struct{}

BaseSyncHook provides no-op implementations of all SyncHook methods. Embed it in your struct and override only the methods you need: 

type MyHook struct { crdt.BaseSyncHook }

func (h *MyHook) BeforeInboundChange(ctx context.Context, c *crdt.ChangeRecord) (*crdt.ChangeRecord, error) {
    // your logic here
    return c, nil
}

func (BaseSyncHook) AfterInboundChange 

func (BaseSyncHook) AfterInboundChange(_ context.Context, _ *ChangeRecord) error

AfterInboundChange does nothing.

func (BaseSyncHook) BeforeInboundChange 

func (BaseSyncHook) BeforeInboundChange(_ context.Context, c *ChangeRecord) (*ChangeRecord, error)

BeforeInboundChange passes the change through unchanged.

func (BaseSyncHook) BeforeOutboundChange 

func (BaseSyncHook) BeforeOutboundChange(_ context.Context, c *ChangeRecord) (*ChangeRecord, error)

BeforeOutboundChange passes the change through unchanged.

func (BaseSyncHook) BeforeOutboundRead 

func (BaseSyncHook) BeforeOutboundRead(_ context.Context, cs []ChangeRecord) ([]ChangeRecord, error)

BeforeOutboundRead passes the slice through unchanged.

type CRDTPlugin 

type CRDTPlugin interface {
	// Name returns a unique identifier for this plugin.
	Name() string
}

CRDTPlugin is the interface for extending the CRDT server with custom logic. Implement any subset of the specialized interfaces below to hook into specific parts of the CRDT lifecycle. Use BaseCRDTPlugin for no-op defaults.

Plugins are registered via SyncController.AddPlugin or WithControllerPlugin option and are called in registration order.

Example: 

type AuditPlugin struct { crdt.BaseCRDTPlugin }

func (p *AuditPlugin) Name() string { return "audit" }

func (p *AuditPlugin) AfterMerge(ctx context.Context, ev *MergeEvent) error {
    log.Printf("merged %s/%s field=%s winner=%s", ev.Table, ev.PK, ev.Field, ev.WinnerNodeID)
    return nil
}

type CRDTType 

type CRDTType string //nolint:revive // CRDTType is the established public API name

CRDTType identifies the conflict resolution strategy for a field. 

const (
	// TypeLWW is a Last-Writer-Wins register. The value with the highest HLC wins.
	TypeLWW CRDTType = "lww"

	// TypeCounter is a PN-Counter. Each node tracks its own increments and
	// decrements; the global value is the sum across all nodes.
	TypeCounter CRDTType = "counter"

	// TypeSet is an Observed-Remove Set (OR-Set) with add-wins semantics.
	// Concurrent add and remove of the same element results in the element
	// being present (add wins).
	TypeSet CRDTType = "set"

	// TypeList is a Replicated Growable Array (RGA) for ordered sequences.
	// Elements have stable positions and can be inserted, deleted, or moved
	// concurrently without conflicts.
	TypeList CRDTType = "list"

	// TypeDocument is a recursive CRDT map that supports nested paths.
	// Each nested field is independently mergeable with its own CRDT type,
	// enabling JSON-like nested structures.
	TypeDocument CRDTType = "document"

	// TypeText is a character-level collaborative text sequence with
	// formatting attributes (origin-span fragments; see text.go).
	TypeText CRDTType = "text"
)

type ChangeRecord 

type ChangeRecord struct {
	Table     string          `json:"table"`
	PK        string          `json:"pk"`
	Field     string          `json:"field"`
	CRDTType  CRDTType        `json:"crdt_type"`
	HLC       HLC             `json:"hlc"`
	NodeID    string          `json:"node_id"`
	Value     json.RawMessage `json:"value,omitempty"`
	Tombstone bool            `json:"tombstone,omitempty"`

	// Type-specific payloads (only one is set based on CRDTType).
	CounterDelta *CounterDelta `json:"counter_delta,omitempty"`
	SetOp        *SetOperation `json:"set_op,omitempty"`
	ListOp       *ListOp       `json:"list_op,omitempty"`
	TextOp       *TextOp       `json:"text_op,omitempty"`

	// State optionally carries the field's full CRDT state for state-based
	// propagation (sets/lists/documents sync losslessly this way). When set,
	// ApplyChange merges it directly and ignores the op payloads.
	State *FieldState `json:"state,omitempty"`
}

ChangeRecord represents a single field-level change for sync transport.

type Clock 

type Clock interface {
	// Now returns a new HLC value that is causally after all previously
	// observed values.
	Now() HLC

	// Update merges a received remote HLC into the local clock state,
	// ensuring the next Now() is causally after both the local and remote values.
	Update(remote HLC)
}

Clock is the interface for generating HLC values.

type ClockOption 

type ClockOption func(*HybridClock)

ClockOption configures a HybridClock.

func WithMaxDrift 

func WithMaxDrift(d time.Duration) ClockOption

WithMaxDrift sets the maximum tolerable clock drift. Update() will return values clamped to this drift from the physical clock.

func WithNowFunc 

func WithNowFunc(fn func() time.Time) ClockOption

WithNowFunc overrides the wall clock source (useful for testing).

type ConnectionInterceptor 

type ConnectionInterceptor interface {
	// OnClientConnect is called when a client connects via SSE or WebSocket.
	OnClientConnect(ctx context.Context, nodeID string, transport string) error

	// OnClientDisconnect is called when a client disconnects.
	OnClientDisconnect(ctx context.Context, nodeID string, transport string) error
}

ConnectionInterceptor intercepts client connections (WebSocket, SSE).

type CounterDelta 

type CounterDelta struct {
	Increment int64 `json:"inc"`
	Decrement int64 `json:"dec"`
}

CounterDelta represents a single node's counter change.

type CursorPosition 

type CursorPosition struct {
	// X/Y for canvas-style cursors.
	X float64 `json:"x,omitempty"`
	Y float64 `json:"y,omitempty"`

	// Offset for text-style cursors (character position).
	Offset int `json:"offset,omitempty"`

	// Line/Column for code-style cursors.
	Line   int `json:"line,omitempty"`
	Column int `json:"column,omitempty"`

	// SelectionStart/End for text selections.
	SelectionStart int `json:"selection_start,omitempty"`
	SelectionEnd   int `json:"selection_end,omitempty"`

	// Field identifies which field/element the cursor is in.
	Field string `json:"field,omitempty"`
}

CursorPosition represents a cursor or selection in a document.

type DocumentCRDTState 

type DocumentCRDTState struct {
	// Fields maps dot-separated paths to their field-level CRDT state.
	// Paths use "." as the separator for nesting.
	// Example: "address.city" → LWW, "tags" → Set, "views" → Counter.
	Fields map[string]*FieldState `json:"fields"`
}

DocumentCRDTState holds nested document CRDT state. Each path (e.g., "address.city", "tags", "meta.count") is independently mergeable with its own CRDT type, enabling JSON-like nested structures where different subtrees can use different conflict resolution strategies.

func DocumentFromFieldState 

func DocumentFromFieldState(fs *FieldState) *DocumentCRDTState

DocumentFromFieldState reconstructs a DocumentCRDTState from a FieldState.

func MergeDocument 

func MergeDocument(local, remote *DocumentCRDTState) (*DocumentCRDTState, error)

MergeDocument merges two document CRDT states by merging each path independently using the MergeEngine. This is commutative, associative, and idempotent.

func NewDocumentCRDTState 

func NewDocumentCRDTState() *DocumentCRDTState

NewDocumentCRDTState creates an empty document CRDT state.

func (*DocumentCRDTState) DeleteField 

func (d *DocumentCRDTState) DeleteField(path string)

DeleteField removes a field at the given path and all its children.

func (*DocumentCRDTState) GetField 

func (d *DocumentCRDTState) GetField(path string) *FieldState

GetField returns the resolved value at the given path. Returns nil if the path doesn't exist.

func (*DocumentCRDTState) Paths 

func (d *DocumentCRDTState) Paths() []string

Paths returns all field paths sorted alphabetically.

func (*DocumentCRDTState) Resolve 

func (d *DocumentCRDTState) Resolve() map[string]any

Resolve converts the document state to a nested map structure. This is the "materialized view" of the document.

func (*DocumentCRDTState) SetField 

func (d *DocumentCRDTState) SetField(path string, value any, clock HLC, nodeID string) error

SetField sets a field at the given path with LWW semantics.

func (*DocumentCRDTState) SetFieldState 

func (d *DocumentCRDTState) SetFieldState(path string, fs *FieldState)

SetFieldState sets a field at the given path with an explicit FieldState. Use this for non-LWW types (counters, sets, lists).

func (*DocumentCRDTState) ToFieldState 

func (d *DocumentCRDTState) ToFieldState(clock HLC, nodeID string) *FieldState

ToFieldState converts to the generic FieldState representation.

type ExecResult 

type ExecResult interface {
	RowsAffected() (int64, error)
}

ExecResult is the result of an exec operation.

type Executor 

type Executor interface {
	ExecContext(ctx context.Context, query string, args ...any) (ExecResult, error)
	QueryContext(ctx context.Context, query string, args ...any) (Rows, error)
}

Executor is the minimal query interface needed by MetadataStore. Both grove.DB (via driver) and grove.Tx satisfy this via adapter.

type FieldHistoryEntry 

type FieldHistoryEntry struct {
	HLC    HLC             `json:"hlc"`
	NodeID string          `json:"node_id"`
	Value  json.RawMessage `json:"value,omitempty"`
	Type   CRDTType        `json:"type"`
}

FieldHistoryEntry is a single version of a field's state.

type FieldHistoryRequest 

type FieldHistoryRequest struct {
	Table    string `json:"table"`
	PK       string `json:"pk"`
	Field    string `json:"field"`
	SinceHLC HLC    `json:"since_hlc,omitempty"`
	Limit    int    `json:"limit,omitempty"`
}

FieldHistoryRequest asks for the change history of a specific field.

type FieldHistoryResponse 

type FieldHistoryResponse struct {
	Table   string              `json:"table"`
	PK      string              `json:"pk"`
	Field   string              `json:"field"`
	Entries []FieldHistoryEntry `json:"entries"`
}

FieldHistoryResponse contains the change history for a field.

type FieldState 

type FieldState struct {
	// Type is the CRDT type of this field.
	Type CRDTType `json:"type"`

	// HLC is the clock value of the last write (used by LWW).
	HLC HLC `json:"hlc"`

	// NodeID is the node that produced the last write (used by LWW).
	NodeID string `json:"node_id"`

	// Value is the current resolved value (LWW) or nil for Counter/Set.
	Value json.RawMessage `json:"value,omitempty"`

	// CounterState holds per-node increments/decrements (Counter only).
	CounterState *PNCounterState `json:"counter_state,omitempty"`

	// SetState holds the OR-Set state (Set only).
	SetState *ORSetState `json:"set_state,omitempty"`

	// ListState holds the RGA list state (List only).
	ListState *RGAListState `json:"list_state,omitempty"`

	// DocState holds nested document CRDT state (Document only).
	DocState *DocumentCRDTState `json:"doc_state,omitempty"`

	// TextState holds collaborative text state (Text only).
	TextState *TextState `json:"text_state,omitempty"`
}

FieldState holds the CRDT state for a single field.

func ApplyChange 

func ApplyChange(engine *MergeEngine, local *FieldState, c *ChangeRecord) (*FieldState, error)

ApplyChange folds one ChangeRecord into a field's local state and returns the merged result. It is the canonical op-application seam: transports, stores and downstream systems (e.g. fabriq's document plane) fold update logs through it so every replica applies an op identically.

Application is type-aware:

  • a change carrying State (a full FieldState) merges state-based;
  • counter changes fold CounterDelta as that node's delta snapshot;
  • set changes fold SetOp (adds tag entries; removes mark observed tags — exact when the op names Tags, otherwise every local tag older than the op's HLC, the legacy observed-remove approximation);
  • list changes fold ListOp (insert/delete/move); deletes for unseen node ids are kept as tombstones so late-arriving inserts stay deleted;
  • text changes fold TextOp (insert/delete/format);
  • lww and document changes fold Value (document values are {path, value} path writes, or path deletes when the record is tombstoned).

local may be nil (first op for the field). The change is never mutated.

type ForgeWSConn 

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

ForgeWSConn adapts a Forge Connection (from forge.WebSocketHandler) to the WebSocketConn interface used by CRDT transport. This allows the CRDT WebSocket handler to work natively with Forge's built-in WebSocket upgrade machinery.

Usage inside a Forge WebSocket handler: 

sync.WebSocket("/ws", func(ctx forge.Context, conn forge.Connection) error {
    adapter := crdt.NewForgeWSConn(conn)
    handler := crdt.NewWebSocketHandler(ctrl, adapter, logger)
    return handler.Serve(conn.Context())
})

func NewForgeWSConn 

func NewForgeWSConn(conn forgeConnection) *ForgeWSConn

NewForgeWSConn creates a WebSocketConn adapter from a forge.Connection.

func (*ForgeWSConn) Close 

func (f *ForgeWSConn) Close() error

Close implements WebSocketConn using forge.Connection.Close().

func (*ForgeWSConn) ID 

func (f *ForgeWSConn) ID() string

ID returns the forge connection ID.

func (*ForgeWSConn) ReadMessage 

func (f *ForgeWSConn) ReadMessage() ([]byte, error)

ReadMessage implements WebSocketConn using forge.Connection.Read().

func (*ForgeWSConn) WriteMessage 

func (f *ForgeWSConn) WriteMessage(data []byte) error

WriteMessage implements WebSocketConn using forge.Connection.Write().

type HLC 

type HLC struct {
	// Timestamp is the physical time in nanoseconds since Unix epoch.
	Timestamp int64 `json:"ts"`

	// Counter is the logical counter, incremented when the physical clock
	// hasn't advanced since the last event.
	Counter uint32 `json:"c"`

	// NodeID identifies the node that produced this clock value.
	NodeID string `json:"node"`
}

HLC is a Hybrid Logical Clock value. It combines a physical timestamp (wall clock) with a logical counter to provide a totally ordered, causally consistent clock that requires no coordination between nodes.

func (HLC) After 

func (h HLC) After(other HLC) bool

After returns true if h is strictly after other.

func (HLC) Compare 

func (h HLC) Compare(other HLC) int

Compare returns -1 if h < other, 0 if equal, 1 if h > other. Ordering: Timestamp first, then Counter, then NodeID (lexicographic tiebreak).

func (HLC) IsZero 

func (h HLC) IsZero() bool

IsZero returns true if the HLC has not been set.

func (HLC) MarshalJSON 

func (h HLC) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler: ts as an exact decimal string.

func (HLC) String 

func (h HLC) String() string

func (*HLC) UnmarshalJSON 

func (h *HLC) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler: ts as string or legacy number.

type HTTPClient 

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

HTTPClient is a Transport implementation that communicates via HTTP.

func HTTPTransport 

func HTTPTransport(baseURL string) *HTTPClient

HTTPTransport creates a new HTTP transport client pointing at the given base URL. The sync server should be mounted at that URL (see NewHTTPHandler). 

transport := crdt.HTTPTransport("https://cloud.example.com/sync")

func HTTPTransportWithClient 

func HTTPTransportWithClient(baseURL string, client *http.Client) *HTTPClient

HTTPTransportWithClient creates an HTTP transport with a custom http.Client.

func (*HTTPClient) Pull 

func (c *HTTPClient) Pull(ctx context.Context, req *PullRequest) (*PullResponse, error)

Pull requests changes from the remote node.

func (*HTTPClient) Push 

func (c *HTTPClient) Push(ctx context.Context, req *PushRequest) (*PushResponse, error)

Push sends local changes to the remote node.

type HistoryRequest 

type HistoryRequest struct {
	Table string `json:"table"`
	PK    string `json:"pk"`
	// AtHLC is the point in time to query. If zero, returns current state.
	AtHLC HLC `json:"at_hlc,omitempty"`
}

HistoryRequest asks for the state of a record at a specific point in time.

type HistoryResponse 

type HistoryResponse struct {
	Table string `json:"table"`
	PK    string `json:"pk"`
	AtHLC HLC    `json:"at_hlc"`
	State *State `json:"state"`
}

HistoryResponse contains the historical state of a record.

type HybridClock 

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

HybridClock is the default Clock implementation using a Hybrid Logical Clock.

func NewHybridClock 

func NewHybridClock(nodeID string, opts ...ClockOption) *HybridClock

NewHybridClock creates a new HLC for the given node.

func (*HybridClock) Now 

func (c *HybridClock) Now() HLC

Now produces a new HLC that is causally after the last observed value.

func (*HybridClock) Update 

func (c *HybridClock) Update(remote HLC)

Update merges a remote HLC into the local state. The next call to Now() will return a value that is causally after both local and remote.

type InspectField 

type InspectField struct {
	Type   CRDTType `json:"type"`
	NodeID string   `json:"node_id"`
	HLC    HLC      `json:"hlc"`

	// LWW fields
	Value any `json:"value,omitempty"`

	// Counter fields
	CounterValue int64               `json:"counter_value,omitempty"`
	NodeCounters map[string][2]int64 `json:"node_counters,omitempty"` // nodeID → [inc, dec]

	// Set fields
	Elements []any `json:"elements,omitempty"`

	// List fields
	ListLength int `json:"list_length,omitempty"`

	// Document fields
	DocPaths []string `json:"doc_paths,omitempty"`
}

InspectField is a human-readable representation of a single field's state.

type InspectResult 

type InspectResult struct {
	Table     string                   `json:"table"`
	PK        string                   `json:"pk"`
	Tombstone bool                     `json:"tombstone"`
	Fields    map[string]*InspectField `json:"fields"`
}

InspectResult is a human-readable representation of a record's CRDT state. Useful for debugging and testing.

func InspectState 

func InspectState(state *State) *InspectResult

InspectState converts a State into a human-readable InspectResult.

func (*InspectResult) String 

func (r *InspectResult) String() string

String returns a human-readable summary of the inspect result.

type LWWRegister 

type LWWRegister struct {
	Value  json.RawMessage `json:"value"`
	Clock  HLC             `json:"hlc"`
	NodeID string          `json:"node_id"`
}

LWWRegister is a Last-Writer-Wins register. The value with the highest HLC wins. Ties are broken deterministically by node ID.

func LWWFromFieldState 

func LWWFromFieldState(fs *FieldState) *LWWRegister

LWWFromFieldState reconstructs an LWWRegister from a FieldState.

func MergeLWW 

func MergeLWW(local, remote *LWWRegister) *LWWRegister

MergeLWW merges two LWW registers, returning the winning value. The register with the higher HLC wins. If HLCs are equal, the higher node ID wins (deterministic tiebreak).

func NewLWWRegister 

func NewLWWRegister(value any, clock HLC, nodeID string) (*LWWRegister, error)

NewLWWRegister creates a new LWW register with the given value and clock.

func (*LWWRegister) Decode 

func (r *LWWRegister) Decode(dest any) error

Decode unmarshals the register's value into dest.

func (*LWWRegister) ToFieldState 

func (r *LWWRegister) ToFieldState() *FieldState

ToFieldState converts to the generic FieldState representation.

type ListOp 

type ListOp struct {
	Op       ListOpType      `json:"op"`
	NodeID   HLC             `json:"node_id,omitempty"`
	ParentID HLC             `json:"parent_id,omitempty"`
	Value    json.RawMessage `json:"value,omitempty"`
}

ListOp represents a list operation for the sync transport.

type ListOpType 

type ListOpType string

ListOpType identifies the list operation. 

const (
	ListOpInsert ListOpType = "insert"
	ListOpDelete ListOpType = "delete"
	ListOpMove   ListOpType = "move"
)

type MergeEngine 

type MergeEngine struct{}

MergeEngine resolves concurrent writes by dispatching to the appropriate CRDT merge function based on field type.

func NewMergeEngine 

func NewMergeEngine() *MergeEngine

NewMergeEngine creates a new MergeEngine.

func (*MergeEngine) MergeField 

func (m *MergeEngine) MergeField(local, remote *FieldState) (*FieldState, error)

MergeField merges a single field's state from two sources. Both states must have the same CRDT type.

func (*MergeEngine) MergeState 

func (m *MergeEngine) MergeState(local, remote *State) (*State, error)

MergeState merges two full record states. Fields present in only one state are kept as-is. Fields present in both are merged using MergeField. Tombstones are resolved by taking the one with the higher HLC.

type MergeEvent 

type MergeEvent struct {
	Table  string      `json:"table"`
	PK     string      `json:"pk"`
	Field  string      `json:"field"`
	Local  *FieldState `json:"local"`
	Remote *FieldState `json:"remote"`
	Result *FieldState `json:"result"` // Set after merge (in AfterMerge).

	// WinnerNodeID is set in AfterMerge — the node whose value won.
	WinnerNodeID string `json:"winner_node_id,omitempty"`

	// ConflictDetected is true when both local and remote had changes.
	ConflictDetected bool `json:"conflict_detected"`
}

MergeEvent provides context about a merge operation.

type MergeInterceptor 

type MergeInterceptor interface {
	// BeforeMerge is called before two field states are merged.
	// Return a modified remote state, nil to skip the merge, or an error to abort.
	BeforeMerge(ctx context.Context, ev *MergeEvent) (*FieldState, error)

	// AfterMerge is called after a merge completes with the winning state.
	// Use for audit logging, analytics, or triggering side effects.
	AfterMerge(ctx context.Context, ev *MergeEvent) error
}

MergeInterceptor intercepts merge operations, allowing custom merge logic, validation, or auditing of conflict resolution decisions.

type MetadataInterceptor 

type MetadataInterceptor interface {
	// BeforeMetadataWrite is called before writing field state to the shadow table.
	// Return a modified state, nil to skip the write, or an error to abort.
	BeforeMetadataWrite(ctx context.Context, ev *MetadataWriteEvent) (*FieldState, error)

	// AfterMetadataWrite is called after a successful shadow table write.
	AfterMetadataWrite(ctx context.Context, ev *MetadataWriteEvent) error

	// BeforeMetadataRead is called before reading state from the shadow table.
	// Return an error to deny the read.
	BeforeMetadataRead(ctx context.Context, ev *MetadataReadEvent) error

	// AfterMetadataRead is called after reading state, allowing transformation.
	// Return a modified state or the original.
	AfterMetadataRead(ctx context.Context, ev *MetadataReadEvent, state *State) (*State, error)
}

MetadataInterceptor intercepts reads and writes to the shadow table.

type MetadataReadEvent 

type MetadataReadEvent struct {
	Table string `json:"table"`
	PK    string `json:"pk"`
	// AtHLC is set for time-travel reads (zero for current state).
	AtHLC HLC `json:"at_hlc,omitempty"`
}

MetadataReadEvent describes a pending shadow table read.

type MetadataRow 

type MetadataRow struct {
	PKHash    string          `json:"pk_hash"`
	FieldName string          `json:"field_name"`
	HLCTS     int64           `json:"hlc_ts"`
	HLCCount  uint32          `json:"hlc_counter"`
	NodeID    string          `json:"node_id"`
	Tombstone bool            `json:"tombstone"`
	CRDTState json.RawMessage `json:"crdt_state"`
}

MetadataRow is a single row in the shadow table.

type MetadataStore 

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

MetadataStore reads and writes CRDT metadata in shadow tables. It operates via a generic Executor interface so it works with any Grove driver (pg, sqlite, turso, etc.).

func NewMetadataStore 

func NewMetadataStore(exec Executor) *MetadataStore

NewMetadataStore creates a new MetadataStore with the given executor.

func (*MetadataStore) CleanTombstones 

func (ms *MetadataStore) CleanTombstones(ctx context.Context, table string, olderThan int64) error

CleanTombstones removes tombstones older than the given HLC.

func (*MetadataStore) ReadChangesSince 

func (ms *MetadataStore) ReadChangesSince(ctx context.Context, table string, since HLC, limits ...int) ([]ChangeRecord, error)

ReadChangesSince reads change records from the shadow table that happened after the given HLC timestamp. Used by the sync protocol. An optional limit can be provided (first value used); 0 means use DefaultChangesLimit.

func (*MetadataStore) ReadFieldHistory 

func (ms *MetadataStore) ReadFieldHistory(ctx context.Context, table, pk, field string, since HLC, limit int) ([]FieldHistoryEntry, error)

ReadFieldHistory reads the change history for a specific field.

func (*MetadataStore) ReadState 

func (ms *MetadataStore) ReadState(ctx context.Context, table, pk string) (*State, error)

ReadState reads the full CRDT state for a record from the shadow table.

func (*MetadataStore) ReadStateAt 

func (ms *MetadataStore) ReadStateAt(ctx context.Context, table, pk string, at HLC) (*State, error)

ReadStateAt reads the CRDT state for a record as it existed at a specific HLC timestamp. This queries the shadow table for all field states with hlc_ts <= the target time.

func (*MetadataStore) WriteFieldState 

func (ms *MetadataStore) WriteFieldState(ctx context.Context, table, pk, field string, fs *FieldState) error

WriteFieldState writes a single field's CRDT state to the shadow table.

func (*MetadataStore) WriteFieldStatesAtomic 

func (ms *MetadataStore) WriteFieldStatesAtomic(ctx context.Context, table, pk string, fields map[string]*FieldState) error

WriteFieldStatesAtomic writes multiple field states in a single transaction when the executor supports it. Falls back to individual writes otherwise.

func (*MetadataStore) WriteTombstone 

func (ms *MetadataStore) WriteTombstone(ctx context.Context, table, pk string, clock HLC, nodeID string) error

WriteTombstone marks a record as deleted in the shadow table.

type MetadataWriteEvent 

type MetadataWriteEvent struct {
	Table  string      `json:"table"`
	PK     string      `json:"pk"`
	Field  string      `json:"field"`
	State  *FieldState `json:"state"`
	NodeID string      `json:"node_id"`
}

MetadataWriteEvent describes a pending shadow table write.

type Metrics 

type Metrics struct {
	// Sync metrics.
	PullCount     atomic.Int64
	PushCount     atomic.Int64
	PullLatencyNs atomic.Int64 // last pull latency in nanoseconds
	PushLatencyNs atomic.Int64 // last push latency in nanoseconds
	PullErrors    atomic.Int64
	PushErrors    atomic.Int64
	ChangesPulled atomic.Int64
	ChangesPushed atomic.Int64
	ChangesMerged atomic.Int64

	// Conflict metrics.
	ConflictsTotal atomic.Int64
	ConflictsLWW   atomic.Int64
	ConflictsSet   atomic.Int64
	ConflictsList  atomic.Int64

	// Presence metrics.
	ActiveRooms        atomic.Int64
	ActiveParticipants atomic.Int64
	PresenceUpdates    atomic.Int64

	// Connection metrics.
	SSEConnections atomic.Int64
	WSConnections  atomic.Int64

	// Validation metrics.
	ValidationErrors atomic.Int64
}

Metrics collects CRDT operational metrics. It uses atomic counters for lock-free concurrent access. Register a MetricsCollector to receive periodic snapshots, or read counters directly.

func (*Metrics) RecordConflict 

func (m *Metrics) RecordConflict(crdtType CRDTType)

RecordConflict increments the conflict counter for the given CRDT type.

func (*Metrics) Reset 

func (m *Metrics) Reset()

Reset zeroes all counters.

func (*Metrics) Snapshot 

func (m *Metrics) Snapshot() MetricsSnapshot

Snapshot returns a point-in-time copy of all metrics.

type MetricsSnapshot 

type MetricsSnapshot struct {
	Timestamp time.Time `json:"timestamp"`

	PullCount     int64 `json:"pull_count"`
	PushCount     int64 `json:"push_count"`
	PullLatencyMs int64 `json:"pull_latency_ms"`
	PushLatencyMs int64 `json:"push_latency_ms"`
	PullErrors    int64 `json:"pull_errors"`
	PushErrors    int64 `json:"push_errors"`
	ChangesPulled int64 `json:"changes_pulled"`
	ChangesPushed int64 `json:"changes_pushed"`
	ChangesMerged int64 `json:"changes_merged"`

	ConflictsTotal int64 `json:"conflicts_total"`
	ConflictsLWW   int64 `json:"conflicts_lww"`
	ConflictsSet   int64 `json:"conflicts_set"`
	ConflictsList  int64 `json:"conflicts_list"`

	ActiveRooms        int64 `json:"active_rooms"`
	ActiveParticipants int64 `json:"active_participants"`
	PresenceUpdates    int64 `json:"presence_updates"`

	SSEConnections   int64 `json:"sse_connections"`
	WSConnections    int64 `json:"ws_connections"`
	ValidationErrors int64 `json:"validation_errors"`
}

MetricsSnapshot is a point-in-time copy of all metrics.

type ORSetState 

type ORSetState struct {
	// Entries maps element (JSON-encoded) → set of tags that added it.
	// An element is in the set if it has at least one tag not in Removed.
	Entries map[string][]Tag `json:"entries"`

	// Removed tracks tags that have been observed and removed.
	Removed map[string]bool `json:"removed"`
}

ORSetState holds the state for an Observed-Remove Set with add-wins semantics. Each element is tracked with a unique tag (nodeID + HLC) so that concurrent add and remove of the same element resolves to the element being present (the add wins).

func MergeSet 

func MergeSet(local, remote *ORSetState) *ORSetState

MergeSet merges two OR-Set states. The result is the union of all entries with the union of all removed tags. This is commutative, associative, and idempotent.

func NewORSetState 

func NewORSetState() *ORSetState

NewORSetState creates an empty OR-Set.

func SetFromFieldState 

func SetFromFieldState(fs *FieldState) *ORSetState

SetFromFieldState reconstructs an ORSetState from a FieldState.

func (*ORSetState) Add 

func (s *ORSetState) Add(element any, nodeID string, clock HLC) error

Add inserts an element into the set with the given tag.

func (*ORSetState) Compact 

func (s *ORSetState) Compact(before HLC) int

Compact drops observed-removed tags older than the horizon along with their removal markers, pruning entries left tagless. Returns the number of tags dropped.

func (*ORSetState) Contains 

func (s *ORSetState) Contains(element any) (bool, error)

Contains returns true if the element is in the effective set.

func (*ORSetState) Elements 

func (s *ORSetState) Elements() []json.RawMessage

Elements returns the effective set of elements (those with at least one non-removed tag).

func (*ORSetState) Remove 

func (s *ORSetState) Remove(element any) error

Remove removes an element by marking all its current tags as removed.

func (*ORSetState) ToFieldState 

func (s *ORSetState) ToFieldState(clock HLC, nodeID string) *FieldState

ToFieldState converts to the generic FieldState representation.

type Option 

type Option func(*Plugin)

Option configures the CRDT plugin.

func WithClock 

func WithClock(c Clock) Option

WithClock sets the clock implementation. Defaults to a HybridClock using the node ID.

func WithMaxClockDrift 

func WithMaxClockDrift(d time.Duration) Option

WithMaxClockDrift sets the maximum tolerable clock drift between nodes. Remote HLC values that exceed this drift from the local clock are clamped. Defaults to 5 seconds.

func WithNodeID 

func WithNodeID(id string) Option

WithNodeID sets the unique identifier for this node. Required.

func WithSyncHook 

func WithSyncHook(hook SyncHook) Option

WithSyncHook adds a sync hook to the plugin. Sync hooks intercept changes during sync operations for validation, transformation, filtering, or auditing. Multiple hooks are called in registration order.

func WithTables 

func WithTables(tables ...string) Option

WithTables restricts the CRDT plugin to the specified tables. If not set, the plugin operates on all tables that have crdt: tags.

func WithTombstoneTTL 

func WithTombstoneTTL(d time.Duration) Option

WithTombstoneTTL sets how long tombstones are retained before being eligible for garbage collection. Defaults to 7 days.

type PNCounterState 

type PNCounterState struct {
	// Increments maps nodeID → total increments from that node.
	Increments map[string]int64 `json:"inc"`

	// Decrements maps nodeID → total decrements from that node.
	Decrements map[string]int64 `json:"dec"`
}

PNCounterState holds the per-node increment and decrement maps for a PN-Counter (Positive-Negative Counter). The global value is: 

sum(all increments) - sum(all decrements)

Each node only ever writes to its own entry; merging takes the max of each node's counters.

func CounterFromFieldState 

func CounterFromFieldState(fs *FieldState) *PNCounterState

CounterFromFieldState reconstructs a PNCounterState from a FieldState.

func MergeCounter 

func MergeCounter(local, remote *PNCounterState) *PNCounterState

MergeCounter merges two PN-Counter states by taking the max of each node's increments and decrements. This is commutative, associative, and idempotent.

func NewPNCounterState 

func NewPNCounterState() *PNCounterState

NewPNCounterState creates an empty PN-Counter state.

func (*PNCounterState) Decrement 

func (c *PNCounterState) Decrement(nodeID string, delta int64)

Decrement adds delta to the given node's decrement counter.

func (*PNCounterState) Increment 

func (c *PNCounterState) Increment(nodeID string, delta int64)

Increment adds delta to the given node's increment counter.

func (*PNCounterState) ToFieldState 

func (c *PNCounterState) ToFieldState(clock HLC, nodeID string) *FieldState

ToFieldState converts to the generic FieldState representation.

func (*PNCounterState) Value 

func (c *PNCounterState) Value() int64

Value returns the current counter value (sum of increments minus sum of decrements).

type ParticipantData 

type ParticipantData struct {
	// Name is the display name of the participant.
	Name string `json:"name,omitempty"`

	// Color is the assigned collaboration color (hex).
	Color string `json:"color,omitempty"`

	// Avatar is a URL to the participant's avatar image.
	Avatar string `json:"avatar,omitempty"`

	// Cursor is the participant's current cursor position.
	Cursor *CursorPosition `json:"cursor,omitempty"`

	// IsTyping indicates whether the participant is currently typing.
	IsTyping bool `json:"is_typing,omitempty"`

	// ActiveField is the field currently being edited (for form-style UIs).
	ActiveField string `json:"active_field,omitempty"`

	// Status is a custom status string (e.g., "idle", "editing", "viewing").
	Status string `json:"status,omitempty"`

	// Extra holds arbitrary user-defined data.
	Extra map[string]any `json:"extra,omitempty"`
}

ParticipantData is the standard presence payload for room participants. Consumers can extend this with custom fields via the Extra map.

type Plugin 

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

Plugin is the CRDT plugin for Grove. It implements the grove plugin.Plugin interface along with WithHooks and WithMigrations capabilities.

Create a plugin with New() and register it with the hook engine: 

p := crdt.New(crdt.WithNodeID("node-1"))
db.Hooks().AddHook(p, hook.Scope{Tables: []string{"documents"}})

func New 

func New(opts ...Option) *Plugin

New creates a new CRDT plugin with the given options.

func (*Plugin) AfterMutation 

func (p *Plugin) AfterMutation(ctx context.Context, qc *hook.QueryContext, data, _ any) error

AfterMutation is called after every INSERT, UPDATE, or DELETE on a CRDT-enabled table. It writes CRDT metadata to the shadow table.

func (*Plugin) BeforeQuery 

func (p *Plugin) BeforeQuery(_ context.Context, _ *hook.QueryContext) (*hook.HookResult, error)

BeforeQuery is a no-op by default. In merge-on-read mode it could inject filters, but the default merge-on-write approach doesn't need it.

func (*Plugin) CleanupTombstones 

func (p *Plugin) CleanupTombstones(ctx context.Context, table string) error

CleanupTombstones removes tombstones older than the configured TTL.

func (*Plugin) Clock 

func (p *Plugin) Clock() Clock

Clock returns the plugin's clock instance.

func (*Plugin) EnsureShadowTable 

func (p *Plugin) EnsureShadowTable(ctx context.Context, table string) error

EnsureShadowTable creates the shadow table for the given table if it doesn't exist. This should be called during migration or initialization.

func (*Plugin) Init 

func (p *Plugin) Init(_ context.Context, _ any) error

Init is called when the plugin is registered with a DB.

func (*Plugin) Inspect 

func (p *Plugin) Inspect(ctx context.Context, table, pk string) (*State, error)

Inspect returns the full CRDT state for a record. Useful for debugging.

func (*Plugin) MergeEngine 

func (p *Plugin) MergeEngine() *MergeEngine

MergeEngine returns the plugin's merge engine.

func (*Plugin) MetadataStore 

func (p *Plugin) MetadataStore() *MetadataStore

MetadataStore returns the metadata store (nil until SetExecutor is called).

func (*Plugin) Name 

func (p *Plugin) Name() string

Name returns the plugin identifier.

func (*Plugin) NodeID 

func (p *Plugin) NodeID() string

NodeID returns the plugin's node identifier.

func (*Plugin) RunTombstoneCleanup 

func (p *Plugin) RunTombstoneCleanup(ctx context.Context, interval time.Duration, tables ...string)

RunTombstoneCleanup starts a background loop that periodically removes tombstones older than the configured TTL for the given tables. The loop runs until the context is cancelled.

func (*Plugin) SetExecutor 

func (p *Plugin) SetExecutor(exec Executor)

SetExecutor sets the database executor for metadata operations. This is called by the hooks when they have access to the database.

func (*Plugin) SyncHooks 

func (p *Plugin) SyncHooks() *SyncHookChain

SyncHooks returns the plugin's sync hook chain.

type PluginChain 

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

PluginChain manages registered CRDT plugins and dispatches events.

func NewPluginChain 

func NewPluginChain() *PluginChain

NewPluginChain creates a new plugin chain.

func (*PluginChain) Add 

func (pc *PluginChain) Add(p CRDTPlugin)

Add registers a plugin.

func (*PluginChain) DispatchAfterHistoryRead 

func (pc *PluginChain) DispatchAfterHistoryRead(ctx context.Context, table, pk string, state *State) (*State, error)

DispatchAfterHistoryRead calls AfterHistoryRead on all TimeTravelInterceptor plugins.

func (*PluginChain) DispatchAfterMerge 

func (pc *PluginChain) DispatchAfterMerge(ctx context.Context, ev *MergeEvent)

DispatchAfterMerge calls AfterMerge on all MergeInterceptor plugins.

func (*PluginChain) DispatchAfterMetadataRead 

func (pc *PluginChain) DispatchAfterMetadataRead(ctx context.Context, ev *MetadataReadEvent, state *State) (*State, error)

DispatchAfterMetadataRead calls AfterMetadataRead on all MetadataInterceptor plugins.

func (*PluginChain) DispatchAfterMetadataWrite 

func (pc *PluginChain) DispatchAfterMetadataWrite(ctx context.Context, ev *MetadataWriteEvent)

DispatchAfterMetadataWrite calls AfterMetadataWrite on all MetadataInterceptor plugins.

func (*PluginChain) DispatchAfterPresenceEvent 

func (pc *PluginChain) DispatchAfterPresenceEvent(ctx context.Context, event *PresenceEvent)

DispatchAfterPresenceEvent calls AfterPresenceEvent on all PresenceInterceptor plugins.

func (*PluginChain) DispatchAfterRoomJoin 

func (pc *PluginChain) DispatchAfterRoomJoin(ctx context.Context, roomID, nodeID string)

DispatchAfterRoomJoin calls AfterRoomJoin on all RoomInterceptor plugins.

func (*PluginChain) DispatchBeforeHistoryRead 

func (pc *PluginChain) DispatchBeforeHistoryRead(ctx context.Context, table, pk string, atHLC HLC) error

DispatchBeforeHistoryRead calls BeforeHistoryRead on all TimeTravelInterceptor plugins.

func (*PluginChain) DispatchBeforeMerge 

func (pc *PluginChain) DispatchBeforeMerge(ctx context.Context, ev *MergeEvent) (*FieldState, error)

DispatchBeforeMerge calls BeforeMerge on all MergeInterceptor plugins.

func (*PluginChain) DispatchBeforeMetadataRead 

func (pc *PluginChain) DispatchBeforeMetadataRead(ctx context.Context, ev *MetadataReadEvent) error

DispatchBeforeMetadataRead calls BeforeMetadataRead on all MetadataInterceptor plugins.

func (*PluginChain) DispatchBeforeMetadataWrite 

func (pc *PluginChain) DispatchBeforeMetadataWrite(ctx context.Context, ev *MetadataWriteEvent) (*FieldState, error)

DispatchBeforeMetadataWrite calls BeforeMetadataWrite on all MetadataInterceptor plugins.

func (*PluginChain) DispatchBeforePresenceUpdate 

func (pc *PluginChain) DispatchBeforePresenceUpdate(ctx context.Context, update *PresenceUpdate) (*PresenceUpdate, error)

DispatchBeforePresenceUpdate calls BeforePresenceUpdate on all PresenceInterceptor plugins.

func (*PluginChain) DispatchBeforeRoomJoin 

func (pc *PluginChain) DispatchBeforeRoomJoin(ctx context.Context, roomID, nodeID string) error

DispatchBeforeRoomJoin calls BeforeRoomJoin on all RoomInterceptor plugins.

func (*PluginChain) DispatchOnClientConnect 

func (pc *PluginChain) DispatchOnClientConnect(ctx context.Context, nodeID, transport string) error

DispatchOnClientConnect calls OnClientConnect on all ConnectionInterceptor plugins.

func (*PluginChain) DispatchOnClientDisconnect 

func (pc *PluginChain) DispatchOnClientDisconnect(ctx context.Context, nodeID, transport string)

DispatchOnClientDisconnect calls OnClientDisconnect on all ConnectionInterceptor plugins.

func (*PluginChain) Len 

func (pc *PluginChain) Len() int

Len returns the number of registered plugins.

func (*PluginChain) Plugins 

func (pc *PluginChain) Plugins() []CRDTPlugin

Plugins returns all registered plugins.

type PresenceEvent 

type PresenceEvent struct {
	// Type is "join", "update", or "leave".
	Type string `json:"type"`

	// NodeID identifies the client whose presence changed.
	NodeID string `json:"node_id"`

	// Topic is the presence scope.
	Topic string `json:"topic"`

	// Data is the user-defined presence payload (empty for "leave" events).
	Data json.RawMessage `json:"data,omitempty"`
}

PresenceEvent is broadcast over SSE to notify clients of presence changes.

type PresenceInterceptor 

type PresenceInterceptor interface {
	// BeforePresenceUpdate is called before updating presence.
	// Return a modified update, nil to reject, or an error to abort.
	BeforePresenceUpdate(ctx context.Context, update *PresenceUpdate) (*PresenceUpdate, error)

	// AfterPresenceEvent is called after a presence event is emitted.
	AfterPresenceEvent(ctx context.Context, event *PresenceEvent) error
}

PresenceInterceptor intercepts presence events for custom logic.

type PresenceManager 

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

PresenceManager manages ephemeral presence state for connected clients. State is kept entirely in-memory (never persisted to the database) and cleaned up automatically via TTL expiry. It is safe for concurrent use.

func NewPresenceManager 

func NewPresenceManager(ttl time.Duration, onChange func(PresenceEvent), logger log.Logger) *PresenceManager

NewPresenceManager creates a presence manager with the given TTL and change callback. The callback is invoked on join, update, and leave events (including TTL-based expiry). It starts a background goroutine for TTL cleanup; call Close() to stop it.

func (*PresenceManager) Close 

func (pm *PresenceManager) Close()

Close stops the background cleanup goroutine.

func (*PresenceManager) Get 

func (pm *PresenceManager) Get(topic string) []PresenceState

Get returns all active presence states for a topic.

func (*PresenceManager) GetTopicsForNode 

func (pm *PresenceManager) GetTopicsForNode(nodeID string) []string

GetTopicsForNode returns all topics that a node has presence in.

func (*PresenceManager) Remove 

func (pm *PresenceManager) Remove(topic, nodeID string) *PresenceEvent

Remove explicitly removes a node's presence from a topic. Returns a "leave" event. No-op if the entry doesn't exist.

func (*PresenceManager) RemoveNode 

func (pm *PresenceManager) RemoveNode(nodeID string) []PresenceEvent

RemoveNode removes all presence entries for the given node (e.g., on SSE disconnect). Broadcasts a "leave" event for each removed entry.

func (*PresenceManager) Update 

func (pm *PresenceManager) Update(update PresenceUpdate) PresenceEvent

Update upserts a presence entry for the given node and topic. Returns the resulting event ("join" for new entries, "update" for existing).

type PresenceSnapshot 

type PresenceSnapshot struct {
	// Topic is the presence scope.
	Topic string `json:"topic"`

	// States is the list of active presence entries for the topic.
	States []PresenceState `json:"states"`
}

PresenceSnapshot is the response for GET /sync/presence?topic=...

type PresenceState 

type PresenceState struct {
	// NodeID identifies the client that owns this presence entry.
	NodeID string `json:"node_id"`

	// Topic is the presence scope (e.g. "documents:doc-1" or "lobby").
	Topic string `json:"topic"`

	// Data is the user-defined presence payload (cursor position, typing state, etc.).
	Data json.RawMessage `json:"data"`

	// UpdatedAt is the time this entry was last updated.
	UpdatedAt time.Time `json:"updated_at"`

	// ExpiresAt is the time this entry expires if not refreshed.
	ExpiresAt time.Time `json:"expires_at"`
}

PresenceState holds a single client's presence data for a topic. Topics are typically "table:pk" for per-document presence, but can be any arbitrary string for rooms or channels.

type PresenceUpdate 

type PresenceUpdate struct {
	// NodeID identifies the client sending the update.
	NodeID string `json:"node_id"`

	// Topic is the presence scope.
	Topic string `json:"topic"`

	// Data is the user-defined presence payload. Set to null to leave.
	Data json.RawMessage `json:"data"`
}

PresenceUpdate is the request body for POST /sync/presence.

type PullRequest 

type PullRequest struct {
	// Tables to pull changes for.
	Tables []string `json:"tables"`

	// Since is the HLC after which to return changes.
	Since HLC `json:"since"`

	// NodeID of the requesting node.
	NodeID string `json:"node_id"`

	// Filter enables selective sync (partial replication).
	// When set, only changes matching the filter are returned.
	Filter *SyncFilter `json:"filter,omitempty"`
}

PullRequest asks a remote node for changes since a given point.

type PullResponse 

type PullResponse struct {
	// Changes are the field-level change records.
	Changes []ChangeRecord `json:"changes"`

	// LatestHLC is the highest HLC in the response.
	LatestHLC HLC `json:"latest_hlc"`
}

PullResponse contains changes from the remote node.

type PushRequest 

type PushRequest struct {
	// Changes to push.
	Changes []ChangeRecord `json:"changes"`

	// NodeID of the pushing node.
	NodeID string `json:"node_id"`
}

PushRequest sends local changes to a remote node.

type PushResponse 

type PushResponse struct {
	// Merged is the number of changes that were merged.
	Merged int `json:"merged"`

	// LatestHLC is the remote node's latest HLC after merging.
	LatestHLC HLC `json:"latest_hlc"`
}

PushResponse acknowledges a push.

type RGAListState 

type RGAListState struct {
	// Nodes stores all nodes (including tombstoned) keyed by their ID string.
	Nodes map[string]*RGANode `json:"nodes"`
	// contains filtered or unexported fields
}

RGAListState holds the full state of an RGA list.

func ListFromFieldState 

func ListFromFieldState(fs *FieldState) *RGAListState

ListFromFieldState reconstructs an RGAListState from a FieldState.

func MergeList 

func MergeList(local, remote *RGAListState) *RGAListState

MergeList merges two RGA list states. The result contains all nodes from both lists with tombstones preserved. This is commutative, associative, and idempotent.

func NewRGAListState 

func NewRGAListState() *RGAListState

NewRGAListState creates an empty RGA list.

func (*RGAListState) Compact 

func (l *RGAListState) Compact(before HLC) int

Compact drops tombstoned leaf nodes older than the horizon, cascading until no more can be dropped. Returns the number of nodes removed.

func (*RGAListState) Delete 

func (l *RGAListState) Delete(id HLC)

Delete marks the node with the given ID as tombstoned. The order cache stays valid: tombstones remain traversal anchors and are filtered on read.

func (*RGAListState) Elements 

func (l *RGAListState) Elements() []json.RawMessage

Elements returns the visible (non-tombstoned) elements in order.

func (*RGAListState) Insert 

func (l *RGAListState) Insert(value any, parentID HLC, nodeID string, clock HLC) error

Insert adds a new element after the given parentID. If parentID is zero, the element is prepended to the list.

func (*RGAListState) Len 

func (l *RGAListState) Len() int

Len returns the number of visible (non-tombstoned) elements.

func (*RGAListState) Move 

func (l *RGAListState) Move(id, newParentID HLC, nodeID string, clock HLC)

Move moves an element to a new position after the given parentID. Implemented as tombstone + re-insert with new ID.

func (*RGAListState) NodeIDs 

func (l *RGAListState) NodeIDs() []HLC

NodeIDs returns the IDs of visible elements in order. Useful for addressing specific positions.

func (*RGAListState) ToFieldState 

func (l *RGAListState) ToFieldState(clock HLC, nodeID string) *FieldState

ToFieldState converts to the generic FieldState representation.

type RGANode 

type RGANode struct {
	// ID uniquely identifies this node in the list.
	ID HLC `json:"id"`
	// NodeID is the node that created this element.
	NodeID string `json:"node_id"`
	// ParentID references the node after which this element was inserted.
	// Zero HLC means this is inserted at the head of the list.
	ParentID HLC `json:"parent_id"`
	// Value is the JSON-encoded element value.
	Value json.RawMessage `json:"value"`
	// Tombstone marks this node as deleted.
	Tombstone bool `json:"tombstone,omitempty"`
}

RGANode is a single element in the RGA (Replicated Growable Array). Each node has a unique ID (NodeID + HLC), an optional parent reference for causal ordering, and a tombstone flag for deletions.

type Room 

type Room struct {
	// ID is the unique room identifier (also used as the presence topic).
	ID string `json:"id"`

	// Type classifies the room (e.g., "document", "channel", "canvas").
	Type string `json:"type,omitempty"`

	// Metadata holds arbitrary room-level data (title, permissions, etc.).
	Metadata json.RawMessage `json:"metadata,omitempty"`

	// MaxParticipants limits how many can join (0 = unlimited).
	MaxParticipants int `json:"max_participants,omitempty"`

	// CreatedAt is when the room was first created.
	CreatedAt time.Time `json:"created_at"`

	// CreatedBy is the nodeID that created the room.
	CreatedBy string `json:"created_by,omitempty"`
}

Room represents a collaboration space with metadata and participants.

type RoomEvent 

type RoomEvent struct {
	Type   RoomEventType   `json:"type"`
	RoomID string          `json:"room_id"`
	NodeID string          `json:"node_id,omitempty"`
	Data   json.RawMessage `json:"data,omitempty"`
}

RoomEvent is a rich event emitted by the RoomManager.

type RoomEventType 

type RoomEventType string

RoomEventType identifies room lifecycle events. 

const (
	RoomEventCreated          RoomEventType = "room_created"
	RoomEventDestroyed        RoomEventType = "room_destroyed"
	RoomEventParticipantJoin  RoomEventType = "participant_join"
	RoomEventParticipantLeave RoomEventType = "participant_leave"
	RoomEventMetadataUpdated  RoomEventType = "metadata_updated"
)

type RoomHook 

type RoomHook interface {
	// OnRoomCreated is called when a room is created.
	OnRoomCreated(ctx context.Context, room *Room) error
	// OnRoomDestroyed is called when a room's last participant leaves.
	OnRoomDestroyed(ctx context.Context, room *Room) error
	// OnParticipantJoin is called when a participant joins a room.
	OnParticipantJoin(ctx context.Context, room *Room, nodeID string) error
	// OnParticipantLeave is called when a participant leaves a room.
	OnParticipantLeave(ctx context.Context, room *Room, nodeID string) error
}

RoomHook is a callback for room lifecycle events.

type RoomInfo 

type RoomInfo struct {
	Room
	// ParticipantCount is the current number of participants.
	ParticipantCount int `json:"participant_count"`
	// Participants lists current participant states.
	Participants []PresenceState `json:"participants"`
}

RoomInfo is the public view of a room including live participant info.

type RoomInterceptor 

type RoomInterceptor interface {
	// BeforeRoomJoin is called before a participant joins a room.
	// Return an error to deny the join (e.g., access control).
	BeforeRoomJoin(ctx context.Context, roomID, nodeID string) error

	// AfterRoomJoin is called after a participant has joined.
	AfterRoomJoin(ctx context.Context, roomID, nodeID string) error

	// BeforeRoomLeave is called before a participant leaves.
	BeforeRoomLeave(ctx context.Context, roomID, nodeID string) error

	// AfterRoomLeave is called after a participant has left.
	AfterRoomLeave(ctx context.Context, roomID, nodeID string) error

	// OnRoomCreated is called when a new room is created.
	OnRoomCreated(ctx context.Context, room *Room) error

	// OnRoomDestroyed is called when a room is destroyed (last participant leaves).
	OnRoomDestroyed(ctx context.Context, room *Room) error
}

RoomInterceptor intercepts room lifecycle events.

type RoomManager 

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

RoomManager provides a structured API for managing presence rooms. A room is a named collaboration space (a document, a channel, a canvas) with tracked participants, metadata, and lifecycle events.

Rooms are backed by the PresenceManager for TTL-based presence, but add:

  • Room metadata (title, type, max participants, custom data)
  • Room lifecycle callbacks (created, destroyed, participant limits)
  • Participant count tracking
  • Cursor/selection position tracking helpers
  • Batch join/leave operations
  • Room listing and querying

func NewRoomManager 

func NewRoomManager(presence *PresenceManager, logger log.Logger) *RoomManager

NewRoomManager creates a room manager backed by the given presence manager.

func (*RoomManager) AddHook 

func (rm *RoomManager) AddHook(hook RoomHook)

AddHook registers a room lifecycle hook.

func (*RoomManager) Close 

func (rm *RoomManager) Close()

Close cleans up the room manager.

func (*RoomManager) CreateDocumentRoom 

func (rm *RoomManager) CreateDocumentRoom(ctx context.Context, table, pk string, opts ...RoomOption) (*Room, error)

CreateDocumentRoom creates a room for collaborating on a specific document.

func (*RoomManager) CreateRoom 

func (rm *RoomManager) CreateRoom(ctx context.Context, id, roomType string, opts ...RoomOption) (*Room, error)

CreateRoom creates or returns an existing room.

func (*RoomManager) GetDocumentParticipants 

func (rm *RoomManager) GetDocumentParticipants(table, pk string) []PresenceState

GetDocumentParticipants returns participants for a document room.

func (*RoomManager) GetRoom 

func (rm *RoomManager) GetRoom(id string) *Room

GetRoom returns a room by ID, or nil if not found.

func (*RoomManager) GetRoomInfo 

func (rm *RoomManager) GetRoomInfo(id string) *RoomInfo

GetRoomInfo returns a room with live participant info.

func (*RoomManager) JoinDocumentRoom 

func (rm *RoomManager) JoinDocumentRoom(ctx context.Context, table, pk, nodeID string, data json.RawMessage) error

JoinDocumentRoom joins the room for a specific document.

func (*RoomManager) JoinRoom 

func (rm *RoomManager) JoinRoom(ctx context.Context, roomID, nodeID string, data json.RawMessage) error

JoinRoom adds a participant to a room. Creates the room if it doesn't exist.

func (*RoomManager) LeaveAllRooms 

func (rm *RoomManager) LeaveAllRooms(ctx context.Context, nodeID string)

LeaveAllRooms removes a participant from all rooms.

func (*RoomManager) LeaveDocumentRoom 

func (rm *RoomManager) LeaveDocumentRoom(ctx context.Context, table, pk, nodeID string)

LeaveDocumentRoom leaves the room for a specific document.

func (*RoomManager) LeaveRoom 

func (rm *RoomManager) LeaveRoom(ctx context.Context, roomID, nodeID string)

LeaveRoom removes a participant from a room. If the room becomes empty, it is destroyed.

func (*RoomManager) ListRooms 

func (rm *RoomManager) ListRooms() []RoomInfo

ListRooms returns all active rooms with participant counts.

func (*RoomManager) ListRoomsByType 

func (rm *RoomManager) ListRoomsByType(roomType string) []RoomInfo

ListRoomsByType returns rooms filtered by type.

func (*RoomManager) ParticipantCount 

func (rm *RoomManager) ParticipantCount(roomID string) int

ParticipantCount returns the number of participants in a room.

func (*RoomManager) SetRoomMetadata 

func (rm *RoomManager) SetRoomMetadata(roomID string, metadata any) error

SetRoomMetadata updates the metadata for a room.

func (*RoomManager) UpdateCursor 

func (rm *RoomManager) UpdateCursor(roomID, nodeID string, cursor CursorPosition)

UpdateCursor is a convenience method to update a participant's cursor position.

func (*RoomManager) UpdateTypingStatus 

func (rm *RoomManager) UpdateTypingStatus(roomID, nodeID string, isTyping bool)

UpdateTypingStatus is a convenience method to update a participant's typing state.

type RoomOption 

type RoomOption func(*Room)

RoomOption configures a Room during creation.

func WithMaxParticipants 

func WithMaxParticipants(maxCount int) RoomOption

WithMaxParticipants sets the maximum number of participants.

func WithRoomCreator 

func WithRoomCreator(nodeID string) RoomOption

WithRoomCreator sets who created the room.

func WithRoomMetadata 

func WithRoomMetadata(metadata any) RoomOption

WithRoomMetadata sets initial room metadata.

func WithRoomType 

func WithRoomType(t string) RoomOption

WithRoomType sets the room type.

type Rows 

type Rows interface {
	Next() bool
	Scan(dest ...any) error
	Close() error
	Err() error
}

Rows is an iterator over query results.

type SetOp 

type SetOp string

SetOp is the type of set operation. 

const (
	SetOpAdd    SetOp = "add"
	SetOpRemove SetOp = "remove"
)

type SetOperation 

type SetOperation struct {
	Op       SetOp           `json:"op"`       // "add" or "remove"
	Elements json.RawMessage `json:"elements"` // JSON array of elements

	// Tags names the observed add-tags a remove is deleting (exact
	// observed-remove semantics). Removes without tags fall back to
	// removing every local tag older than the op's HLC.
	Tags []Tag `json:"tags,omitempty"`
}

SetOperation represents an add or remove operation on an OR-Set.

type State 

type State struct {
	// Table is the source table name.
	Table string `json:"table"`

	// PK is the string-encoded primary key.
	PK string `json:"pk"`

	// Fields maps field name → field-level CRDT state.
	Fields map[string]*FieldState `json:"fields"`

	// Tombstone is true if the record has been deleted.
	Tombstone bool `json:"tombstone"`

	// TombstoneHLC is the clock value of the delete operation (if tombstoned).
	TombstoneHLC HLC `json:"tombstone_hlc,omitempty"`
}

State holds the full CRDT state for a single record (all fields).

func NewState 

func NewState(table, pk string) *State

NewState creates an empty State for the given table and primary key.

func (*State) Compact 

func (st *State) Compact(before HLC) int

Compact walks every field of a record state, compacting the types that accumulate tombstones. Returns the total units dropped.

type StreamingOption 

type StreamingOption func(*StreamingTransport)

StreamingOption configures a StreamingTransport.

func WithStreamLogger 

func WithStreamLogger(l log.Logger) StreamingOption

WithStreamLogger sets the logger for the streaming transport.

func WithStreamReconnect 

func WithStreamReconnect(d time.Duration) StreamingOption

WithStreamReconnect sets the delay before reconnecting after a disconnection from the SSE stream. Defaults to 5 seconds.

func WithStreamTables 

func WithStreamTables(tables ...string) StreamingOption

WithStreamTables restricts which tables the SSE stream subscribes to.

type StreamingTransport 

type StreamingTransport struct {
	*HTTPClient // Embeds for Pull/Push.
	// contains filtered or unexported fields
}

StreamingTransport wraps an HTTPClient and adds SSE streaming for real-time change propagation. It satisfies the Transport interface for pull/push operations and additionally supports StreamChanges for SSE-based real-time sync.

Use NewStreamingTransport to create one: 

t := crdt.NewStreamingTransport("https://cloud.example.com/sync",
    crdt.WithStreamTables("documents"),
    crdt.WithStreamReconnect(5 * time.Second),
)

// Use as a Transport for pull/push:
syncer := crdt.NewSyncer(plugin, crdt.WithTransport(t))

// Or stream changes in real-time:
go t.StreamChanges(ctx, since, func(change crdt.ChangeRecord) {
    // process each change as it arrives
})

func NewStreamingTransport 

func NewStreamingTransport(baseURL string, opts ...StreamingOption) *StreamingTransport

NewStreamingTransport creates a streaming transport that supports both pull/push (via embedded HTTPClient) and SSE streaming for real-time changes.

The baseURL should point to the sync server root (e.g., "https://cloud.example.com/sync"). The SSE endpoint is assumed to be at baseURL + "/stream".

func (*StreamingTransport) StreamChanges 

func (t *StreamingTransport) StreamChanges(ctx context.Context, since HLC, handler func(ChangeRecord)) error

StreamChanges connects to the remote SSE endpoint and processes changes in real-time. It blocks until the context is cancelled. On disconnection, it automatically reconnects after the configured delay.

The handler function is called for each ChangeRecord received from the SSE stream. The since parameter specifies the starting HLC; subsequent reconnections use the latest HLC received.

type SyncController 

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

SyncController handles CRDT sync operations. It provides handlers for pull, push, and streaming endpoints that can be registered with any router (Forge, net/http, chi, etc.).

For Forge apps, use grove/extension.WithCRDT() to auto-register routes. For standalone use, call NewHTTPHandler() to get an http.Handler.

func NewSyncController 

func NewSyncController(plugin *Plugin, opts ...SyncControllerOption) *SyncController

NewSyncController creates a new sync controller for the given plugin.

func (*SyncController) AddPlugin 

func (c *SyncController) AddPlugin(p CRDTPlugin)

AddPlugin registers a CRDT plugin for intercepting operations. Plugins are called in registration order. The plugin only needs to implement the interceptor interfaces it cares about.

func (*SyncController) Close 

func (c *SyncController) Close()

Close cleans up the controller's resources (presence manager, etc.).

func (*SyncController) HandleFieldHistory 

func (c *SyncController) HandleFieldHistory(ctx context.Context, req *FieldHistoryRequest) (*FieldHistoryResponse, error)

HandleFieldHistory returns the change history of a specific field.

func (*SyncController) HandleGetPresence 

func (c *SyncController) HandleGetPresence(_ context.Context, topic string) (*PresenceSnapshot, error)

HandleGetPresence returns a snapshot of all active presence for a topic.

func (*SyncController) HandleHistory 

func (c *SyncController) HandleHistory(ctx context.Context, req *HistoryRequest) (*HistoryResponse, error)

HandleHistory returns the state of a record at a specific point in time.

func (*SyncController) HandlePresenceUpdate 

func (c *SyncController) HandlePresenceUpdate(ctx context.Context, update *PresenceUpdate) (*PresenceEvent, error)

HandlePresenceUpdate processes a presence update and returns the resulting event. Returns nil if presence is not enabled.

func (*SyncController) HandlePull 

func (c *SyncController) HandlePull(ctx context.Context, req *PullRequest) (*PullResponse, error)

HandlePull processes a pull request and returns changes since the given HLC. This is the core logic used by both Forge and HTTP handlers.

func (*SyncController) HandlePush 

func (c *SyncController) HandlePush(ctx context.Context, req *PushRequest) (*PushResponse, error)

HandlePush processes a push request, merging remote changes locally. This is the core logic used by both Forge and HTTP handlers.

func (*SyncController) Logger 

func (c *SyncController) Logger() log.Logger

Logger returns the controller's logger.

func (*SyncController) Metrics 

func (c *SyncController) Metrics() *Metrics

Metrics returns the metrics collector, or nil if metrics are disabled.

func (*SyncController) PluginChain 

func (c *SyncController) PluginChain() *PluginChain

PluginChain returns the plugin chain for inspection or testing.

func (*SyncController) Presence 

func (c *SyncController) Presence() *PresenceManager

Presence returns the presence manager, or nil if presence is disabled.

func (*SyncController) PresenceChannel 

func (c *SyncController) PresenceChannel() <-chan PresenceEvent

PresenceChannel returns the channel for receiving presence events to broadcast over SSE streams. Returns nil if presence is disabled.

func (*SyncController) Rooms 

func (c *SyncController) Rooms() *RoomManager

Rooms returns the room manager, or nil if room management is disabled.

func (*SyncController) StreamChangesSince 

func (c *SyncController) StreamChangesSince(ctx context.Context, tables []string, since HLC) (<-chan []ChangeRecord, error)

StreamChangesSince returns a channel that yields new changes as they appear. The caller should poll or watch for changes. This is used by SSE handlers.

func (*SyncController) TimeTravelEnabled 

func (c *SyncController) TimeTravelEnabled() bool

TimeTravelEnabled returns true if the time-travel feature is enabled.

type SyncControllerOption 

type SyncControllerOption func(*SyncController)

SyncControllerOption configures a SyncController.

func WithControllerPlugin 

func WithControllerPlugin(plugin CRDTPlugin) SyncControllerOption

WithControllerPlugin registers a CRDT plugin on the sync controller. Plugins intercept merge operations, metadata reads/writes, presence, room events, time-travel queries, and client connections.

Example: 

type AuditPlugin struct { crdt.BaseCRDTPlugin }
func (p *AuditPlugin) Name() string { return "audit" }
func (p *AuditPlugin) AfterMerge(ctx context.Context, ev *crdt.MergeEvent) error {
    log.Printf("merge: %s/%s field=%s winner=%s", ev.Table, ev.PK, ev.Field, ev.WinnerNodeID)
    return nil
}

crdt.WithControllerPlugin(&AuditPlugin{})

func WithControllerSyncHook 

func WithControllerSyncHook(hook SyncHook) SyncControllerOption

WithControllerSyncHook adds a sync hook to the controller. These hooks are called in addition to any plugin-level hooks.

func WithMetrics 

func WithMetrics() SyncControllerOption

WithMetrics enables metrics collection on the sync controller.

func WithPresenceBufferSize 

func WithPresenceBufferSize(size int) SyncControllerOption

WithPresenceBufferSize sets the buffer size for the presence event broadcast channel. When the buffer is full, new events are dropped. Defaults to 256.

func WithPresenceEnabled 

func WithPresenceEnabled(enabled bool) SyncControllerOption

WithPresenceEnabled enables the presence subsystem on the controller. When enabled, the controller creates an in-memory PresenceManager for ephemeral awareness data (typing indicators, cursors, user info). Disabled by default — zero overhead when not used.

func WithPresenceTTL 

func WithPresenceTTL(d time.Duration) SyncControllerOption

WithPresenceTTL sets the TTL for presence entries. After this duration without a heartbeat, presence entries are automatically removed and a "leave" event is broadcast. Defaults to 30 seconds.

func WithRoomManager 

func WithRoomManager(enabled bool) SyncControllerOption

WithRoomManager enables the room management subsystem on the controller. Requires presence to be enabled. Provides structured room lifecycle, participant tracking, cursor position tracking, and room listing.

func WithStreamKeepAlive 

func WithStreamKeepAlive(d time.Duration) SyncControllerOption

WithStreamKeepAlive sets the interval for SSE keep-alive comments. Defaults to 15 seconds.

func WithStreamPollInterval 

func WithStreamPollInterval(d time.Duration) SyncControllerOption

WithStreamPollInterval sets how frequently the SSE stream handler checks for new changes. Defaults to 1 second.

func WithTimeTravelEnabled 

func WithTimeTravelEnabled(enabled bool) SyncControllerOption

WithTimeTravelEnabled enables the opt-in time-travel feature. When enabled, the sync server exposes /history and /field-history endpoints.

func WithTimeTravelMaxDepth 

func WithTimeTravelMaxDepth(depth int) SyncControllerOption

WithTimeTravelMaxDepth sets the maximum number of history entries returned.

func WithValidation 

func WithValidation(cfg *ValidationConfig) SyncControllerOption

WithValidation enables input validation with the given config. Use DefaultValidationConfig() for sensible defaults.

type SyncFilter 

type SyncFilter struct {
	// PKFilter restricts sync to records matching these primary keys.
	PKFilter []string `json:"pk_filter,omitempty"`

	// FieldFilter restricts sync to only these fields.
	FieldFilter []string `json:"field_filter,omitempty"`
}

SyncFilter defines selective sync criteria for partial replication. When set on a PullRequest, only records matching the filter are returned.

type SyncHook 

type SyncHook interface {
	// BeforeInboundChange is called before a remote change is merged locally.
	// Return a modified change to transform it, nil to skip it, or an error to abort.
	BeforeInboundChange(ctx context.Context, change *ChangeRecord) (*ChangeRecord, error)

	// AfterInboundChange is called after a remote change has been merged locally.
	// This is useful for audit logging or triggering side effects.
	AfterInboundChange(ctx context.Context, change *ChangeRecord) error

	// BeforeOutboundChange is called before a local change is sent to a remote peer.
	// Return a modified change to transform it, nil to skip it, or an error to abort.
	BeforeOutboundChange(ctx context.Context, change *ChangeRecord) (*ChangeRecord, error)

	// BeforeOutboundRead is called before changes are returned in a pull response.
	// Receives the full slice; return a filtered or modified slice.
	BeforeOutboundRead(ctx context.Context, changes []ChangeRecord) ([]ChangeRecord, error)
}

SyncHook intercepts changes during sync operations. Implement this interface to validate, transform, filter, or audit data flowing between nodes during sync.

Use BaseSyncHook for a no-op default that you can selectively override.

type SyncHookChain 

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

SyncHookChain composes multiple SyncHooks into a sequential chain. Each hook is called in order. For Before* methods, the output of one hook becomes the input of the next. A nil return skips the change. An error return aborts the chain immediately.

func NewSyncHookChain 

func NewSyncHookChain(hooks ...SyncHook) *SyncHookChain

NewSyncHookChain creates a chain with the given hooks.

func (*SyncHookChain) Add 

func (c *SyncHookChain) Add(hook SyncHook)

Add appends a hook to the chain.

func (*SyncHookChain) AfterInboundChange 

func (c *SyncHookChain) AfterInboundChange(ctx context.Context, change *ChangeRecord) error

AfterInboundChange calls each hook in order. If any returns an error, subsequent hooks are still called (best-effort notification).

func (*SyncHookChain) BeforeInboundChange 

func (c *SyncHookChain) BeforeInboundChange(ctx context.Context, change *ChangeRecord) (*ChangeRecord, error)

BeforeInboundChange calls each hook in order. If any hook returns nil, the change is skipped. If any returns an error, the chain aborts.

func (*SyncHookChain) BeforeOutboundChange 

func (c *SyncHookChain) BeforeOutboundChange(ctx context.Context, change *ChangeRecord) (*ChangeRecord, error)

BeforeOutboundChange calls each hook in order. If any returns nil, the change is skipped. If any returns an error, the chain aborts.

func (*SyncHookChain) BeforeOutboundRead 

func (c *SyncHookChain) BeforeOutboundRead(ctx context.Context, changes []ChangeRecord) ([]ChangeRecord, error)

BeforeOutboundRead calls each hook in order. Each hook receives the output of the previous hook. An error aborts the chain.

func (*SyncHookChain) Len 

func (c *SyncHookChain) Len() int

Len returns the number of hooks in the chain.

type SyncReport 

type SyncReport struct {
	Pulled    int `json:"pulled"`
	Pushed    int `json:"pushed"`
	Merged    int `json:"merged"`
	Conflicts int `json:"conflicts"`
}

SyncReport summarizes the result of a sync operation.

func (*SyncReport) String 

func (r *SyncReport) String() string

type Syncer 

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

Syncer orchestrates the push-pull sync protocol between nodes. It supports single-peer (edge-to-cloud), multi-peer (hub-and-spoke), and P2P topologies via the Transport interface. 

syncer := crdt.NewSyncer(db, plugin,
    crdt.WithTransport(crdt.HTTPTransport("https://cloud.example.com/sync")),
    crdt.WithSyncInterval(30 * time.Second),
)
go syncer.Run(ctx)

func NewSyncer 

func NewSyncer(plugin *Plugin, opts ...SyncerOption) *Syncer

NewSyncer creates a new Syncer for the given plugin.

func (*Syncer) PushChange 

func (s *Syncer) PushChange(ctx context.Context, table, pk, field string, crdtType CRDTType, value json.RawMessage, clock HLC) error

PushChange sends a single change event to all peers. This is useful for CDC-driven sync where changes are pushed in real-time. If sync hooks are configured, BeforeOutboundChange is called before pushing.

func (*Syncer) Run 

func (s *Syncer) Run(ctx context.Context) error

Run starts a background sync loop that runs until the context is cancelled.

func (*Syncer) StreamSync 

func (s *Syncer) StreamSync(ctx context.Context) error

StreamSync connects to all peers that support SSE streaming and processes changes in real-time. This runs alongside the periodic poll-based Sync for lower latency on supported transports. Falls back gracefully if a transport does not support streaming.

Blocks until the context is cancelled. Start it in a goroutine: 

go syncer.StreamSync(ctx)

func (*Syncer) Sync 

func (s *Syncer) Sync(ctx context.Context) (*SyncReport, error)

Sync performs a single round of sync with all configured peers. For each peer: pull remote changes, merge locally, then push local changes.

type SyncerOption 

type SyncerOption func(*Syncer)

SyncerOption configures a Syncer.

func WithPeers 

func WithPeers(peers ...Transport) SyncerOption

WithPeers adds multiple peer transports for hub-and-spoke or P2P sync.

func WithRetry 

func WithRetry(attempts int, baseDelay, maxDelay time.Duration) SyncerOption

WithRetry configures retry behavior for sync operations. Defaults: 3 attempts, 1s base delay, 30s max delay.

func WithSyncInterval 

func WithSyncInterval(d time.Duration) SyncerOption

WithSyncInterval sets the interval for background sync. Defaults to 30 seconds.

func WithSyncTables 

func WithSyncTables(tables ...string) SyncerOption

WithSyncTables restricts which tables are synced.

func WithTransport 

func WithTransport(t Transport) SyncerOption

WithTransport sets the transport for sync communication.

type Tag 

type Tag struct {
	NodeID string `json:"node"`
	HLC    HLC    `json:"hlc"`
}

Tag uniquely identifies an add operation.

type TextDelta 

type TextDelta struct {
	Insert     string                     `json:"insert"`
	Attributes map[string]json.RawMessage `json:"attributes,omitempty"`
}

TextDelta is one Quill-style segment of the visible text.

type TextFragment 

type TextFragment struct {
	Origin    HLC                  `json:"origin"`
	Start     int                  `json:"start"`
	Content   string               `json:"content"`
	Length    int                  `json:"length"`
	Parent    TextRef              `json:"parent,omitempty"`
	Tombstone bool                 `json:"tombstone,omitempty"`
	Attrs     map[string]AttrState `json:"attrs,omitempty"`
}

TextFragment is a stored piece of an origin's span.

type TextOp 

type TextOp struct {
	Op      TextOpType                 `json:"op"`
	Ref     TextRef                    `json:"ref,omitempty"`
	Origin  HLC                        `json:"origin,omitempty"`
	Content string                     `json:"content,omitempty"`
	Spans   []TextSpan                 `json:"spans,omitempty"`
	Attrs   map[string]json.RawMessage `json:"attrs,omitempty"`
}

TextOp is a text operation for the sync transport. Inserts carry the content, the ref they anchor at and the Origin span they belong to (creator-decided: the op's own HLC for a new span, or the extended span's origin). Deletes and formats carry the exact address Spans they cover.

type TextOpType 

type TextOpType string

TextOpType identifies a text operation. 

const (
	TextOpInsert TextOpType = "insert"
	TextOpDelete TextOpType = "delete"
	TextOpFormat TextOpType = "format"
)

type TextRef 

type TextRef struct {
	Origin HLC `json:"origin"`
	Offset int `json:"offset"`
}

TextRef addresses one character: rune Offset within the insertion span identified by Origin. The zero TextRef is the document head.

func (TextRef) IsHead 

func (r TextRef) IsHead() bool

IsHead reports whether the ref is the document-head sentinel.

type TextSpan 

type TextSpan struct {
	Origin HLC `json:"origin"`
	Start  int `json:"start"`
	Length int `json:"length"`
}

TextSpan names a contiguous address range within one origin.

type TextState 

type TextState struct {
	// Frags maps origin key → fragments sorted by Start (disjoint ranges).
	Frags map[string][]*TextFragment `json:"frags"`
	// contains filtered or unexported fields
}

TextState holds the full text CRDT state.

func MergeText 

func MergeText(local, remote *TextState) *TextState

MergeText merges two text states: per-origin union with both sides normalized to the finest common fragment partition, then per-fragment tombstone-OR and per-attribute LWW. Commutative, associative, idempotent. Inputs are never mutated.

func NewTextState 

func NewTextState() *TextState

NewTextState creates an empty text CRDT.

func TextFromFieldState 

func TextFromFieldState(fs *FieldState) *TextState

TextFromFieldState reconstructs a TextState from a FieldState.

func (*TextState) Apply 

func (t *TextState) Apply(op *TextOp, nodeID string, clock HLC) error

Apply folds one TextOp into the state. Inserts place content at creator-chosen addresses and are idempotent (duplicate delivery is ignored). Delete/format ops touch the exact spans they name and assume CAUSAL DELIVERY per origin — they must arrive after the insert whose addresses they reference (the same assumption Yjs makes; fabriq's seq-ordered log and grove's HLC-ordered pull both satisfy it). Arbitrary-order convergence across replicas is provided by MergeText.

func (*TextState) Compact 

func (t *TextState) Compact(before HLC) int

Compact skeletonizes tombstoned fragments whose origin is older than the horizon: content is freed and adjacent skeletons coalesce, but addresses are preserved so every anchor stays resolvable. Returns the number of fragments freed or coalesced away.

func (*TextState) Delete 

func (t *TextState) Delete(ref TextRef, length int) (*TextOp, error)

Delete tombstones length visible characters starting at ref's character and returns the op (with the exact spans it resolved) to broadcast.

func (*TextState) Delta 

func (t *TextState) Delta() []TextDelta

Delta returns the visible text as attribute-run segments, merging adjacent segments with identical attributes.

func (*TextState) Format 

func (t *TextState) Format(ref TextRef, length int, attrs map[string]json.RawMessage, nodeID string, clock HLC) (*TextOp, error)

Format applies attribute writes to length visible characters starting at ref's character. A JSON null value clears the attribute. Returns the op.

func (*TextState) IndexOf 

func (t *TextState) IndexOf(ref TextRef) (int, bool)

IndexOf returns the current visible index of the character at ref. A tombstoned character collapses to the index it would occupy (the position of the next visible character) — cursor semantics.

func (*TextState) Insert 

func (t *TextState) Insert(ref TextRef, s, nodeID string, clock HLC) (*TextOp, error)

Insert inserts s after the character at ref (or at the document head for the zero ref) and returns the op to broadcast. Sequential inserts by the same node at its own span's tail extend that span (run coalescing).

func (*TextState) Len 

func (t *TextState) Len() int

Len returns the number of visible characters (runes).

func (*TextState) RefAt 

func (t *TextState) RefAt(index int) (TextRef, bool)

RefAt returns the stable address of the index-th visible character.

func (*TextState) SetString 

func (t *TextState) SetString(s, nodeID string, clock HLC) ([]*TextOp, error)

SetString reconciles the text toward the given whole string using a common prefix/suffix diff — the ORM write path for crdt:"text" fields, where the caller only has the new full value. Returns the ops emitted.

func (*TextState) ToFieldState 

func (t *TextState) ToFieldState(clock HLC, nodeID string) *FieldState

ToFieldState converts to the generic FieldState representation.

func (*TextState) Value 

func (t *TextState) Value() string

Value returns the visible text.

type TimeTravelConfig 

type TimeTravelConfig struct {
	// Enabled controls whether the time-travel endpoints are registered.
	// Defaults to false (disabled).
	Enabled bool

	// MaxHistoryDepth limits how many historical versions to return.
	// 0 means unlimited. Defaults to 100.
	MaxHistoryDepth int
}

TimeTravelConfig controls the opt-in time-travel feature. When enabled, the sync server exposes a /history endpoint for querying historical state at any point in time via HLC.

type TimeTravelInterceptor 

type TimeTravelInterceptor interface {
	// BeforeHistoryRead is called before reading historical state.
	// Return an error to deny the read (e.g., access control on history).
	BeforeHistoryRead(ctx context.Context, table, pk string, atHLC HLC) error

	// AfterHistoryRead is called after reading historical state.
	// Return a modified state for redaction or transformation.
	AfterHistoryRead(ctx context.Context, table, pk string, state *State) (*State, error)
}

TimeTravelInterceptor intercepts time-travel queries.

type Transport 

type Transport interface {
	// Pull requests changes from a remote node since the given HLC.
	Pull(ctx context.Context, req *PullRequest) (*PullResponse, error)

	// Push sends local changes to a remote node.
	Push(ctx context.Context, req *PushRequest) (*PushResponse, error)
}

Transport is the interface for sync communication between nodes. Implement this interface to support custom transport layers (WebSocket, NATS, Kafka, gRPC, etc.).

type TxExecutor 

type TxExecutor interface {
	Executor
	BeginTx(ctx context.Context) (TxHandle, error)
}

TxExecutor extends Executor with transaction support. When the underlying executor supports transactions, MetadataStore uses them to wrap multi-field writes atomically.

type TxHandle 

type TxHandle interface {
	Executor
	Commit() error
	Rollback() error
}

TxHandle represents an active transaction.

type ValidationConfig 

type ValidationConfig struct {
	// MaxChangeValueSize is the maximum byte size for a single field value (default: 1MB).
	MaxChangeValueSize int
	// MaxChangesPerPush is the maximum number of changes in a single push (default: 10000).
	MaxChangesPerPush int
	// MaxHLCDrift is the maximum allowed HLC timestamp drift from server time (default: 1 hour).
	MaxHLCDrift time.Duration
	// MaxRoomMetadataSize is the maximum byte size for room metadata (default: 100KB).
	MaxRoomMetadataSize int
	// MaxParticipantDataSize is the maximum byte size for participant presence data (default: 10KB).
	MaxParticipantDataSize int
}

ValidationConfig controls input validation for sync operations.

func DefaultValidationConfig 

func DefaultValidationConfig() *ValidationConfig

DefaultValidationConfig returns sensible defaults.

func (*ValidationConfig) ValidateChangeRecord 

func (vc *ValidationConfig) ValidateChangeRecord(change *ChangeRecord) error

ValidateChangeRecord validates a single change record.

func (*ValidationConfig) ValidatePresenceData 

func (vc *ValidationConfig) ValidatePresenceData(data []byte) error

ValidatePresenceData validates presence payload size.

func (*ValidationConfig) ValidatePushRequest 

func (vc *ValidationConfig) ValidatePushRequest(req *PushRequest) error

ValidatePushRequest validates a push request.

func (*ValidationConfig) ValidateRoomMetadata 

func (vc *ValidationConfig) ValidateRoomMetadata(data []byte) error

ValidateRoomMetadata validates room metadata payload size.

type WSMessageType 

type WSMessageType string

WSMessageType identifies the type of WebSocket message. 

const (
	WSPullRequest    WSMessageType = "pull_request"
	WSPullResponse   WSMessageType = "pull_response"
	WSPushRequest    WSMessageType = "push_request"
	WSPushResponse   WSMessageType = "push_response"
	WSChange         WSMessageType = "change"
	WSChanges        WSMessageType = "changes"
	WSPresenceUpdate WSMessageType = "presence_update"
	WSPresenceEvent  WSMessageType = "presence_event"
	WSPresenceGet    WSMessageType = "presence_get"
	WSPresenceSnap   WSMessageType = "presence_snapshot"
	WSSubscribe      WSMessageType = "subscribe"
	WSUnsubscribe    WSMessageType = "unsubscribe"
	WSError          WSMessageType = "error"
	WSPing           WSMessageType = "ping"
	WSPong           WSMessageType = "pong"
)

type WebSocketConn 

type WebSocketConn interface {
	// ReadMessage reads the next message from the connection.
	ReadMessage() ([]byte, error)
	// WriteMessage sends a message over the connection.
	WriteMessage(data []byte) error
	// Close closes the connection.
	Close() error
}

WebSocketConn is the interface that any WebSocket implementation must satisfy. This allows plugging in gorilla/websocket, nhooyr.io/websocket, or any other library.

type WebSocketDialer 

type WebSocketDialer func(ctx context.Context) (WebSocketConn, error)

WebSocketDialer is a function that establishes a new WebSocket connection. Used by StartWithReconnect to re-establish connections after disconnection.

type WebSocketHandler 

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

WebSocketHandler processes WebSocket messages on the server side. It wraps a SyncController and handles pull/push/subscribe requests received over WebSocket.

func NewWebSocketHandler 

func NewWebSocketHandler(ctrl *SyncController, conn WebSocketConn, logger log.Logger) *WebSocketHandler

NewWebSocketHandler creates a new server-side WebSocket handler.

func (*WebSocketHandler) Serve 

func (h *WebSocketHandler) Serve(ctx context.Context) error

Serve reads messages from the WebSocket connection and processes them. Call this in a goroutine per client connection.

type WebSocketMessage 

type WebSocketMessage struct {
	Type      WSMessageType   `json:"type"`
	Payload   json.RawMessage `json:"payload"`
	RequestID string          `json:"request_id,omitempty"`
}

WebSocketMessage is the JSON framing format for multiplexed WebSocket messages. When using the binary protobuf format, use the proto WebSocketFrame instead.

type WebSocketOption 

type WebSocketOption func(*WebSocketTransport)

WebSocketOption configures a WebSocketTransport.

func WithWSLogger 

func WithWSLogger(l log.Logger) WebSocketOption

WithWSLogger sets the logger.

func WithWSPingInterval 

func WithWSPingInterval(d time.Duration) WebSocketOption

WithWSPingInterval sets the ping/keepalive interval.

func WithWSReconnectDelay 

func WithWSReconnectDelay(d time.Duration) WebSocketOption

WithWSReconnectDelay sets the reconnect delay.

func WithWSTables 

func WithWSTables(tables ...string) WebSocketOption

WithWSTables restricts which tables to subscribe to.

type WebSocketTransport 

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

WebSocketTransport implements the Transport interface over WebSocket connections. It provides bidirectional multiplexed communication for pull, push, presence, and real-time change streaming over a single connection.

func NewWebSocketTransport 

func NewWebSocketTransport(conn WebSocketConn, opts ...WebSocketOption) *WebSocketTransport

NewWebSocketTransport creates a new WebSocket-based transport.

func (*WebSocketTransport) Close 

func (t *WebSocketTransport) Close() error

Close closes the WebSocket connection.

func (*WebSocketTransport) OnChange 

func (t *WebSocketTransport) OnChange(handler func(ChangeRecord))

OnChange registers a handler for incoming change events.

func (*WebSocketTransport) OnPresence 

func (t *WebSocketTransport) OnPresence(handler func(PresenceEvent))

OnPresence registers a handler for incoming presence events.

func (*WebSocketTransport) Pull 

func (t *WebSocketTransport) Pull(ctx context.Context, req *PullRequest) (*PullResponse, error)

Pull requests changes from the remote node via WebSocket.

func (*WebSocketTransport) Push 

func (t *WebSocketTransport) Push(ctx context.Context, req *PushRequest) (*PushResponse, error)

Push sends local changes to the remote node via WebSocket.

func (*WebSocketTransport) Start 

func (t *WebSocketTransport) Start(ctx context.Context) error

Start begins the read loop for processing incoming messages. Call this in a goroutine after creating the transport.

func (*WebSocketTransport) StartWithReconnect 

func (t *WebSocketTransport) StartWithReconnect(ctx context.Context, dial WebSocketDialer) error

StartWithReconnect begins the read loop and automatically reconnects on disconnection using the provided dialer. It blocks until the context is cancelled. Between reconnection attempts, it waits for reconnectDelay.

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.