livequery

type AuthzFunc func(ctx context.Context, q LiveQuery) error

type Change struct {
	AggID    string
	Version  int64
	Deleted  bool
	Vals     map[string]any // nil when Deleted
	Raw      json.RawMessage
	StreamID string
	At       time.Time
}

type ChannelFor func(ctx context.Context, q LiveQuery) (channel string, err error)

type Cursor struct {
	Values []any `json:"values"`
}

type DeltaOp int

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

type EngineOptions struct {
	Cushion int
	Buffer  int
	// Members optionally seeds Streamed subscriptions' membership from the full
	// matching id set; nil falls back to the snapshot page.
	Members MemberLister
}

type Feed interface {
	Changes(ctx context.Context, q LiveQuery, fromWatermark string) (<-chan Change, func(), error)
}

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

type LiveDelta struct {
	Op       DeltaOp         `json:"op"`
	AggID    string          `json:"agg_id,omitempty"`
	Version  int64           `json:"version,omitempty"`
	Row      json.RawMessage `json:"row,omitempty"`
	OldIndex int             `json:"old_index"`
	NewIndex int             `json:"new_index"`
	Cursor   Cursor          `json:"cursor,omitempty"`
	StreamID string          `json:"stream_id,omitempty"`
	At       time.Time       `json:"at"`
}

type LiveQuery struct {
	Entity string
	Where  query.Where
	Sort   []SortKey
	Limit  int // window size N
	Cursor *Cursor
	Mode   Mode
}

type MemberLister interface {
	Members(ctx context.Context, q LiveQuery) ([]string, error)
}

type Mode int

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

type Refiller interface {
	After(ctx context.Context, q LiveQuery, after Cursor, limit int) ([]Row, error)
}

type Registration struct {
	SubID     string
	TenantID  string
	Entity    string
	Partition int // data partition (cluster.PartitionOf); set by the caller
	Mode      Mode
	Query     LiveQuery
	GatewayID string // the gateway terminating this subscription's connection
	Watermark string // last delivered stream position, for gapless resume
}

type Row struct {
	AggID   string
	Version int64
	Cursor  Cursor
	Raw     json.RawMessage
	Vals    map[string]any
}

type Snapshot struct {
	SubID     string
	Rows      []Row
	Watermark string // event-stream id at snapshot time; live applied strictly after
}

type Snapshotter interface {
	Snapshot(ctx context.Context, q LiveQuery, limit int) ([]Row, error)
}

type SortKey struct {
	Column string
	Desc   bool
}

type SubscriptionRegistry interface {
	// Put records or updates a subscription (idempotent on SubID).
	Put(ctx context.Context, r Registration) error
	// Delete removes a subscription on clean unsubscribe.
	Delete(ctx context.Context, subID string) error
	// ByPartition lists the subscriptions for a (tenant, entity) — convenience
	// for single-entity inspection.
	ByPartition(ctx context.Context, tenantID, entity string) ([]Registration, error)
	// ByPartitionNum lists every subscription in data partition p, across all
	// tenants/entities — the query a reassigned shard runs to rebuild what it
	// now owns after a failover or rebalance.
	ByPartitionNum(ctx context.Context, p int) ([]Registration, error)
	// ByGateway lists a gateway's subscriptions — for gateway recovery.
	ByGateway(ctx context.Context, gatewayID string) ([]Registration, error)
}

type Tailer interface {
	Tail(ctx context.Context, channel, fromID string, deliver func(query.Delta)) error
}

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