storage

package
v0.30.38 Latest Latest
Warning

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

 Go to latest Published: Sep 7, 2026 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Index

Constants

View Source 
const DefaultStmtCacheSize = 256

DefaultStmtCacheSize is the maximum number of prepared statements kept in the cache before LRU eviction. Each entry holds a *sql.Stmt against the reader pool. For a typical xolu workload the number of distinct SQL shapes is bounded (tens, not thousands), so 256 is generous while keeping memory bounded.

Variables

View Source 
var (
	// ErrNotFound is returned when an entity is not found
	ErrNotFound = errors.New("entity not found")
	// ErrAlreadyExists is returned when an entity already exists
	ErrAlreadyExists = errors.New("entity already exists")
	// ErrInvalidEntity is returned when entity name is invalid
	ErrInvalidEntity = errors.New("invalid entity name")
	// ErrInvalidID is returned when ID is invalid
	ErrInvalidID = errors.New("invalid ID")
	// ErrConflict is returned when an optimistic concurrency check fails
	ErrConflict = errors.New("version conflict")
	// ErrNotSupported is returned when a storage backend does not implement
	// a particular operation. Handlers that receive this error should map it
	// to an appropriate HTTP 501 Not Implemented response.
	ErrNotSupported = errors.New("operation not supported by this storage backend")
)

Functions

func DenormaliseDecimalColumns 

func DenormaliseDecimalColumns(spec *AdaptedTableSpec, dialect StorageDialect, colVals []interface{})

DenormaliseDecimalColumns applies dialect-specific decimal denormalisation to column values read from the database. This converts scaled integers back to client-facing decimal strings. Called on the read path (get, list) after SQL scan.

func GenerateCreateTableSQL 

func GenerateCreateTableSQL(spec *AdaptedTableSpec, dialect StorageDialect) string

GenerateCreateTableSQL produces the CREATE TABLE statement for an adapted table using the given dialect.

func GenerateIndexSQL 

func GenerateIndexSQL(spec *AdaptedTableSpec, dialect StorageDialect) []string

GenerateIndexSQL produces CREATE INDEX statements for the adapted table using the given dialect.

func ListStores 

func ListStores() []string

ListStores returns all registered store types

func MigrateAdaptedTable 

func MigrateAdaptedTable(
	ctx context.Context,
	db *sql.DB,
	registry *AdaptedRegistry,
	entity string,
	newSchema map[string]interface{},
	dialect StorageDialect,
) error

MigrateAdaptedTable applies schema changes to an existing adapted table. It computes the diff between the stored spec and the new schema, then:

  1. Rejects type changes (incompatible, require manual intervention)
  2. Adds new columns via ALTER TABLE ADD COLUMN
  3. Drops removed columns via ALTER TABLE DROP COLUMN (SQLite 3.35+), after migrating any existing data to the _extra overflow column
  4. Updates indexes (drop old, create new)
  5. Updates the metadata row in the per-tenant t<X>_n_sch registry
  6. Updates the in-memory registry

The entire migration runs in a single transaction.

func NormaliseDecimalColumns 

func NormaliseDecimalColumns(spec *AdaptedTableSpec, dialect StorageDialect, colVals []interface{}) error

NormaliseDecimalColumns applies dialect-specific decimal normalisation to column values produced by PartitionData. This transforms validated decimal strings into scaled integers for storage. Called on the write path (create, update) before SQL execution.

func PartitionData 

func PartitionData(spec *AdaptedTableSpec, data map[string]interface{}) (columnValues []interface{}, extra map[string]interface{})

PartitionData separates a data map into schema-column values and overflow. Returns:

  • columnValues: ordered values matching spec.Columns, ready for INSERT
  • extra: map of fields not in the schema (nil if none or !hasExtra)

REF fields are decomposed: {"type":"REF","entity":"users","id":42} becomes two column values: "users" (for REF_{field}_entity) and 42 (for REF_{field}_id).

func ReassembleData 

func ReassembleData(spec *AdaptedTableSpec, columnValues []interface{}, extra map[string]interface{}, id int, version int) map[string]interface{}

ReassembleData reconstructs a map[string]interface{} from column values and an optional overflow map. This is the inverse of PartitionData.

func RegisterAdaptedTable 

func RegisterAdaptedTable(ctx context.Context, db *sql.DB, registry *AdaptedRegistry, entity string, schema map[string]interface{}, dialect StorageDialect, tenantID tenant.TenantID) error

RegisterAdaptedTable derives a table spec from a JSON Schema, creates the table and indexes in the database, and records the spec in the per-tenant t<X>_n_sch metadata table.

If the table already exists with the same schema hash, this is a no-op. If the schema has changed, the caller must handle migration separately (Phase 4 of the design).

func RegisterStore 

func RegisterStore(name string, factory StoreFactory)

RegisterStore registers a new store implementation

func SeqIncrementTx 

func SeqIncrementTx(ctx context.Context, tx *sql.Tx, tenantID tenant.TenantID, name string) (int64, error)

SeqIncrementTx performs an atomic named-sequence increment on a caller's transaction and returns the new current value, without committing. It is the tx-scoped core shared by the standalone /seq/{name}/next path and by FSM set clauses that contain NEXT VALUE FOR. On an exhausted non-cyclic sequence, or a missing sequence, it returns sql.ErrNoRows.

Types

type AdaptedEdgeStore 

type AdaptedEdgeStore interface {
	// RegisterAdaptedEdge derives a native-column table for the given
	// relationship label from its JSON Schema, creates the table and indexes
	// in the database, and records the adapted spec in t<X>_e_sch.
	// The label must already be registered in t<X>_e_sch via
	// RegisterEdgeSchema; if it is not, RegisterAdaptedEdge registers it
	// implicitly.
	RegisterAdaptedEdge(ctx context.Context, rel string, schema map[string]interface{}) error

	// IsEdgeAdapted reports whether rel has an adapted table spec in the
	// in-memory registry (i.e. whether edge properties for this label will
	// be stored in t<X>_edata_<label> rather than t<X>_edges).
	IsEdgeAdapted(rel string) bool
}

AdaptedEdgeStore is an optional interface implemented by backends that support adapted (native-column) tables for edge properties.

type AdaptedRegistry 

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

AdaptedRegistry tracks which entity types have adapted tables.

func LoadAdaptedRegistry 

func LoadAdaptedRegistry(ctx context.Context, db *sql.DB, tenantID tenant.TenantID) (*AdaptedRegistry, error)

LoadAdaptedRegistry reads the per-tenant t<X>_n_sch metadata table and populates the registry. Called at store startup.

func NewAdaptedRegistry 

func NewAdaptedRegistry() *AdaptedRegistry

NewAdaptedRegistry creates an empty registry.

func (*AdaptedRegistry) Entities 

func (r *AdaptedRegistry) Entities() []string

Entities returns a sorted list of all adapted entity types.

func (*AdaptedRegistry) Get 

func (r *AdaptedRegistry) Get(entity string) *AdaptedTableSpec

Get returns the adapted table spec for an entity, or nil if the entity uses blob storage.

func (*AdaptedRegistry) IsAdapted 

func (r *AdaptedRegistry) IsAdapted(entity string) bool

IsAdapted reports whether an entity type has an adapted table.

func (*AdaptedRegistry) Set 

func (r *AdaptedRegistry) Set(entity string, spec *AdaptedTableSpec)

Set registers an adapted table spec for an entity.

type AdaptedTableSpec 

type AdaptedTableSpec struct {
	Entity     string             `json:"entity"`      // Entity/relationship label name
	Kind       tenant.ElementKind `json:"kind"`        // ElementNode or ElementEdge
	TenantID   tenant.TenantID    `json:"tenant_id"`   // Owning tenant; used to derive the table name
	Columns    []ColumnDef        `json:"columns"`     // Ordered column definitions
	SchemaHash string             `json:"schema_hash"` // SHA-256 of canonical schema JSON
	HasExtra   bool               `json:"has_extra"`   // Whether _extra overflow column is present
	Indexes    []IndexDef         `json:"indexes"`     // Indexes to create
}

AdaptedTableSpec describes the full column layout of an adapted table.

func DeriveAdaptedTableSpec 

func DeriveAdaptedTableSpec(entity string, schema map[string]interface{}, dialect StorageDialect, tenantID tenant.TenantID) (*AdaptedTableSpec, error)

DeriveAdaptedTableSpec examines a JSON Schema document and produces a complete AdaptedTableSpec describing the adapted table layout.

The dialect parameter determines backend-specific column types.

This is a convenience wrapper that creates a SchemaIntrospector from the raw JSON Schema map. For direct use with queryfy (future), call DeriveAdaptedTableSpecFrom with a queryfy-backed introspector.

func DeriveAdaptedTableSpecFrom 

func DeriveAdaptedTableSpecFrom(entity string, schema SchemaIntrospector, dialect StorageDialect, schemaHash string, tenantID tenant.TenantID) (*AdaptedTableSpec, error)

DeriveAdaptedTableSpecFrom derives an AdaptedTableSpec from a SchemaIntrospector. This is the backend-agnostic core that works with any schema representation (JSON Schema maps, queryfy objects, or anything else that implements SchemaIntrospector).

func (*AdaptedTableSpec) ColumnByName 

func (s *AdaptedTableSpec) ColumnByName(name string) (ColumnDef, bool)

ColumnByName returns the ColumnDef for a given SQL column name.

func (*AdaptedTableSpec) ColumnNames 

func (s *AdaptedTableSpec) ColumnNames() []string

ColumnNames returns all column names in order (excluding system columns).

func (*AdaptedTableSpec) FieldToColumn 

func (s *AdaptedTableSpec) FieldToColumn(jsonField string) []string

FieldToColumn maps a JSON field name to its column name(s). For REF fields, this returns two names: REF_{field}_entity, REF_{field}_id. For all other fields, it returns a single name equal to the field name.

func (*AdaptedTableSpec) IsSchemaField 

func (s *AdaptedTableSpec) IsSchemaField(jsonField string) bool

IsSchemaField reports whether a JSON field name is a declared schema field.

func (*AdaptedTableSpec) TableName 

func (s *AdaptedTableSpec) TableName() string

TableName returns the SQL table name for this adapted table. Routes to the node or edge naming convention based on Kind:

  • ElementNode → t<XXXX>_ndata_<entity> (e.g. t0001_ndata_user)
  • ElementEdge → t<XXXX>_edata_<label> (e.g. t0001_edata_KNOWS)

type AdaptiveLock 

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

AdaptiveLock provides a mutex that automatically engages under contention.

Under normal load, operations proceed without any Go-side locking, relying on SQLite's WAL mode and busy_timeout for concurrency control. When the write success rate drops below a configurable threshold (indicating heavy contention), the lock engages a sync.RWMutex to serialise writes and eliminate SQLITE_BUSY errors entirely. When contention subsides, the lock disengages and returns to lock-free operation.

This gives the best of both worlds: full WAL concurrency under normal load, and guaranteed zero errors under burst load.

func NewAdaptiveLock 

func NewAdaptiveLock(threshold int) *AdaptiveLock

NewAdaptiveLock creates an adaptive lock with the given success-rate threshold. threshold is expressed as a percentage (e.g. 95 means engage the mutex when the success rate drops below 95%). Valid range: 0-100. 0 disables the lock entirely; 100 keeps it permanently engaged (equivalent to a plain mutex).

func (*AdaptiveLock) Engaged 

func (al *AdaptiveLock) Engaged() bool

Engaged returns whether the mutex is currently engaged.

func (*AdaptiveLock) Lock 

func (al *AdaptiveLock) Lock() bool

Lock acquires a write lock if the adaptive lock is engaged. Returns true if the lock was acquired (caller must call Unlock).

func (*AdaptiveLock) RLock 

func (al *AdaptiveLock) RLock() bool

RLock acquires a read lock if the adaptive lock is engaged. Returns true if the lock was acquired (caller must call RUnlock).

func (*AdaptiveLock) RUnlock 

func (al *AdaptiveLock) RUnlock()

RUnlock releases the read lock. Only call if RLock returned true.

func (*AdaptiveLock) RecordFailure 

func (al *AdaptiveLock) RecordFailure()

RecordFailure records a failed (SQLITE_BUSY) operation. If the threshold is non-zero, this immediately engages the lock — the burst is happening now and waiting for the monitor tick would let more failures through. The monitor goroutine handles disengagement once the window is clean.

func (*AdaptiveLock) RecordSuccess 

func (al *AdaptiveLock) RecordSuccess()

RecordSuccess records a successful operation.

func (*AdaptiveLock) SetThreshold 

func (al *AdaptiveLock) SetThreshold(threshold int)

SetThreshold dynamically updates the success-rate threshold at runtime. threshold is a percentage (0-100). Takes effect on the next monitor tick.

func (*AdaptiveLock) Stop 

func (al *AdaptiveLock) Stop()

Stop terminates the background monitor goroutine.

func (*AdaptiveLock) Threshold 

func (al *AdaptiveLock) Threshold() int

Threshold returns the current threshold as a percentage (0-100).

func (*AdaptiveLock) Unlock 

func (al *AdaptiveLock) Unlock()

Unlock releases the write lock. Only call if Lock returned true.

type AggregateQueryable 

type AggregateQueryable interface {
	// AggregateQuery executes a pre-built aggregate SQL query and returns
	// the grouped results as maps. Unlike QueryWithPlan (which scans
	// data+_version blobs), this scans arbitrary columns/expressions
	// and returns them by alias.
	AggregateQuery(ctx context.Context, sql string, args []interface{}, aliases []string) ([]map[string]interface{}, error)

	// IsAdaptedEntity reports whether the given entity uses an adapted
	// table (column-per-field storage). The OQL planner uses this to
	// decide whether aggregate push-down is possible.
	IsAdaptedEntity(entity string) bool

	// AdaptedColumnInfo returns the SQL column name for a JSON field in
	// an adapted entity. Returns ("", false) if the entity is not adapted
	// or the field is not a known column. For decimal columns, also returns
	// the scale so the caller can denormalise aggregated values.
	AdaptedColumnInfo(entity, jsonField string) (colName string, scale int, isDecimal bool, ok bool)

	// AdaptedTableName returns the SQL table name for an adapted entity.
	// Returns ("", false) if the entity is not adapted.
	AdaptedTableName(entity string) (string, bool)

	// StorageDialectFor returns the StorageDialect for the given entity,
	// or nil if the entity is not adapted. This allows SQL generators to
	// access dialect methods without importing backend-specific types.
	StorageDialectFor(entity string) StorageDialect
}

AggregateQueryable is an optional interface for storage backends that support native GROUP BY + aggregate push-down for adapted tables.

When an entity has an adapted table, aggregate queries can run entirely in SQL against native columns instead of fetching all rows into Go. The OQL executor checks for this interface via type assertion.

type Batcher 

type Batcher interface {
	BatchCreate(ctx context.Context, entity string, items []map[string]interface{}) ([]int, error)
	BatchDelete(ctx context.Context, entity string, ids []int) error
}

Batcher defines optional batch operation support

type ColumnChange 

type ColumnChange struct {
	Name       string
	OldSQLType string
	NewSQLType string
	OldType    string
	NewType    string
}

ColumnChange records an incompatible type change for a column.

type ColumnDef 

type ColumnDef struct {
	Name        string `json:"name"`          // Column name (e.g., "age", "REF_author_entity")
	JSONField   string `json:"json_field"`    // Original JSON field name (e.g., "age", "author")
	Type        string `json:"type"`          // JSON Schema type: string, integer, number, boolean, array, object
	Format      string `json:"format"`        // JSON Schema format: "", "decimal", "ref", "email", etc.
	SQLType     string `json:"sql_type"`      // Backend-specific SQL type (e.g., "TEXT", "INTEGER", "REAL")
	Required    bool   `json:"required"`      // Whether the field is in the schema's required array
	Precision   int    `json:"precision"`     // For decimal: total significant digits
	Scale       int    `json:"scale"`         // For decimal: digits after decimal point
	IsREF       bool   `json:"is_ref"`        // True if this column is part of a REF decomposition
	IsREFEntity bool   `json:"is_ref_entity"` // True for the _entity column of a REF pair
	IsREFID     bool   `json:"is_ref_id"`     // True for the _id column of a REF pair
}

ColumnDef describes a single column in an adapted table.

type CommitAppend 

type CommitAppend struct {
	Entity string                 `json:"entity"`
	ID     *int                   `json:"id,omitempty"`
	Data   map[string]interface{} `json:"data"`
}

CommitAppend describes one record to insert in a Commit operation. If ID is nil, the backend auto-generates an ID. If ID is non-nil and a record with that ID already exists in the entity type, ErrAlreadyExists is returned and the entire commit is rolled back.

type CommitAppendResult 

type CommitAppendResult struct {
	Entity string `json:"entity"`
	ID     int    `json:"id"`
}

CommitAppendResult describes one inserted record in a Commit response. ID is always set; for auto-generated IDs it contains the assigned value.

type CommitFsmWalk 

type CommitFsmWalk struct {
	Machine int                    `json:"machine"`
	Input   string                 `json:"input"`
	Payload map[string]interface{} `json:"payload,omitempty"`
}

CommitFsmWalk describes an FSM walk to execute atomically with the commit. This is a v2 type; it is defined here so that CommitRequest can carry it without a circular import. The walk is executed by the server's v2 FSM handler when API v2 is enabled; when v2 is disabled, the field is ignored and the request is rejected if no append or timeseries work is present.

type CommitFsmWalkResult 

type CommitFsmWalkResult struct {
	Machine    int                    `json:"machine"`
	Previous   string                 `json:"previous"`
	Current    string                 `json:"current"`
	Terminal   bool                   `json:"terminal"`
	Outputs    []string               `json:"outputs,omitempty"`
	Vars       map[string]interface{} `json:"vars,omitempty"`
	Definition interface{}            `json:"-"`
}

CommitFsmWalkResult describes the outcome of an FSM walk executed atomically within a commit. This is a v2 type.

type CommitRequest 

type CommitRequest struct {
	Update     CommitUpdate    `json:"update"`
	Append     []CommitAppend  `json:"append"`
	Timeseries []CommitTSEvent `json:"timeseries,omitempty"`
	// FsmWalk is set when a state machine walk must be atomic with the
	// entity write. Populated only when API v2 is enabled. Nil otherwise.
	FsmWalk *CommitFsmWalk `json:"fsm_walk,omitempty"`
}

CommitRequest is the payload for the atomic commit endpoint. It performs one conditional upsert (Update), zero or more unconditional entity inserts (Append), and zero or more timeseries events (Timeseries). At least one of Append, Timeseries, or FsmWalk must be non-empty.

When Timeseries is non-empty the server writes those events to the Pebble timeseries store BEFORE opening the SQLite transaction. If the Pebble write succeeds but the SQLite transaction subsequently fails, the server issues a synchronous DeleteKeys call to tombstone the written events before returning the error to the caller. See docs/COMMIT_ENDPOINT.md.

FsmWalk is a v2 field. When API v2 is disabled, it is accepted in the request body (JSON unmarshalling ignores unknown fields) but is never acted upon — the server treats a request with only FsmWalk set the same as a request with all fields empty, returning XOLU-CM003. This ensures v1-only deployments that accidentally receive a v2 request body fail cleanly rather than silently discarding the walk.

type CommitResult 

type CommitResult struct {
	Update     CommitUpdateResult   `json:"update"`
	Appended   []CommitAppendResult `json:"appended"`
	TSAccepted int                  `json:"ts_accepted,omitempty"`
	// FsmWalk is populated when the commit included an fsm_walk field and
	// API v2 is enabled. Nil otherwise.
	FsmWalk *CommitFsmWalkResult `json:"fsm_walk,omitempty"`
}

CommitResult is returned on a successful Commit.

type CommitTSEvent 

type CommitTSEvent struct {
	Timeline int64     `json:"timeline"`
	Dims     []uint64  `json:"dims"`
	Time     time.Time `json:"time"`
	Nums     []float64 `json:"nums,omitempty"`
	Payload  []byte    `json:"payload,omitempty"`
}

CommitTSEvent is one timeseries event carried inside a CommitRequest. It maps directly onto timeseries.Event; the Timeline must already be defined for the tenant via POST /ts/timelines before /commit is called.

type CommitUpdate 

type CommitUpdate struct {
	Entity  string                 `json:"entity"`
	ID      int                    `json:"id"`
	Version *int                   `json:"version,omitempty"`
	Data    map[string]interface{} `json:"data"`
}

CommitUpdate describes the entity to upsert in a Commit operation. If Version is non-nil, the write is conditional: it proceeds only if the stored _version equals *Version. A mismatch returns ErrConflict.

type CommitUpdateResult 

type CommitUpdateResult struct {
	Entity  string `json:"entity"`
	ID      int    `json:"id"`
	Created bool   `json:"created"`
	Version int    `json:"version"`
}

CommitUpdateResult describes the outcome of the upsert in a Commit. Created is true when a new record was inserted; false when an existing record was overwritten. Version is the _version value after the commit.

type EdgeFTSResult 

type EdgeFTSResult struct {
	Rel    string // relationship label
	EdgeID int    // surrogate edge ID (0 for topology-only edges)
}

EdgeFTSResult is one row returned by SearchEdges.

type EdgeFTSStore 

type EdgeFTSStore interface {
	// IndexEdgeContent extracts searchable text from props and writes a row
	// to t<X>_efts keyed by (rel, edgeID). Idempotent — updates on conflict.
	IndexEdgeContent(ctx context.Context, rel string, edgeID int, props map[string]interface{}) error

	// SearchEdges executes a full-text search against t<X>_efts and returns
	// matching (rel, edgeID) pairs in BM25 rank order. limit ≤ 0 returns all.
	SearchEdges(ctx context.Context, query string, limit int) ([]EdgeFTSResult, error)
}

EdgeFTSStore is an optional interface implemented by backends that support full-text search over edge property content via t<X>_efts.

type EdgeLister 

type EdgeLister interface {
	// ListEdges returns all property rows for rel from t<X>_edges (blob path).
	// Each row includes edge_id, rel, and all properties from the JSON blob.
	ListEdges(ctx context.Context, rel string) ([]map[string]interface{}, error)

	// IsEdgeLabel reports whether rel is a registered edge label (adapted or
	// blob) rather than a node entity type. Used by the OQL executor to route
	// FROM <label> queries correctly.
	IsEdgeLabel(ctx context.Context, rel string) (bool, error)

	// ResolveEdgeRelName returns the canonical (registry-cased) relationship
	// label for rel. OQL normalises entity names to lowercase; this method
	// restores the original casing so adapted.Get() and table queries work.
	// Returns rel unchanged when no canonical name is found.
	ResolveEdgeRelName(ctx context.Context, rel string) string
}

EdgeLister is an optional interface for storage backends that can list all edge property rows for a given relationship label.

SELECT * FROM KNOWS in OQL routes here when KNOWS is a blob edge label. For adapted edge labels (t<X>_edata_KNOWS), the existing adapted.Get() path in List() already handles it without this interface.

type EdgePropertyStore 

type EdgePropertyStore interface {
	// AddEdgeWithProps persists edge topology and writes a property blob if
	// props is non-nil and non-empty. Returns the assigned surrogate edge ID
	// (0 if no property row was written) and any error.
	AddEdgeWithProps(ctx context.Context, from, to, relationship string, props map[string]interface{}) (edgeID int, err error)

	// GetEdge retrieves the property blob for a single edge identified by its
	// surrogate edge ID. Returns ErrNotFound if the edge has no property row.
	GetEdge(ctx context.Context, edgeID int) (*EdgePropsResult, error)

	// GetManyEdges retrieves property blobs for multiple edges in a single
	// query. Edge IDs with no property row are absent from the result map;
	// they are not errors. The caller must not assume the result map contains
	// all requested IDs.
	GetManyEdges(ctx context.Context, edgeIDs []int) (map[int]*EdgePropsResult, error)
}

EdgePropertyStore is an optional interface implemented by backends that support edge property storage. When the store implements this interface, callers can write and retrieve property blobs associated with a specific edge (identified by the surrogate edge ID in t<X>_eseq).

Backends that do not implement this interface (e.g. test stubs) return ErrNotFound for all GetEdge calls; callers must handle this gracefully.

type EdgePropsResult 

type EdgePropsResult struct {
	EdgeID     int
	Rel        string
	Properties map[string]interface{}
}

EdgePropsResult is the result of a GetEdge call. Properties contains the JSON property map stored for the edge; it is nil when the edge has no property row (EdgeID == 0 in the topology table).

type EdgeSchemaStore 

type EdgeSchemaStore interface {
	// RegisterEdgeSchema persists a JSON Schema for the given relationship
	// label in t<X>_e_sch and suppresses the unregistered-label warning for
	// that label. Idempotent when called with the same schema hash.
	RegisterEdgeSchema(ctx context.Context, rel string, schema map[string]interface{}) error

	// SuppressEdgeSchemaWarning silences the unregistered-label warning for
	// rel without registering a schema. Useful when the caller intentionally
	// uses a schemaless label with properties and does not want log noise.
	// The suppression is in-memory only and resets on restart.
	SuppressEdgeSchemaWarning(rel string)

	// IsEdgeSchemaRegistered reports whether rel has a persisted schema entry
	// in t<X>_e_sch.
	IsEdgeSchemaRegistered(ctx context.Context, rel string) (bool, error)
}

EdgeSchemaStore is an optional interface implemented by backends that support the per-tenant edge schema registry (t<X>_e_sch).

Registering a schema for an edge label suppresses the warn-once log that fires when AddEdgeWithProps is called for an unregistered label, and is a prerequisite for Stage 7 (adapted edge tables). Labels used only for topology (plain AddEdge, no properties) never trigger the warning and do not need registration.

type EntityAdapter added in v0.26.0 

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

EntityAdapter is entity's dxp.Participant. One per SQLiteStore; safe for concurrent use.

func NewEntityAdapter added in v0.26.0 

func NewEntityAdapter(store *SQLiteStore, cache *dxp.MemCache) *EntityAdapter

NewEntityAdapter wires store into cache (SetDxpClaims — shared with fsm's adapter if both are registered against the same store) and returns an EntityAdapter ready to register with a dxp coordinator under the primitive key "entity".

func (*EntityAdapter) Execute added in v0.26.0 

func (a *EntityAdapter) Execute(ctx context.Context, store dxp.ParticipantStore, c dxp.Claim) (dxp.Result, error)

Execute applies the pending op via the store's existing saveInTx or createInTx — no new write path needed for either, unlike bal and cal, because both were already externally-transactable (built for the /commit path) before this session's dxp work. createInTx was checked directly, not assumed, to be genuinely tx-scoped throughout — every helper it calls (adaptedCreate, syncGraphEdges, indexForFTS) takes tx as a parameter and never falls back to a non-tx read, the same class of bug found and fixed in cal's own adapter (T-82). createInTx's returned allocated id is discarded here — Execute's current signature has no Result to carry it in (see dxp-coordinator-design.md §10); this is a concrete example of exactly what that would be for, once built.

func (*EntityAdapter) PostCommit added in v0.26.0 

func (a *EntityAdapter) PostCommit(ctx context.Context, c dxp.Claim) error

PostCommit is a no-op — entity has no derived/advisory plane a commit signal would need to update. Implemented so EntityAdapter satisfies dxp.Participant without a second interface-change ripple later.

func (*EntityAdapter) Release added in v0.26.0 

func (a *EntityAdapter) Release(ctx context.Context, c dxp.Claim) error

Release drops txn's stashed params, if any. Idempotent and unconditional, matching bal.Adapter.Release and FsmAdapter.Release.

func (*EntityAdapter) Reserve added in v0.26.0 

func (a *EntityAdapter) Reserve(ctx context.Context, tenant string, op dxp.OpParams,
	txn, participantID string, deadline int64, w dxp.Weight) (dxp.Claim, error)

Reserve admits an update or an append, dispatching on op's concrete type (both share the primitive key "entity", so the coordinator routes both through this same adapter — not two separate Participants). Read-only, per T-54's memory-only-reservation rule.

Admission rule, matching fsm.Adapter's: a live PESSIMISTIC claim on a resource refuses any new reservation of either weight (exclusive — the row is locked mid-update); a live OPTIMISTIC claim only refuses a new PESSIMISTIC one; OPTIMISTIC siblings coexist.

func (*EntityAdapter) Validate added in v0.26.0 

func (a *EntityAdapter) Validate(ctx context.Context, c dxp.Claim) error

Validate re-checks admission against current data, dispatching on the pending op's concrete type. A conflict here is what a competitor's commit looks like from this adapter's vantage; as with bal and fsm, classifying it as DXP007 (lost) versus DXP003 (drift) needs coordinator-level context this adapter doesn't have, so it returns ErrConflict / ErrAlreadyExists / a plain error and leaves that classification up the stack.

type EntityAppendParams added in v0.26.0 

type EntityAppendParams struct {
	Entity string                 `json:"entity"`
	ID     *int                   `json:"id,omitempty"`
	Data   map[string]interface{} `json:"data"`
}

EntityAppendParams is entity's dxp.OpParams for the CREATE path — entity's own vocabulary calls this "Append" (matching CommitAppend, the non-dxp /commit path's identical type), not "Create"; matched here rather than inventing a parallel term.

Two admission shapes, matching CommitAppend's own documented contract exactly:

  • ID != nil: a caller-chosen id. Reserve refuses if that id already exists (ErrAlreadyExists) — the same conflict an UPDATE on the same id would hit, deliberately sharing dxpEntityResource's resource-key namespace with EntityUpdateParams so the two correctly contend for the same row rather than silently coexisting.
  • ID == nil: server-allocated via the entity type's own sequence (nodeSeqTable, an atomic `next_id = next_id + 1 RETURNING`). There is no id to check for conflict before Execute actually allocates one — two concurrent auto-id creates get two different ids by the sequence's own construction, not by anything this adapter arranges. Reserve still Holds a claim, under a txn-scoped resource key that cannot collide with anything else, purely so the coordinator's own bookkeeping (attendance, Release) stays uniform across every participant regardless of whether that participant has a real resource to guard.

func (EntityAppendParams) Primitive added in v0.26.0 

func (EntityAppendParams) Primitive() string

Primitive satisfies dxp.OpParams.

type EntityLister 

type EntityLister interface {
	ListEntities(ctx context.Context) ([]string, error)
}

EntityLister defines optional entity type listing support

type EntityUpdateParams added in v0.26.0 

type EntityUpdateParams struct {
	Entity        string                 `json:"entity"`
	ID            int                    `json:"id"`
	Data          map[string]interface{} `json:"data"`
	ExpectVersion *int                   `json:"expect_version,omitempty"`
}

EntityUpdateParams is entity's dxp.OpParams (T-54's typed-per- primitive decision): a version-guarded update to one existing entity row.

Covers the UPDATE path — the entity named by (Entity, ID) must already exist. See EntityAppendParams, below, for CREATE (T-84, 2026-07-29) — a distinct admission shape (existence must be false, not true; auto-generated ids have no existence to check at all before Execute) handled by its own type, dispatched by this adapter's own Reserve/Validate/Execute via a type switch on dxp.OpParams's concrete type, not a separate Participant.

ExpectVersion mirrors CommitUpdate.Version: nil means unconditional (last-writer-wins on Execute, matching Save's own default), non-nil makes both Reserve's admission check and Execute's write CAS on the stored _version.

func (EntityUpdateParams) Primitive added in v0.26.0 

func (EntityUpdateParams) Primitive() string

Primitive satisfies dxp.OpParams.

type FieldIntrospector 

type FieldIntrospector interface {
	// JSONType returns the JSON Schema type string:
	// "string", "integer", "number", "boolean", "array", "object"
	JSONType() string

	// Format returns the declared format ("email", "decimal", "ref", "")
	Format() string

	// EnumValues returns the declared enum values, or nil if none.
	EnumValues() []string

	// Meta returns a metadata value by key (e.g., "decimalPrecision").
	// Returns (nil, false) if the key is not set.
	Meta(key string) (interface{}, bool)
}

FieldIntrospector provides read access to a single field's type and constraints. This maps to queryfy's per-type schema introspection (StringSchema.FormatType, NumberSchema.RangeConstraints, etc.).

type FieldQueryable 

type FieldQueryable interface {
	// ListWithFields returns all records for an entity type, but each
	// record contains only the specified fields plus _version. Fields
	// not present in a particular record's JSON are omitted from the
	// returned map (no null padding).
	//
	// For adapted entities this falls through to the regular List path
	// (adapted tables already select native columns efficiently).
	ListWithFields(ctx context.Context, entity string, fields []string) ([]map[string]interface{}, error)

	// QueryWithFields executes a pre-built SQL query (WHERE push-down)
	// and returns results with only the specified fields extracted from
	// the data blob. Like QueryWithPlan but avoids full deserialisation.
	QueryWithFields(ctx context.Context, sqlQuery string, args []interface{}, fields []string) ([]map[string]interface{}, error)
}

FieldQueryable is an optional interface for storage backends that support selective field extraction from blob entities.

Instead of deserialising every JSON blob into a full map, a FieldQueryable backend can extract only the requested fields during the scan loop. For blob entities this avoids allocating maps with dozens of unused keys.

The OQL executor checks for this interface via type assertion. If the store satisfies FieldQueryable and the query's SELECT list names specific fields (not SELECT *), the executor may call ListWithFields instead of List.

type FilterableStore 

type FilterableStore interface {
	FieldQueryable

	// ListWithFieldsAndFilter returns records for an entity type,
	// extracting only the specified fields and applying the predicate
	// set during tokenisation. Rows that fail the predicates are never
	// materialised as maps.
	//
	// The caller must ensure that fields includes all columns needed
	// for the SELECT list. Predicate fields may or may not overlap
	// with output fields.
	ListWithFieldsAndFilter(ctx context.Context, entity string, fields []string, preds *jsonic.PredicateSet) ([]map[string]interface{}, error)
}

FilterableStore is an optional extension of FieldQueryable that supports predicate evaluation during JSON tokenisation (B4 push-down).

Instead of extracting all rows and filtering in Go afterward, a FilterableStore evaluates simple predicates inline during the token walk, skipping map allocation for rows that don't match.

The OQL executor checks for this interface via type assertion. If the store satisfies FilterableStore and the WHERE clause can be expressed as a jsonic.PredicateSet, the executor passes predicates down.

type FsmAdapter added in v0.26.0 

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

FsmAdapter is fsm's dxp.Participant. One per SQLiteStore; safe for concurrent use.

func NewFsmAdapter added in v0.26.0 

func NewFsmAdapter(store *SQLiteStore, cache *dxp.MemCache) *FsmAdapter

NewFsmAdapter wires store into cache (SetDxpClaims) and returns an FsmAdapter ready to register with a dxp coordinator under the primitive key "fsm".

func (*FsmAdapter) Execute added in v0.26.0 

func (a *FsmAdapter) Execute(ctx context.Context, store dxp.ParticipantStore, c dxp.Claim) (dxp.Result, error)

Execute re-resolves tp's transition against tx (the coordinator's shared transaction, freshly — not the Reserve-time snapshot, so a change since Reserve is caught here rather than trusted from stale data) and applies it via fsmApplyTransitionInTx, whose CAS on the freshly-observed state is the mechanism that actually enforces "nothing moved this machine out from under us." Bypasses FsmWalkInTx's own claims gate deliberately — Execute IS the dxp-authorised path; gating it on its own claim would self-block.

func (*FsmAdapter) PostCommit added in v0.26.0 

func (a *FsmAdapter) PostCommit(ctx context.Context, c dxp.Claim) error

PostCommit is a no-op — fsm has no derived/advisory plane a commit signal would need to update. Implemented so FsmAdapter satisfies dxp.Participant without a second interface-change ripple later.

func (*FsmAdapter) Release added in v0.26.0 

func (a *FsmAdapter) Release(ctx context.Context, c dxp.Claim) error

Release drops txn's stashed params, if any. Idempotent and unconditional, matching bal.Adapter.Release.

func (*FsmAdapter) Reserve added in v0.26.0 

func (a *FsmAdapter) Reserve(ctx context.Context, tenant string, op dxp.OpParams,
	txn, participantID string, deadline int64, w dxp.Weight) (dxp.Claim, error)

Reserve resolves tp's transition read-only (fsmResolveInTx — no write, per T-54's memory-only-reservation rule) and, on a legal transition, Holds a claim.

Admission rule (proposal §5c's fsm resolution, made precise for mixed weights on one machine, which the design left unstated): a live PESSIMISTIC claim always refuses a new reservation, of either weight — pessimistic means exclusive, "the machine is locked mid-step". A live OPTIMISTIC claim only refuses a new PESSIMISTIC reservation (pessimistic wants exclusivity even against optimistic siblings); optimistic reservations coexist with each other freely, per "competing reserved transitions from one state coexist."

func (*FsmAdapter) Validate added in v0.26.0 

func (a *FsmAdapter) Validate(ctx context.Context, c dxp.Claim) error

Validate re-resolves the same transition read-only. Any change since Reserve — the machine moved, the guard now fails, the machine went terminal — surfaces as fsmResolveInTx's own error; Validate does not itself classify DXP007 (lost to a competitor) versus DXP003 (drift), for the same reason bal's Validate doesn't: that needs coordinator- level context this adapter doesn't have. See bal.Adapter.Validate.

type FsmTransitionParams added in v0.26.0 

type FsmTransitionParams struct {
	TenantID      tenant.TenantID        `json:"-"` // never trusted from participant params — the coordinator always sets this from the instance's own actual tenant
	MachineID     int64                  `json:"machine_id"`
	Input         string                 `json:"input"`
	Payload       map[string]interface{} `json:"payload,omitempty"`
	QueryBindings map[string]interface{} `json:"query_bindings,omitempty"`
}

FsmTransitionParams is fsm's dxp.OpParams (T-54's typed-per-primitive decision): everything FsmWalkInTx needs to resolve and apply one transition, minus the tentative-write step itself.

TenantID is carried explicitly (fsm's storage layer is keyed by tenant.TenantID, not a string prefix like bal's) rather than derived from the tenant string Reserve/Execute receive — Reserve cross-checks the two agree (TenantID.String() == tenant) so a caller passing a mismatched pair fails loudly instead of silently splitting the same logical tenant across two cache shards.

func (FsmTransitionParams) Primitive added in v0.26.0 

func (FsmTransitionParams) Primitive() string

Primitive satisfies dxp.OpParams.

type FsmWalkError 

type FsmWalkError struct {
	Code    string
	Message string
}

FsmWalkError is a typed error carrying an XOLU-FSM code so the server layer can map it to the correct HTTP status. Walk failures inside a commit are surfaced by the server as XOLU-FSM008.

func (*FsmWalkError) Error 

func (e *FsmWalkError) Error() string

type FsmWalkResult 

type FsmWalkResult struct {
	Previous  string
	Current   string
	Terminal  bool
	Outputs   []string
	Vars      map[string]interface{}
	HistoryID int64

	// Definition is the machine's definition spec as a map[string]interface{},
	// exposed read-only to event consumers (the "definition" namespace in FSM
	// event data) so jsonplates can reference definition facts.
	//
	// It is a map (not the fsmDefinitionSpec struct) because queryfy's path
	// engine traverses maps, not structs. It is decoded fresh from the snapshot
	// JSON, so it is an independent copy with no aliasing back to the running
	// machine's spec — no separate deep copy is needed for safety, and the
	// running machine cannot be mutated through it. See docs/EVENT_PENDING.md §6b.
	Definition interface{}
}

FsmWalkResult is the outcome of a successful walk.

type FsmWalker 

type FsmWalker interface {
	FsmWalkInTx(ctx context.Context, tx *sql.Tx, tenantID tenant.TenantID,
		machineID int64, input string, payload map[string]interface{},
		queryBindings map[string]interface{}) (*FsmWalkResult, error)
}

FsmWalker is implemented by stores that can execute an FSM walk on a caller-supplied transaction. It lets the server run the standalone POST /fsm/machine/{id}/walk through the same code path as the commit-embedded walk, without depending on the concrete store type.

type GraphEdge 

type GraphEdge struct {
	SourceEntity string
	SourceID     int
	TargetEntity string
	TargetID     int
	Relationship string
	EdgeID       int
}

GraphEdge holds the columns of one row from the graph edges table. EdgeID is 0 when the edge has no property row.

type GraphEdgeScanner 

type GraphEdgeScanner interface {
	ScanGraphEdges(ctx context.Context, tenantID tenant.TenantID, fn func(GraphEdge) error) error
}

GraphEdgeScanner is an optional interface for storage backends that can stream graph edges directly from their edge table without deserialising full entity JSON. Implementing this interface enables O(edges) startup graph hydration instead of O(entities × JSON size).

ScanGraphEdges calls fn once per edge row. Iteration stops on the first non-nil error returned by fn. A nil error from ScanGraphEdges means all rows were scanned (or fn stopped iteration early with a sentinel — callers must define their own sentinel if needed).

tenantID scopes the scan to a specific tenant's edge table. Pass 0 for the default (tenant-0) table. Future SQL backends may extend this to scan all tenant tables in a single call; the current SQLite implementation scans one tenant at a time, matching the existing startup scope.

type GraphIntegrity 

type GraphIntegrity interface {
	VerifyGraphIntegrity(ctx context.Context) error
	RebuildGraph(ctx context.Context) error
}

GraphIntegrity defines optional graph integrity checking

type GraphNeighbors 

type GraphNeighbors interface {
}

GraphNeighbors defines optional graph neighbor queries

type IDGenerator 

type IDGenerator interface {
	NextID(ctx context.Context, entity string) (int, error)
}

IDGenerator defines interface for ID generation strategies

type IndexDef 

type IndexDef struct {
	Name    string   `json:"name"`    // Index name
	Columns []string `json:"columns"` // Column names
	Unique  bool     `json:"unique"`  // Whether the index is unique
}

IndexDef describes an index on an adapted table.

type InfoProvider 

type InfoProvider interface {
	Info() StoreInfo
}

InfoProvider allows stores to provide metadata about their capabilities

type MetaSubject added in v0.16.13 

type MetaSubject struct {
	Kind string // entity name, or dotted namespaced kind
	Key  string // canonical key text (validated per kind)
}

MetaSubject is a validated subject address.

func EntitySubject added in v0.16.13 

func EntitySubject(entity string, id int) MetaSubject

EntitySubject builds the subject for an entity row — the cascade-delete path's constructor.

func ParseMetaSubject added in v0.16.13 

func ParseMetaSubject(kind, key string, entityNameOK func(string) error) (MetaSubject, error)

ParseMetaSubject validates a (kind, key) pair from the API boundary and returns the canonical subject. entityNameOK validates undotted kinds as entity names (the server passes its validateEntityName; storage-level callers may pass nil to accept any well-formed entity name shape).

func (MetaSubject) String added in v0.16.13 

func (s MetaSubject) String() string

String renders the canonical "kind/key" form used in errors and logs.

type Migrator 

type Migrator interface {
	Migrate(ctx context.Context) error
	Version(ctx context.Context) (int, error)
}

Migrator defines optional schema migration support Useful for database backends

type PagedLister 

type PagedLister interface {
	// ListPaged returns a single page of entities, plus the total count.
	// limit and offset are applied at the storage layer (SQL LIMIT/OFFSET).
	ListPaged(ctx context.Context, entity string, limit, offset int) (*PagedResult, error)
}

PagedLister is an optional interface for storage backends that support server-side pagination. Backends that implement this avoid loading every record into memory for paginated list requests.

type PagedResult 

type PagedResult struct {
	Data       []map[string]interface{}
	TotalItems int
}

PagedResult holds a page of results plus the total count.

type QueryCapabilities 

type QueryCapabilities struct {
	Where   bool // Can filter with json_extract predicates
	OrderBy bool // Can sort by json_extract fields
	Limit   bool // Can apply TOP/LIMIT
	Count   bool // Can return entity count without full scan
}

QueryCapabilities reports what a storage backend can handle natively via predicate push-down. Used by the OQL planner to decide which operations to delegate to the storage engine versus executing in Go.

type Queryable 

type Queryable interface {
	// Capabilities reports which query operations this backend handles
	// natively. The planner will not attempt to push down an operation
	// unless the corresponding capability is true.
	Capabilities() QueryCapabilities

	// CountEntities returns the number of records for an entity type
	// without fetching the records themselves. Used by the planner to
	// determine whether push-down is worthwhile (the fixed overhead of
	// a push-down query exceeds Go-side cost for small datasets).
	CountEntities(ctx context.Context, entity string) (int, error)

	// QueryWithPlan executes a pre-built SQL query with parameterised
	// arguments and returns the results as maps, in the same format as
	// List(). The SQL is generated by the OQL planner's SQL generator
	// and always selects from the data column with json_extract().
	QueryWithPlan(ctx context.Context, sql string, args []interface{}) ([]map[string]interface{}, error)
}

Queryable is an optional interface for storage backends that support predicate push-down. Backends that do not implement it receive the full Go-side execution path for all operations.

The OQL planner checks for this interface via type assertion. If the store satisfies Queryable and the entity cardinality exceeds the push-down threshold, the planner may generate SQL to delegate WHERE, ORDER BY, and LIMIT operations to the storage engine.

type RefTargetMissingError added in v0.16.13 

type RefTargetMissingError struct {
	SourceEntity string
	SourceID     int
	TargetEntity string
	TargetID     int
}

RefTargetMissingError reports that a write was refused because a REF field names a target that does not exist at commit time — the create-side sibling of RestrictViolationError, checked inside the write's own transaction (G-12 create-side closure; @R02.3 shipped early). Maps to XOLU-RI003 / HTTP 400 at the handler.

func (*RefTargetMissingError) Error added in v0.16.13 

func (e *RefTargetMissingError) Error() string

type RestrictViolationError added in v0.16.3 

type RestrictViolationError struct {
	Referrers []string
}

RestrictViolationError reports that a delete was refused because live referrers with a restrict on_delete policy exist. Referrers holds up to a bounded number of "entity:id" keys for the error message.

func (*RestrictViolationError) Error added in v0.16.3 

func (e *RestrictViolationError) Error() string

type SQLiteConfig 

type SQLiteConfig struct {
	DBPath            string
	EnableWAL         bool // Write-Ahead Logging for better concurrency
	EnableForeignKeys bool
	CacheSize         int             // Page cache size in KB
	BusyTimeout       int             // Milliseconds to wait on locked database
	FullTextEnabled   bool            // Enable FTS5 full-text search indexing
	GraphEnabled      bool            // Enable graph edge table maintenance
	TenantID          tenant.TenantID // 0 = no tenant scoping

	// Performance tuning (zero = use backend defaults)
	//   SQLite defaults: MaxOpenConns=1 (WAL single-writer),
	//   MaxIdleConns=1, ReadPoolSize=NumCPU.
	MaxOpenConns        int // Max open write connections (0 = backend default)
	MaxIdleConns        int // Max idle write connections (0 = backend default)
	ReadPoolSize        int // Max open read connections (0 = backend default)
	ContentionThreshold int // Adaptive lock threshold 0-100 (default 95)

	// PerFileTenants mirrors StoreConfig.SQLitePerFileTenants.
	// When true, tenant isolation is provided by separate database files
	// rather than a tenant_id column; schema DDL and all query methods
	// omit tenant_id accordingly.
	PerFileTenants bool
}

SQLiteConfig holds SQLite-specific configuration

type SQLiteStorageDialect 

type SQLiteStorageDialect struct{}

SQLiteStorageDialect implements StorageDialect for SQLite. Table names encode the tenant (t<XXXX>_*) so no tenant_id column is needed in any data or schema table.

func (*SQLiteStorageDialect) ColumnType 

func (d *SQLiteStorageDialect) ColumnType(jsonType, format string, precision, scale int) string

func (*SQLiteStorageDialect) CreateIndexSQL 

func (d *SQLiteStorageDialect) CreateIndexSQL(spec *AdaptedTableSpec) []string

func (*SQLiteStorageDialect) CreateTableSQL 

func (d *SQLiteStorageDialect) CreateTableSQL(spec *AdaptedTableSpec) string

func (*SQLiteStorageDialect) DeleteSQL 

func (d *SQLiteStorageDialect) DeleteSQL(spec *AdaptedTableSpec) string

func (*SQLiteStorageDialect) DenormaliseDecimal 

func (d *SQLiteStorageDialect) DenormaliseDecimal(value string, precision, scale int) string

DenormaliseDecimal converts a scaled int64 string back to a client-facing decimal string by dividing by 10^scale. 

"1990"  with scale=2 → "19.90"
"-1990" with scale=2 → "-19.90"
"0"     with scale=2 → "0.00"

func (*SQLiteStorageDialect) ExistsSQL 

func (d *SQLiteStorageDialect) ExistsSQL(spec *AdaptedTableSpec) string

func (*SQLiteStorageDialect) InsertSQL 

func (d *SQLiteStorageDialect) InsertSQL(spec *AdaptedTableSpec, hasExtra bool) (string, []string)

func (*SQLiteStorageDialect) Name 

func (d *SQLiteStorageDialect) Name() string

func (*SQLiteStorageDialect) NodeSchemaTableSQL 

func (d *SQLiteStorageDialect) NodeSchemaTableSQL(tenantID tenant.TenantID) string

func (*SQLiteStorageDialect) NormaliseDecimal 

func (d *SQLiteStorageDialect) NormaliseDecimal(value string, precision, scale int) (string, error)

NormaliseDecimal converts a decimal string to a scaled int64 string for SQLite INTEGER storage. The value is multiplied by 10^scale.

Examples for precision=6, scale=2: 

"19.90"    → "1990"
"-19.90"   → "-1990"
"0"        → "0"
"9999.99"  → "999999"
"-0.01"    → "-1"

The scaled integer fits in int64 for precision up to 18.

func (*SQLiteStorageDialect) Placeholder 

func (d *SQLiteStorageDialect) Placeholder(_ int) string

func (*SQLiteStorageDialect) SelectAllSQL 

func (d *SQLiteStorageDialect) SelectAllSQL(spec *AdaptedTableSpec) string

func (*SQLiteStorageDialect) SelectSQL 

func (d *SQLiteStorageDialect) SelectSQL(spec *AdaptedTableSpec) string

func (*SQLiteStorageDialect) SupportsNativeDecimalAggregation 

func (d *SQLiteStorageDialect) SupportsNativeDecimalAggregation() bool

SupportsNativeDecimalAggregation returns false for SQLite. Although SQLite can SUM integers correctly, the scaled representation requires division by the scale factor to produce correct decimal results. Go-side aggregation with shopspring/decimal avoids this complexity and handles AVG correctly.

func (*SQLiteStorageDialect) UpdateSQL 

func (d *SQLiteStorageDialect) UpdateSQL(spec *AdaptedTableSpec, versionCheck bool) string

type SQLiteStore 

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

SQLiteStore implements Store interface using SQLite database.

It maintains two connection pools against the same WAL-mode database:

  • db (writer): MaxOpenConns=1, serialises all writes.
  • readDB (reader): MaxOpenConns=NumCPU, query_only=ON, parallel reads.

Under WAL mode, readers never block the writer and vice-versa.

func NewSQLiteStore 

func NewSQLiteStore(dbPath string, config SQLiteConfig) (*SQLiteStore, error)

NewSQLiteStore creates a new SQLite-based storage with separate reader and writer connection pools. Under WAL mode the writer never blocks readers and vice-versa, so splitting pools maximises concurrency.

func (*SQLiteStore) AdaptedColumnInfo 

func (s *SQLiteStore) AdaptedColumnInfo(entity, jsonField string) (colName string, scale int, isDecimal bool, ok bool)

AdaptedColumnInfo returns column metadata for a JSON field in an adapted entity.

func (*SQLiteStore) AdaptedRegistry 

func (s *SQLiteStore) AdaptedRegistry() *AdaptedRegistry

AdaptedRegistry returns the store's adapted table registry. Returns nil only if the store was not properly initialized.

func (*SQLiteStore) AdaptedTableName 

func (s *SQLiteStore) AdaptedTableName(entity string) (string, bool)

AdaptedTableName returns the SQL table name for an adapted entity.

func (*SQLiteStore) AddEdgeWithProps 

func (s *SQLiteStore) AddEdgeWithProps(ctx context.Context, from, to, relationship string, props map[string]interface{}) (int, error)

AddEdgeWithProps persists edge topology and, if props is non-nil and non-empty, writes a property blob to t<X>_edges and sets edge_id in t<X>_graph. Returns the assigned surrogate edge ID (0 = no props stored).

func (*SQLiteStore) AggregateQuery 

func (s *SQLiteStore) AggregateQuery(ctx context.Context, sql string, args []interface{}, aliases []string) ([]map[string]interface{}, error)

AggregateQuery executes an aggregate SQL query against native columns and returns results keyed by alias names.

func (*SQLiteStore) AllocateNodeID added in v0.26.0 

func (s *SQLiteStore) AllocateNodeID(ctx context.Context, entity string) (int, error)

AllocateNodeID reserves the next id from entity's own per-tenant sequence (the identical atomic upsert createInner's own id allocation uses) WITHOUT creating a row — T-121 (wave 10)'s own need: obj's promote composes bal-decrement + entity-create + obj-attach as one dxp transaction, but dxp's own coordinator has no mechanism for one leg's execution result (a just-created entity's auto-allocated id) to feed another leg's OpParams — every leg's params must be fully known when the transaction is *defined*, before any leg runs (confirmed directly: EntityAdapter.Execute's own create path discards createInTx's returned id entirely). Promote's own handler calls this once, outside the dxp transaction, then supplies the reserved id explicitly to both entity's own EntityAppendParams (the caller-chosen-id path, not auto-allocated) and obj's own attach params — no wire-contract change, the caller never sees or supplies an id themselves.

A transaction that later aborts (bal insufficient funds, obj capacity/cycle refusal) burns this id — the identical, universally- accepted gap-on-abort behaviour every auto-increment sequence has (a rolled-back INSERT ... RETURNING elsewhere in this codebase already produces the same characteristic); not a new risk this introduces.

func (*SQLiteStore) Capabilities 

func (s *SQLiteStore) Capabilities() QueryCapabilities

Capabilities reports that the SQLite backend can handle WHERE, ORDER BY, LIMIT, and COUNT natively via json_extract() push-down.

func (*SQLiteStore) Close 

func (s *SQLiteStore) Close() error

Close closes the database connection

func (*SQLiteStore) Commit 

func (s *SQLiteStore) Commit(ctx context.Context, req CommitRequest) (CommitResult, error)

Commit performs an atomic upsert + one or more inserts in a single SQLite transaction. The upsert supports optional CAS via Update.Version. All operations share one BEGIN/COMMIT boundary; any failure rolls back the entire set.

func (*SQLiteStore) Config 

func (s *SQLiteStore) Config() StoreConfig

Config returns the store's StoreConfig.

func (*SQLiteStore) ContentionLock 

func (s *SQLiteStore) ContentionLock() *AdaptiveLock

ContentiontLock returns the store's adaptive lock, allowing runtime configuration of the contention threshold via SetThreshold().

func (*SQLiteStore) CountEntities 

func (s *SQLiteStore) CountEntities(ctx context.Context, entity string) (int, error)

CountEntities returns the number of records for an entity type without fetching the data. This is a single indexed COUNT(*) — typically <10µs.

func (*SQLiteStore) Create 

func (s *SQLiteStore) Create(ctx context.Context, entity string, data map[string]interface{}) (int, error)

Create inserts a new entity with auto-generated ID

func (*SQLiteStore) DB 

func (s *SQLiteStore) DB() *sql.DB

DB returns the underlying *sql.DB for advanced operations such as batch seeding or direct SQL execution. Use with care — callers must respect the store's locking and schema conventions.

func (*SQLiteStore) Delete 

func (s *SQLiteStore) Delete(ctx context.Context, entity string, id int) error

Delete removes an entity

func (*SQLiteStore) DeleteWithRestrict added in v0.16.3 

func (s *SQLiteStore) DeleteWithRestrict(ctx context.Context, entity string, id int, restrictedBy []string) error

DeleteWithRestrict deletes entity:id, refusing with *RestrictViolationError if any entity named in restrictedBy still references the target. The referrer check runs INSIDE the delete's own transaction (@C04a: a guard must live where its transaction lives), which closes the check-then-act window the handler-level in-memory pre-check cannot (G-12): a concurrent referrer create either commits its edge row before our read (we see it and refuse) or serialises after our commit.

Referrer discovery is two-pronged, matching where REF-derived state authoritatively lives per storage class:

  • blob entities: the SQL edge table (synced transactionally by syncGraphEdges on every write);
  • adapted entities: their own REF_{field}_entity/_id columns, probed spec-driven via the IsREFEntity/IsREFID flags (adapted writes do not populate the edge table).

restrictedBy is the set of referring entity names that carry a restrict policy toward this target (the caller derives it from the schema x-ref registry). Empty/nil means no restrict policies exist and this behaves exactly like Delete.

func (*SQLiteStore) Exists 

func (s *SQLiteStore) Exists(ctx context.Context, entity string, id int) bool

Exists checks if an entity exists

func (*SQLiteStore) FsmWalkInTx 

func (s *SQLiteStore) FsmWalkInTx(ctx context.Context, tx *sql.Tx, tenantID tenant.TenantID,
	machineID int64, input string, payload map[string]interface{}, queryBindings map[string]interface{}) (*FsmWalkResult, error)

FsmWalkInTx executes one transition for the machine on the given tx. It performs: snapshot load, terminal check (XOLU-FSM005), transition lookup (XOLU-FSM003), guard evaluation (XOLU-FSM004), state advance, set-clause evaluation including NEXT VALUE FOR on this tx (XOLU-FSM011), and history append. It does not commit — the caller owns the transaction.

When a dxp cache is wired (SetDxpClaims), a live PESSIMISTIC dxp claim on this machine refuses the walk before it resolves anything (XOLU-FSM004: "machine is locked mid-step" — proposal §5c's fsm resolution) — the substrate-wide rule that every write path, not only the coordinator, must see dxp's holds. Optimistic claims are invisible here by design (§7); a plain walk may race one, and its own dxp Validate/Execute discovers the drift via the CAS in fsmApplyTransitionInTx.

func (*SQLiteStore) FullTextSearch 

func (s *SQLiteStore) FullTextSearch(ctx context.Context, query string, entity string) ([]map[string]interface{}, error)

FullTextSearch performs a full-text search across " + s.nodesTable() + "

func (*SQLiteStore) Get 

func (s *SQLiteStore) Get(ctx context.Context, entity string, id int) (map[string]interface{}, error)

Get retrieves an entity by ID

func (*SQLiteStore) GetEdge 

func (s *SQLiteStore) GetEdge(ctx context.Context, edgeID int) (*EdgePropsResult, error)

GetEdge retrieves property data for a single edge by surrogate ID. Routes to the adapted table (t<X>_edata_<label>) when the relationship label has an adapted spec; otherwise reads from the blob t<X>_edges table.

func (*SQLiteStore) GetMany 

func (s *SQLiteStore) GetMany(ctx context.Context, entity string, ids []int) (map[int]map[string]interface{}, error)

GetMany fetches multiple " + s.nodesTable() + " of the same type in a single query. Returns a map[id]data for every id found; missing ids are absent.

func (*SQLiteStore) GetManyEdges 

func (s *SQLiteStore) GetManyEdges(ctx context.Context, edgeIDs []int) (map[int]*EdgePropsResult, error)

GetManyEdges retrieves property data for multiple edge IDs in one pass. Dispatches each ID to adapted or blob path based on the relationship label. Edge IDs with no property row are absent from the result map.

func (*SQLiteStore) GraphEdgesOracle added in v0.16.13 

func (s *SQLiteStore) GraphEdgesOracle() chronicle.RebuildOracle

GraphEdgesOracle is the first real rebuild oracle (@C §4 extraction #3 consumer; operations roadmap item 5): the edge table is DERIVED state — every row is implied by a REF field in a blob entity's document, written by syncGraphEdges inside the entity's own transaction. Derive re-extracts the implied edges from the authoritative documents; Current reads the live table. Divergence means the derived plane has drifted from the record — the class of fault `iolu db check` exists to catch.

Boundary, stated deliberately: this oracle covers BLOB entities only. Adapted entities never populate the edge table — their REFs live decomposed in REF_{field}_entity/_id columns (the G-12 investigation's finding), and that plane's invariant is pinned by the REF compose/decompose conformance tests, not by this oracle.

func (*SQLiteStore) GraphTenantIDs 

func (s *SQLiteStore) GraphTenantIDs(ctx context.Context) ([]tenant.TenantID, error)

GraphTenantIDs implements TenantIDLister. It returns all tenant IDs for which a graph_tXXXX edge table should be hydrated at startup. Tenant 0 is always included first (it is implicit and never appears in the tenants registry table). Registered non-zero tenants follow in ascending order.

func (*SQLiteStore) IndexEdgeContent 

func (s *SQLiteStore) IndexEdgeContent(ctx context.Context, rel string, edgeID int, props map[string]interface{}) error

Uses DELETE + INSERT to simulate an UPSERT since FTS5 virtual tables do not support ON CONFLICT clauses.

func (*SQLiteStore) Info 

func (s *SQLiteStore) Info() StoreInfo

Info returns store information

func (*SQLiteStore) InitV2Schema 

func (s *SQLiteStore) InitV2Schema(ctx context.Context) error

InitV2Schema implements storage.V2SchemaInitialiser.

func (*SQLiteStore) IsAdaptedEntity 

func (s *SQLiteStore) IsAdaptedEntity(entity string) bool

IsAdaptedEntity reports whether the entity uses an adapted table.

func (*SQLiteStore) IsEdgeAdapted 

func (s *SQLiteStore) IsEdgeAdapted(rel string) bool

IsEdgeAdapted reports whether rel has an adapted (ElementEdge) spec in the in-memory registry.

func (*SQLiteStore) IsEdgeLabel 

func (s *SQLiteStore) IsEdgeLabel(ctx context.Context, rel string) (bool, error)

IsEdgeLabel reports whether rel is a registered edge label (either adapted with a t<X>_edata_<label> table, or blob-only registered in t<X>_e_sch). The lookup is case-insensitive so that OQL's entity-name normalisation (which lowercases FROM <Label> to "label") matches labels registered with any casing (e.g. "KNOWS", "Knows", "knows" all resolve correctly).

func (*SQLiteStore) IsEdgeSchemaRegistered 

func (s *SQLiteStore) IsEdgeSchemaRegistered(ctx context.Context, rel string) (bool, error)

IsEdgeSchemaRegistered reports whether rel has a row in t<X>_e_sch.

func (*SQLiteStore) IsPerFileTenant 

func (s *SQLiteStore) IsPerFileTenant() bool

IsPerFileTenant reports whether this store operates in per-file tenant mode. Implements TenantModeProvider. When true, each tenant has its own database file and the tenant_id column is absent from the schema.

func (*SQLiteStore) List 

func (s *SQLiteStore) List(ctx context.Context, entity string) ([]map[string]interface{}, error)

List returns all `+s.nodesTable()+` of a given type

func (*SQLiteStore) ListEdges 

func (s *SQLiteStore) ListEdges(ctx context.Context, rel string) ([]map[string]interface{}, error)

ListEdges returns all property rows for rel from the blob t<X>_edges table. Uses case-insensitive matching on rel so OQL-normalised names resolve correctly.

func (*SQLiteStore) ListEntities 

func (s *SQLiteStore) ListEntities(ctx context.Context) ([]string, error)

ListEntities returns all distinct entity types in the database. It unions blob entities (from the entities table) with adapted " + s.nodesTable() + " (from the in-memory registry), so that adapted entity types are visible to the OQL validator even when they have no blob rows.

func (*SQLiteStore) ListPaged 

func (s *SQLiteStore) ListPaged(ctx context.Context, entity string, limit, offset int) (*PagedResult, error)

ListPaged returns a single page of entities plus total count, using SQL LIMIT/OFFSET so only the requested page is deserialised.

func (*SQLiteStore) ListWithFields 

func (s *SQLiteStore) ListWithFields(ctx context.Context, entity string, fields []string) ([]map[string]interface{}, error)

ListWithFields returns all records for an entity, extracting only the named fields from each JSON blob. For adapted `+s.nodesTable()+` this falls through to the regular List path (native columns are already efficient).

func (*SQLiteStore) ListWithFieldsAndFilter 

func (s *SQLiteStore) ListWithFieldsAndFilter(ctx context.Context, entity string, fields []string, preds *jsonic.PredicateSet) ([]map[string]interface{}, error)

ListWithFieldsAndFilter returns records for an entity, extracting only the named fields and evaluating predicates inline during tokenisation. Rows that fail the predicates are never materialised as maps. For adapted entities this falls through to the regular path.

func (*SQLiteStore) MigrateBlobEntitiesToAdapted added in v0.26.0 

func (s *SQLiteStore) MigrateBlobEntitiesToAdapted(ctx context.Context, entityType string) (migrated int, err error)

MigrateBlobEntitiesToAdapted moves every existing row of entityType from blob storage (the generic nodes table) into its own adapted table, preserving each row's own ID exactly -- for T-151's strict promotion mode, called only after every row has already been validated against the new schema by the caller (this method itself does not validate; it assumes that already happened and the caller is committed to migrating).

Must be called after RegisterAdaptedEntity has already registered entityType's adapted table -- returns an error immediately if it hasn't (adapted.Get returns nil).

Runs as a single transaction: every row migrates, or none do. Graph edges need no attention here -- a migrated row's REF fields already created their graph edges when the row was originally written (any blob-storage create runs the same graph sync adapted creates do), and migration changes neither the row's own ID nor its content, so those edges remain correct without re-syncing.

Not chunked/batched -- for a very large entity population this is one large transaction, a known, deliberate scope limit for the first version of this feature rather than an oversight; batching with progress tracking is real additional work, not a small follow-up.

func (*SQLiteStore) NodesTable 

func (s *SQLiteStore) NodesTable() string

NodesTable returns the per-tenant blob node store table name (t<XXXX>_nodes). Implements storage.TableNamer; used by the OQL SQL generator to build correct push-down queries without hardcoding the table name.

func (*SQLiteStore) Patch 

func (s *SQLiteStore) Patch(ctx context.Context, entity string, id int, updates map[string]interface{}) error

Patch partially updates an entity

func (*SQLiteStore) PatchValidated 

func (s *SQLiteStore) PatchValidated(ctx context.Context, entity string, id int, updates map[string]interface{}, validate func(merged map[string]interface{}) error) error

PatchValidated applies a partial update inside a transaction and runs the validator against the merged data before committing.

func (*SQLiteStore) Ping 

func (s *SQLiteStore) Ping(ctx context.Context) error

Ping verifies that the database connection is alive.

func (*SQLiteStore) QueryWithFields 

func (s *SQLiteStore) QueryWithFields(ctx context.Context, sqlQuery string, args []interface{}, fields []string) ([]map[string]interface{}, error)

QueryWithFields executes a push-down SQL query and extracts only the named fields from each result's JSON blob.

func (*SQLiteStore) QueryWithPlan 

func (s *SQLiteStore) QueryWithPlan(ctx context.Context, sqlQuery string, args []interface{}) ([]map[string]interface{}, error)

QueryWithPlan executes a pre-built SQL query (generated by the OQL planner) and returns the results as maps, in the same format as List().

func (*SQLiteStore) ReaderDB 

func (s *SQLiteStore) ReaderDB() *sql.DB

ReaderDB returns the underlying reader connection pool. Used by the tenant persister for read-only queries.

func (*SQLiteStore) RebuildGraph 

func (s *SQLiteStore) RebuildGraph(ctx context.Context) error

RebuildGraph rebuilds the tenant edge table from stored entity JSON.

Correctness: uses models.ExtractEntityEdges for REF extraction so that @REFS ([]interface{} of REF maps) and TSREF exclusion are handled identically to the live syncGraphEdges path.

Performance: one PrepareContext call outside the row loop; edges are accumulated and flushed in batches of rebuildBatchSize rather than one ExecContext per edge.

func (*SQLiteStore) RegisterAdaptedEdge 

func (s *SQLiteStore) RegisterAdaptedEdge(ctx context.Context, rel string, schema map[string]interface{}) error

RegisterAdaptedEdge derives a native-column table for rel, creates it, persists the column spec in t<X>_e_sch, and populates the in-memory adapted registry with an ElementEdge spec.

func (*SQLiteStore) RegisterAdaptedEntity 

func (s *SQLiteStore) RegisterAdaptedEntity(ctx context.Context, entity string, schema map[string]interface{}) error

RegisterAdaptedEntity derives an adapted table for the given entity type from its JSON Schema and creates the table if it doesn't exist. This is called by the server layer when a schema is loaded or registered.

func (*SQLiteStore) RegisterEdgeSchema 

func (s *SQLiteStore) RegisterEdgeSchema(ctx context.Context, rel string, schema map[string]interface{}) error

RegisterEdgeSchema persists the JSON Schema for rel in t<X>_e_sch and suppresses the unregistered-label warning for that label. Idempotent when called with the same schema hash; updates when the hash changes.

func (*SQLiteStore) ResolveEdgeRelName 

func (s *SQLiteStore) ResolveEdgeRelName(ctx context.Context, rel string) string

ResolveEdgeRelName returns the canonical (registry-cased) relationship label. OQL normalises entity names to lowercase; this restores the original casing so adapted.Get() and edge table queries work correctly.

func (*SQLiteStore) Save 

func (s *SQLiteStore) Save(ctx context.Context, entity string, id int, data map[string]interface{}) (bool, error)

func (*SQLiteStore) ScanGraphEdges 

func (s *SQLiteStore) ScanGraphEdges(ctx context.Context, tenantID tenant.TenantID, fn func(GraphEdge) error) error

ScanGraphEdges implements GraphEdgeScanner. It streams every row from the tenant-scoped graph_tXXXX edge table, calling fn once per row. Iteration stops on the first non-nil error returned by fn. Rows are read via the reader pool (query_only, parallel-safe). All tenants, including tenant 0, use graph_tXXXX.

func (*SQLiteStore) Search 

func (s *SQLiteStore) Search(ctx context.Context, entity string, field string, query string, matchType string) ([]map[string]interface{}, error)

Search implements field-based search using JSON extraction

func (*SQLiteStore) SearchEdges 

func (s *SQLiteStore) SearchEdges(ctx context.Context, query string, limit int) ([]EdgeFTSResult, error)

SearchEdges queries t<X>_efts using FTS5's MATCH operator and returns matching (rel, edge_id) pairs ordered by BM25 rank (best match first). limit ≤ 0 returns all matches.

func (*SQLiteStore) SetDxpClaims added in v0.26.0 

func (s *SQLiteStore) SetDxpClaims(c *dxp.MemCache)

SetDxpClaims wires this store into the dxp reservation cache (T-54, item 19): once set, FsmWalkInTx refuses to advance a machine that a live PESSIMISTIC dxp claim has locked mid-step ("every write path, not only the coordinator, must see dxp's holds"). nil (the default) is exact pre-dxp behaviour.

func (*SQLiteStore) StorageDialectFor 

func (s *SQLiteStore) StorageDialectFor(entity string) StorageDialect

StorageDialectFor returns the StorageDialect for the given entity, or nil if the entity is not adapted.

func (*SQLiteStore) SuppressEdgeSchemaWarning 

func (s *SQLiteStore) SuppressEdgeSchemaWarning(rel string)

SuppressEdgeSchemaWarning silences the unregistered-label warning for rel without writing anything to the database. The suppression is in-memory only and resets on restart.

func (*SQLiteStore) Update 

func (s *SQLiteStore) Update(ctx context.Context, entity string, id int, data map[string]interface{}) error

Update replaces an entity completely

func (*SQLiteStore) VerifyGraphIntegrity 

func (s *SQLiteStore) VerifyGraphIntegrity(ctx context.Context) error

first violation.

Both reads (" + s.nodesTable() + " and edge table) are issued inside a single read transaction so that concurrent writes cannot produce false violations.

Memory: only one map is materialised (expected edges derived from entity JSON). Actual edges from the edge table are streamed and checked against that map rather than accumulated into a second map.

func (*SQLiteStore) WithLogger 

func (s *SQLiteStore) WithLogger(logger zerolog.Logger) *SQLiteStore

WithLogger attaches a zerolog.Logger to the store. Returns the store so it can be chained: store := NewSQLiteStore(...).WithLogger(logger). Until this is called the store uses zerolog.Nop() and logs nothing.

func (*SQLiteStore) WriterDB 

func (s *SQLiteStore) WriterDB() *sql.DB

WriterDB implements storage.WriterDBProvider.

type SQLiteTenantPersister 

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

SQLiteTenantPersister implements tenant.Persister using the tenants table in the shared SQLite database. It uses the writer pool for saves and the reader pool for loads.

func NewSQLiteTenantPersister 

func NewSQLiteTenantPersister(db, readDB *sql.DB) *SQLiteTenantPersister

NewSQLiteTenantPersister creates a persister backed by the given database connections. Both must point to the same SQLite database. The writer is used for Save; the reader for LoadAll.

func (*SQLiteTenantPersister) LoadAll 

func (p *SQLiteTenantPersister) LoadAll(ctx context.Context) (map[string]tenant.TenantID, error)

LoadAll returns all persisted tenant name-to-ID mappings.

func (*SQLiteTenantPersister) Save 

func (p *SQLiteTenantPersister) Save(ctx context.Context, name string, id tenant.TenantID) error

Save persists a tenant mapping. Idempotent: re-saving the same (name, id) pair is not an error. Conflicts (same ID with different name, or same name with different ID) return an error.

type SchemaBrowser 

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

SchemaBrowser implements SchemaIntrospector over a queryfy ObjectSchema.

func (*SchemaBrowser) AllowsAdditional 

func (b *SchemaBrowser) AllowsAdditional() bool

func (*SchemaBrowser) FieldNames 

func (b *SchemaBrowser) FieldNames() []string

func (*SchemaBrowser) GetField 

func (b *SchemaBrowser) GetField(name string) FieldIntrospector

func (*SchemaBrowser) IsRequired 

func (b *SchemaBrowser) IsRequired(name string) bool

type SchemaDiff 

type SchemaDiff struct {
	Added   []ColumnDef    // Columns in new but not in old
	Dropped []ColumnDef    // Columns in old but not in new
	Changed []ColumnChange // Columns present in both but with incompatible type changes

	IndexesAdded   []IndexDef // Indexes in new but not in old
	IndexesDropped []IndexDef // Indexes in old but not in new

	HasExtraChanged bool // Whether _extra presence changed
	NewHasExtra     bool // The new HasExtra value (only meaningful if HasExtraChanged)
}

SchemaDiff describes the differences between an old and new AdaptedTableSpec. It is the migration plan for schema evolution.

func DiffAdaptedSpecs 

func DiffAdaptedSpecs(old, new *AdaptedTableSpec) *SchemaDiff

DiffAdaptedSpecs compares two AdaptedTableSpecs and produces a migration plan. The old spec represents the currently deployed table; the new spec represents the desired state from the updated schema.

func (*SchemaDiff) HasTypeConflicts 

func (d *SchemaDiff) HasTypeConflicts() bool

HasTypeConflicts reports whether there are incompatible type changes that prevent automatic migration.

func (*SchemaDiff) IsEmpty 

func (d *SchemaDiff) IsEmpty() bool

IsEmpty reports whether the diff contains no changes.

type SchemaIntrospector 

type SchemaIntrospector interface {
	// FieldNames returns all declared field names, sorted alphabetically.
	FieldNames() []string

	// GetField returns the field descriptor for a named field.
	// Returns nil if the field does not exist.
	GetField(name string) FieldIntrospector

	// IsRequired reports whether a field is required.
	IsRequired(name string) bool

	// AllowsAdditional reports whether the schema accepts fields not
	// declared in its field list. Returns true if additionalProperties
	// is absent or true; false if explicitly set to false.
	AllowsAdditional() bool
}

SchemaIntrospector provides read access to an object schema's structure. This is the primary interface consumed by DeriveAdaptedTableSpec.

func NewJSONSchemaIntrospector 

func NewJSONSchemaIntrospector(schema map[string]interface{}) SchemaIntrospector

NewJSONSchemaIntrospector wraps a parsed JSON Schema document. Returns nil if the schema has no "properties" key.

func NewSchemaBrowser 

func NewSchemaBrowser(obj *builders.ObjectSchema) SchemaIntrospector

NewSchemaBrowser wraps a queryfy ObjectSchema for introspection. Returns nil if obj is nil.

type Searcher 

type Searcher interface {
	Search(ctx context.Context, entity string, field string, query string, matchType string) ([]map[string]interface{}, error)
}

Searcher defines optional search capabilities

type StmtCache 

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

StmtCache is a concurrency-safe LRU cache of prepared statements.

The cache is keyed by the SQL string. Since the OQL planner produces parameterised queries with ? placeholders, two queries with different parameter values but the same shape share a single prepared statement.

Lifecycle:

  • Created during SQLiteStore initialisation.
  • Get() returns a cached *sql.Stmt or prepares and caches a new one.
  • Close() closes all cached statements (called from SQLiteStore.Close).
  • Invalidate() removes a single entry (for schema evolution).
  • Reset() closes and removes all entries.

func NewStmtCache 

func NewStmtCache(db *sql.DB, maxSize int) *StmtCache

NewStmtCache creates a statement cache that prepares against db. Pass maxSize=0 to use DefaultStmtCacheSize.

func (*StmtCache) Close 

func (c *StmtCache) Close()

Close closes all cached statements. After Close, the cache must not be used. Called from SQLiteStore.Close().

func (*StmtCache) Get 

func (c *StmtCache) Get(query string) (*sql.Stmt, error)

Get returns a prepared statement for the given SQL. If the statement is not cached, it is prepared and added to the cache, evicting the least-recently-used entry if the cache is full.

The returned *sql.Stmt must NOT be closed by the caller — the cache owns the statement's lifecycle.

func (*StmtCache) Invalidate 

func (c *StmtCache) Invalidate(query string)

Invalidate removes and closes the prepared statement for the given SQL, if cached. Used when a schema change invalidates a statement (e.g. adapted table ALTER TABLE).

func (*StmtCache) Len 

func (c *StmtCache) Len() int

Len returns the number of cached statements. Primarily for testing.

func (*StmtCache) Reset 

func (c *StmtCache) Reset()

Reset closes all cached statements and empties the cache. The cache remains usable — new statements will be prepared on demand.

type StorageDialect 

type StorageDialect interface {
	// Name returns the dialect identifier ("sqlite", "postgres", etc.).
	Name() string

	// Placeholder returns a parameter placeholder for the n-th argument
	// (1-based). SQLite: "?", PostgreSQL: "$1", "$2", etc.
	Placeholder(n int) string

	// ColumnType maps a JSON Schema type + format to the backend's native
	// SQL column type. Examples:
	//   ("string", "")        → "TEXT"
	//   ("integer", "")       → "INTEGER" (SQLite) / "BIGINT" (PostgreSQL)
	//   ("number", "")        → "REAL" (SQLite) / "DOUBLE PRECISION" (PostgreSQL)
	//   ("number", "decimal") → "TEXT" (SQLite) / "NUMERIC(p,s)" (PostgreSQL)
	//   ("boolean", "")       → "INTEGER" (SQLite) / "BOOLEAN" (PostgreSQL)
	//   ("array", "")         → "TEXT" (SQLite) / "JSONB" (PostgreSQL)
	//   ("object", "")        → "TEXT" (SQLite) / "JSONB" (PostgreSQL)
	//
	// For decimals, precision and scale are provided for backends that
	// support native fixed-point (PostgreSQL NUMERIC). Backends that store
	// decimals as text (SQLite) may ignore them.
	ColumnType(jsonType, format string, precision, scale int) string

	// CreateTableSQL generates the CREATE TABLE statement for an adapted
	// table. Implementations must include system columns (id, tenant_id,
	// _extra if hasExtra, _version, created_at, updated_at) and the
	// primary key.
	CreateTableSQL(spec *AdaptedTableSpec) string

	// CreateIndexSQL generates CREATE INDEX statements for the adapted
	// table's indexes.
	CreateIndexSQL(spec *AdaptedTableSpec) []string

	// InsertSQL generates an INSERT statement with the appropriate
	// placeholders. Returns the SQL string and the expected argument
	// order (column names). The caller provides the actual values.
	InsertSQL(spec *AdaptedTableSpec, hasExtra bool) (sql string, columns []string)

	// SelectSQL generates a SELECT statement for a single row by
	// tenant_id and id.
	SelectSQL(spec *AdaptedTableSpec) string

	// SelectAllSQL generates a SELECT statement for all rows in a tenant,
	// ordered by id.
	SelectAllSQL(spec *AdaptedTableSpec) string

	// UpdateSQL generates an UPDATE statement with the appropriate
	// placeholders. The versionCheck parameter controls whether a
	// _version = ? clause is appended to the WHERE.
	UpdateSQL(spec *AdaptedTableSpec, versionCheck bool) string

	// DeleteSQL generates a DELETE statement for a single row.
	DeleteSQL(spec *AdaptedTableSpec) string

	// ExistsSQL generates an EXISTS check for a single row.
	ExistsSQL(spec *AdaptedTableSpec) string

	// NodeSchemaTableSQL generates the DDL for the per-tenant node schema
	// registry table (t<X>_n_sch). Called lazily on first RegisterAdaptedTable
	// for a given tenant, not at store startup.
	NodeSchemaTableSQL(tenantID tenant.TenantID) string

	// NormaliseDecimal transforms a validated decimal string into the
	// storage representation for this backend. SQLite scales to int64.
	// PostgreSQL returns the value unchanged.
	NormaliseDecimal(value string, precision, scale int) (string, error)

	// DenormaliseDecimal transforms the stored representation back to
	// a client-facing string. SQLite divides by 10^scale and formats.
	// PostgreSQL returns the value unchanged.
	DenormaliseDecimal(value string, precision, scale int) string

	// SupportsNativeDecimalAggregation reports whether the backend
	// handles SUM/AVG on decimal columns with exact arithmetic.
	// If false, OQL aggregates in Go using shopspring/decimal.
	SupportsNativeDecimalAggregation() bool
}

StorageDialect defines backend-specific SQL generation for adapted tables.

type Store 

type Store interface {
	// Config returns the store's configuration.
	Config() StoreConfig

	// Entity CRUD operations
	Create(ctx context.Context, entity string, data map[string]interface{}) (int, error)
	Get(ctx context.Context, entity string, id int) (map[string]interface{}, error)
	// GetMany fetches multiple entities of the same type in a single query.
	// Returns a map from id → data for each id that was found; ids that do not
	// exist are absent from the result (no error). The caller must not assume
	// the result map contains all requested ids.
	GetMany(ctx context.Context, entity string, ids []int) (map[int]map[string]interface{}, error)
	Update(ctx context.Context, entity string, id int, data map[string]interface{}) error
	Patch(ctx context.Context, entity string, id int, data map[string]interface{}) error
	// PatchValidated is like Patch but runs a validation function against the
	// merged data inside the transaction. If the validator returns an error,
	// the transaction is rolled back and the error is returned to the caller.
	// This avoids TOCTOU races where a Get-merge-Update sequence can observe
	// stale data between the Get and the Update.
	PatchValidated(ctx context.Context, entity string, id int, data map[string]interface{}, validate func(merged map[string]interface{}) error) error
	Delete(ctx context.Context, entity string, id int) error
	// Save upserts an entity with the caller-specified ID: creates it if it
	// does not exist, overwrites it if it does. Returns (true, nil) when a
	// new record was created and (false, nil) when an existing record was
	// replaced. Never returns an error solely because the ID already exists.
	Save(ctx context.Context, entity string, id int, data map[string]interface{}) (bool, error)

	// Commit performs an atomic upsert + one or more inserts in a single
	// storage transaction. The upsert (req.Update) supports optional
	// optimistic concurrency via Version. Each entry in req.Append is an
	// unconditional insert; a duplicate explicit ID returns ErrAlreadyExists
	// and rolls back the entire commit. Returns ErrConflict when the Update
	// version check fails.
	Commit(ctx context.Context, req CommitRequest) (CommitResult, error)

	// Query operations
	List(ctx context.Context, entity string) ([]map[string]interface{}, error)
	Exists(ctx context.Context, entity string, id int) bool
	Search(ctx context.Context, entity string, field string, query string, matchType string) ([]map[string]interface{}, error)

	// Full-text search (optional - may return empty if not supported)
	FullTextSearch(ctx context.Context, query string, entity string) ([]map[string]interface{}, error)

	// Ping verifies that the storage backend is reachable. Returns nil on
	// success. Used by health and readiness probes.
	Ping(ctx context.Context) error

	// Lifecycle
	Close() error
}

Store defines the core interface for entity storage backends

func NewStore 

func NewStore(name string, config map[string]interface{}) (Store, error)

NewStore creates a new store instance by name

func NewStoreFromConfig 

func NewStoreFromConfig(cfg StoreConfig) (Store, error)

NewStoreFromConfig creates a store directly from a StoreConfig. This is the preferred constructor for tenant-scoped stores.

type StoreConfig 

type StoreConfig struct {
	Type            string          // "sqlite"
	BaseDir         string          // data root; layout paths are derived from this via pkg/storelayout
	DBPath          string          // resolved SQLite database file path (derived from BaseDir by the caller)
	FullTextEnabled bool            // controls FTS indexing in backend
	GraphEnabled    bool            // controls graph edge table maintenance
	TenantID        tenant.TenantID // 0 = no tenant scoping

	// Performance tuning (SQLite-specific; zero = use defaults)
	SQLiteCacheSize           int // Page cache size in KB
	SQLiteBusyTimeout         int // Milliseconds to wait on locked database
	SQLiteMaxOpenConns        int // Max open database connections
	SQLiteMaxIdleConns        int // Max idle database connections
	SQLiteReadPoolSize        int // Max open read connections (0 = auto)
	SQLiteContentionThreshold int // Adaptive lock threshold 0-100

	// SQLitePerFileTenants mirrors config.Config.SQLitePerFileTenants.
	// When true, each tenant gets its own SQLite database file. The flag
	// governs storeForTenant file routing in server.go and is unrelated to
	// the schema DDL (which uses t<XXXX>_* table names regardless of mode).
	SQLitePerFileTenants bool
}

StoreConfig is the canonical configuration for all store backends. A store is constructed with a StoreConfig and scoped to that config for its entire lifetime. TenantID 0 means no tenant scoping.

type StoreFactory 

type StoreFactory func(config map[string]interface{}) (Store, error)

StoreFactory is a function that creates a new Store instance

type StoreInfo 

type StoreInfo struct {
	Type                string // "sqlite"
	Version             string
	SupportsSearch      bool
	SupportsBatch       bool
	SupportsTransaction bool
}

StoreInfo provides metadata about the store implementation

type TableNamer 

type TableNamer interface {
	NodesTable() string
}

TableNamer provides the tenant-scoped table name for the blob node store. The OQL SQL generator uses this to build correct push-down queries without hardcoding "entities". Implementations return tenant.NodesTableName(tenantID).

type TenantIDLister 

type TenantIDLister interface {
	GraphTenantIDs(ctx context.Context) ([]tenant.TenantID, error)
}

enumerate all tenant IDs for which a graph_tXXXX edge table exists. The returned slice must always include tenant 0 (the implicit default). Used during startup graph hydration to restore graph state for all tenants.

Backends that do not implement this interface fall back to scanning only tenant 0 via a direct ScanGraphEdges call.

type TenantModeProvider 

type TenantModeProvider interface {
	IsPerFileTenant() bool
}

TenantModeProvider is implemented by storage backends that support per-file tenant isolation. The OQL layer uses this to decide whether to inject tenant_id scoping into pushed-down SQL queries.

type V2SchemaInitialiser 

type V2SchemaInitialiser interface {
	InitV2Schema(ctx context.Context) error
}

V2SchemaInitialiser is implemented by storage backends that support the API v2 schema. The server calls InitV2Schema once on startup when XOLU_API_V2_ENABLED is true, before registering any v2 routes. The call is idempotent; stores that do not implement this interface cause v2 initialisation to be skipped with a warning.

type WriterDBProvider 

type WriterDBProvider interface {
	WriterDB() *sql.DB
}

WriterDBProvider gives access to the underlying write connection pool. Used by v2 handlers that need direct SQL access to global tables like entity_meta which are not modelled in the Store interface.

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.