subscribe

type Engine ¶

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

Engine is the per-space live-query engine. Cheap when no subs are registered: HasSubscribers is a single atomic load and the afterApply hook short-circuits on it.

func New(spaceId string) *Engine

func (e *Engine) Close() error

func (e *Engine) HasSubscribers() bool

func (e *Engine) NotifyDeleted(spaceId, dataset, objectId string)

func (e *Engine) OnApply(ev Event, postValue PostValueFn)

func (e *Engine) SpaceId() string

func (e *Engine) Subscribe(cfg SubConfig, snapshot SnapshotFn) (*Sub, error)

type Event ¶

type Event struct {
	SpaceId   string
	ObjectId  string
	Dataset   string
	VersionId crdt.VersionId
	Records   []EventRecord
}

Event is a single CRDT apply event handed to the engine.

Pipeline contract: receive change → apply to CRDT → commit the any-store tx → THEN build this event. By the time a consumer sees an Event, every projected $set/$unset is already durable in the controller's any-store; a Query against the same dataset run from the same process reflects the same state.

Carries the routing tuple plus enough about the change for the engine to classify each record (filter + sort + window membership) without re-querying:

• VersionId — per-change DAG order. Forwarded on the wire as SubscriptionEvent.VersionId so fence-and-replay consumers can dedupe across snapshots.

• Records — the post-apply effect of the change, projected to a flat list of $set / $unset ops per record. Ops payloads are deep-cloned off the controller's storage onto event-owned arenas, so an Event is safe to retain past the build call.

func BuildEvent(ch *crdt.Change, recordIds []string, derivedOps [][]crdt.Op, postValue PostValueFn) Event

type EventRecord ¶

type EventRecord struct {
	Id string
	// Created is true when this change first materialised the record.
	// The engine uses it (combined with sentinel/visibility state) to
	// decide Added vs Updated emit semantics.
	Created bool
	// Deleted is true when the change tombstoned this record. Ops is
	// empty in that case; the engine drops the entry from its held
	// set.
	Deleted bool
	// Ops is the projected $set / $unset operations on the post-apply
	// record. Empty when Deleted is true. Surfaced verbatim to
	// SubscriptionEvent SubRecord.Ops for atomic-update consumers.
	Ops []space.EventOp
}

EventRecord is one record's worth of projected change inside an Event. Id is the record id within Dataset (for shared per-space datasets like "objects" this equals ObjectId).

type PostValueFn ¶

type PostValueFn func(recordIndex int) *anyenc.Value

PostValueFn returns the post-apply value for one record in the change being built. Provided by the per-space layer (the spaceobjects Store holds the controller). Index is into ch.Records.

May return nil for tombstoned / dropped records (e.g. a gated op that didn't land); callers should treat nil as "the record no longer exists at this point in the timeline".

type Scope ¶

type Scope struct {
	Shared   bool
	ObjectId string
	Dataset  string
}

Scope locates which events a sub cares about.

Shared=true means "every event whose dataset == ObjectsDataset across every object in the space" — the property firehose. The ObjectId/Dataset fields are ignored in this case.

Shared=false means "events whose ObjectId/Dataset match exactly".

type SnapshotFn ¶

type SnapshotFn func(yield func(id string, doc *anyenc.Value)) error

SnapshotFn is the caller-supplied populator. It runs UNDER engine.mu (held by Subscribe), so apply events fired during the read block waiting on engine.mu and process after the new sub is registered. The callback yields each result row's (id, post-doc) to the engine, which extracts the sort tuple via cfg.Sort and inserts the entry.

The callback must only read already-materialised rows. Anything that can apply a change — loading an object (its replay runs OnApply), writing to the DAG — needs engine.mu itself and deadlocks the engine. Resolve objects before calling Subscribe.

type Sub ¶

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

Sub is the consumer-side handle returned by Engine.Subscribe. It implements space.QuerySubscription. The engine guarantees the mailbox carries only post-snapshot transitions (no replay of the snapshot itself).

func (s *Sub) Close() error

func (s *Sub) Err() error

func (s *Sub) Events() *mb.MB[space.SubscriptionEvent]

type SubConfig ¶

type SubConfig struct {
	Scope       Scope
	Filter      query.Filter // already ANDed with `_deletedAt missing`
	Sort        query.Sort   // required when Limit > 0
	Limit       int          // 0 = unbounded; no sentinel, no fast-reject
	MailboxCap  int          // 0 → defaultMailboxCap; clamped to >= minMailboxCap
	DriftBudget int          // percent of limit; 0 → defaultDriftBudgetPercent; ignored when Limit == 0
}

SubConfig is the input to Engine.Subscribe. Built by queryImpl from the chained Filter / Sort / Limit + caller's QueryOpts. The engine owns the parsed query.Filter / query.Sort after registration; the caller must not mutate them.