storage

package
v1.1.2 Latest Latest
Warning

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

 Go to latest Published: May 26, 2026 License: MIT Imports: 41 Imported by: 0

Documentation

Overview

Package storage - AsyncEngine provides write-behind caching for eventual consistency.

AsyncEngine wraps a storage engine and provides:

  • Immediate writes to in-memory cache (fast)
  • Background writes to underlying engine (async)
  • Reads check cache first, then engine (eventual consistency)

Trade-offs:

  • Much faster writes (returns immediately)
  • Reads may see stale data briefly (eventual consistency)
  • Data loss risk if crash before flush (use with WAL for durability)

Package storage provides storage engine implementations for NornicDB.

BadgerEngine provides persistent disk-based storage using BadgerDB. It implements the Engine interface with full ACID transaction support.

Package storage provides storage engine implementations for NornicDB.

Package storage provides storage engine implementations for NornicDB.

Package storage provides storage engine implementations for NornicDB.

Package storage provides storage engine implementations for NornicDB.

Package storage provides storage engine implementations for NornicDB.

Plan 04-04-06: BadgerEngine metric attachment + per-op observation helpers.

AttachMetrics binds the StorageMetrics + MVCCMetrics bags into the engine and pre-binds the four op-duration observers (MET-25). Subsequent hot-path Get/Put/Delete/Scan calls use observeStorageOp(start, observer) which is zero-overhead when metrics are not attached (the BoundLatency Observer's zero value short-circuits via the typed wrapper's Observe path).

D-16 boundary: storage layer never imports observability for ErrConflict counters — that wiring lives in pkg/cypher (Plan 04-03). This file is the LIMITED storage→observability dependency: storage emits its OWN subsystem families only (op_duration_seconds, bytes, compactions, index_rebuild). Cross-layer counters (transaction_conflicts) stay at the Cypher transaction wrapper.

Package storage provides storage engine implementations for NornicDB.

Package storage provides storage engine implementations for NornicDB.

Package storage - Serialization helpers for BadgerDB.

All new writes use msgpack with a small header. Legacy gob bodies (no header) are still decodable for one purpose only: the offline in-place migration tool that rewrites them as msgpack. The encoder NEVER emits gob.

Package storage provides storage engine implementations for NornicDB.

Package storage - BadgerDB transaction wrapper with ACID guarantees.

This file implements atomic transactions for BadgerDB with full constraint validation and rollback support.

Plan 04-04-04: D-07 30s sweep lifecycle.Component populating nornicdb_storage_bytes{kind} gauges by calling badger.DB.EstimateSize(prefix) for each kind bucket.

Lifecycle semantics (RESEARCH §Architecture Pattern 2):

  • Start spawns a goroutine that ticks every interval (default 30s).
  • Each tick wraps `defer recover()` so a transient Badger panic during compaction does not crash the supervisor errgroup (RISK-8).
  • Shutdown closes the ticker and waits for the goroutine to exit within the supervisor's drain budget (5s typical).

Five `kind` buckets (closed enum AllowedStorageBytesKinds):

  • nodes — prefixNode
  • edges — prefixEdge
  • index — prefixLabelIndex + prefixEdgeBetweenIndex + prefixTemporalIndex (sum across the three index prefixes; user-created indexes are not separately accounted by D-13c)
  • wal — heuristic (vlog size - lsm size from db.Size()); RISK-6
  • search — sum of IndexSizeBytes() across all per-database search services (Plan 04-05 owns the IndexSizeBytes accessor; this sweeper consumes via the SearchSizeFn callback so the storage layer never imports pkg/search)

The sweeper is registered in cmd/nornicdb between pprof and workers components per RESEARCH §Q4 — i.e. it Drains AFTER the workers and BEFORE the telemetry listener, so the final scrape during drain still reflects the last known sizes.

Package storage provides composite engine for composite database support.

CompositeEngine implements the Engine interface by routing operations to multiple constituent databases and merging results transparently.

Package storage provides routing logic for composite engines.

This file implements query analysis and routing for composite databases, determining which constituents should be queried based on query patterns.

Package storage - Constraint validation when constraints are created.

Package storage provides edge provenance logging for NornicDB.

Edge provenance tracks the audit trail for edges - why they were created, when, what evidence supports them, and their lifecycle state.

Feature flag: NORNICDB_EDGE_PROVENANCE_ENABLED (enabled by default)

Usage: 

store := NewEdgeMetaStore()
meta := EdgeMeta{
    Src:        "node-A",
    Dst:        "node-B",
    Label:      "relates_to",
    Score:      0.85,
    SignalType: "similarity",
}
store.Append(ctx, meta)

// Query history
history, _ := store.GetHistory(ctx, "node-A", "node-B", "relates_to")

Plan 04-04-05: D-13c index-name → enum mapping.

classifyIndexName is the pure function that maps an internal index identifier (e.g. "label_Person", "edge_between_KNOWS", "temporal_user_activity", "embedding_chunk", or any user-named string) to one of the closed-enum buckets accepted by observability.AllowedStorageIndexes: 

{"label", "edge_between", "temporal", "embedding", "user_created"}

Anything not matching a known prefix buckets to "user_created" — the keystone of T-04-02 mitigation. Drives 1k arbitrary user index names → cardinality stays at 5.

Pure function, no allocations, safe in hot paths.

Package storage provides storage implementations and data import/export functionality.

This file specifically handles Neo4j JSON import/export functionality, enabling NornicDB to interoperate with Neo4j databases through standard JSON export formats.

Supported Formats:

  • Neo4j APOC JSON exports (nodes.json + relationships.json)
  • Combined Neo4j export format (single JSON file)
  • NornicDB native format with full fidelity

Key Features:

  • Bidirectional Neo4j compatibility
  • Bulk loading for performance
  • Property type preservation
  • Label and relationship type mapping
  • Error handling and validation

Example Usage: 

// Load from Neo4j APOC export
engine := storage.NewMemoryEngine()
err := storage.LoadFromNeo4jJSON(engine, "./neo4j-export/")
if err != nil {
	log.Fatal(err)
}

// Load from combined export file
err = storage.LoadFromNeo4jExport(engine, "./data.json")
if err != nil {
	log.Fatal(err)
}

// Export to Neo4j format
err = storage.SaveToNeo4jExport(engine, "./nornicdb-export.json")
if err != nil {
	log.Fatal(err)
}

Neo4j APOC Export Format:

The APOC export format consists of two files:

  • nodes.json: One JSON object per line, each representing a node
  • relationships.json: One JSON object per line, each representing a relationship

Example nodes.json: 

{"id":"0","labels":["Person"],"properties":{"name":"Alice","age":30}}
{"id":"1","labels":["Person"],"properties":{"name":"Bob","age":25}}

Example relationships.json: 

{"id":"0","type":"KNOWS","startNode":"0","endNode":"1","properties":{"since":2020}}

Combined Export Format:

The combined format includes both nodes and relationships in a single JSON file: 

{
  "nodes": [...],
  "relationships": [...]
}

Data Type Mapping:

Neo4j types are mapped to Go types as follows:

  • String -> string
  • Integer -> int64
  • Float -> float64
  • Boolean -> bool
  • Array -> []interface{}
  • Object -> map[string]interface{}

Performance:

  • Bulk operations are used for efficient loading
  • Streaming JSON parsing for large files
  • Memory-efficient processing
  • Progress reporting for large imports

ELI12 (Explain Like I'm 12):

Think of this like moving between different types of photo albums:

  1. **Neo4j format**: Like a specific brand of photo album with a special way of organizing photos (nodes) and the connections between them (relationships).

  2. **Loading**: Like taking photos from a Neo4j album and putting them into a NornicDB album, making sure each photo goes in the right place and keeps all its information.

  3. **Exporting**: Like taking photos from a NornicDB album and organizing them in the Neo4j format so they can be used in Neo4j tools.

  4. **Bulk operations**: Instead of moving photos one by one, we move whole pages at a time to make it much faster.

This lets you easily move your data between NornicDB and Neo4j!

Package storage provides storage engine implementations for NornicDB.

The storage package defines the Engine interface and provides multiple implementations:

  • MemoryEngine: In-memory storage using BadgerDB's in-memory mode (for testing)
  • BadgerEngine: Persistent disk-based storage

All storage engines are thread-safe and support concurrent operations.

Example Usage: 

// Create in-memory storage (for testing)
engine := storage.NewMemoryEngine()
defer engine.Close()

// Create a node
node := &storage.Node{
	ID:     "user-001",
	Labels: []string{"User"},
	Properties: map[string]any{
		"name": "Alice",
	},
}
engine.CreateNode(node)

Package storage provides namespaced storage engine wrapper for multi-database support.

NamespacedEngine wraps any storage.Engine with automatic key prefixing for database isolation. This enables multiple logical databases (tenants) to share a single physical storage backend while maintaining complete data isolation.

Key Design:

  • All node and edge IDs are prefixed with the namespace: "tenant_a:123" instead of "123"
  • Queries only see data in the current namespace
  • DROP DATABASE = delete all keys with namespace prefix

Thread Safety: 

Delegates to underlying engine's thread safety guarantees.

Example: 

inner := storage.NewBadgerEngine("./data")
tenantA := storage.NewNamespacedEngine(inner, "tenant_a")

// Creates node with ID "tenant_a:123" in BadgerDB
tenantA.CreateNode(&Node{ID: "123", Labels: []string{"Person"}})

// Only sees nodes with "tenant_a:" prefix
nodes, _ := tenantA.AllNodes()

Package storage provides per-node configuration for NornicDB inference.

Per-node config allows fine-grained control over edge materialization: - Pin list: edges to these targets never decay - Deny list: never create edges to these targets - Edge caps: maximum edges per node (in/out/total) - Per-label caps: limits on specific edge types - Trust level: affects confidence thresholds

Feature flags:

  • NORNICDB_PER_NODE_CONFIG_ENABLED=true (enabled by default)
  • NORNICDB_PER_NODE_CONFIG_AUTO_INTEGRATION_ENABLED=true (enabled by default)

Real-World Example 1: User preferences (social network) 

// User wants max 50 friends, never connect to blocked users
store := storage.NewNodeConfigStore()
userConfig := storage.NewNodeConfig("user-alice")
userConfig.MaxOutEdges = 50  // Max 50 outgoing friendships
userConfig.DenyList = []string{"user-spammer", "user-troll"}  // Blocked users
userConfig.PinList = []string{"user-bestfriend"}  // Never decay this friendship
store.Set(userConfig)

// Later, inference engine checks before creating edge
if allowed, _ := store.IsEdgeAllowedWithReason("user-alice", "user-bob", "friend"); allowed {
    db.CreateEdge("user-alice", "user-bob", "friend")
    store.RecordEdgeCreation("user-alice", "user-bob")  // Update count (now at 23/50)
}

Real-World Example 2: Document categorization limits (knowledge base) 

// Document should have max 5 categories, but unlimited references
docConfig := storage.NewNodeConfig("doc-123")
docConfig.LabelConfigs = map[string]storage.LabelConfig{
    "category": {MaxEdges: 5},        // Max 5 category tags
    "references": {MaxEdges: 0},      // Unlimited references (0 = no limit)
    "deprecated": {Disabled: true},   // Never create deprecated edges
}
store.Set(docConfig)

// Try to add 6th category - denied!
allowed, reason := store.IsEdgeAllowedWithReason("doc-123", "category-ai", "category")
// → (false, "label 'category' at max capacity (5/5)")

Real-World Example 3: Low-trust nodes (spam prevention) 

// New users start with low trust - require higher confidence
newUserConfig := storage.NewNodeConfig("user-newbie")
newUserConfig.TrustLevel = storage.TrustLevelLow  // Requires +20% confidence
newUserConfig.MaxOutEdges = 10  // Limited connections until trust increases
store.Set(newUserConfig)

// After user proves trustworthy, upgrade trust
if userIsActive && userNotSpamming {
    newUserConfig.TrustLevel = storage.TrustLevelDefault
    newUserConfig.MaxOutEdges = 100
    store.Set(newUserConfig)  // Update config
}

ELI12 (Explain Like I'm 12):

Per-node config is like house rules for each person:

**Pin List**: "These are my best friends - never unfriend them!"

  • In NornicDB: Edges in pin list never decay, always kept

**Deny List**: "I never want to talk to these people"

  • In NornicDB: Never create edges to denied nodes (like blocking someone)

**Edge Caps**: "I can only have 50 friends max"

  • In NornicDB: Limits prevent one node from connecting to everything

**Trust Level**: "I just moved here, so teachers are extra careful with me"

  • Low trust: Requires stronger evidence before creating edges
  • High trust: Can create edges more easily
  • Pinned: Always trust (like family members)

Think of it like a bouncer at a party:

  • Deny list = banned from party
  • Pin list = VIP, always allowed
  • Max edges = party capacity (50 people max)
  • Trust level = how strict the bouncer is checking IDs

Package storage provides receipt generation for mutation auditing.

Package storage schema management for constraints and indexes.

This file implements Neo4j-compatible schema management including:

  • Unique constraints
  • Property indexes (single and composite)
  • Range indexes (for efficient range queries)
  • Full-text indexes
  • Vector indexes

Schema definitions are stored in memory and enforced during node operations.

Package storage - Transaction types shared between storage implementations.

This file defines shared transaction types used by BadgerTransaction. All transactions in NornicDB use Badger's native transaction system which provides real ACID guarantees through the Write-Ahead Log (WAL).

ACID Guarantees

Transactions provide:

  • Atomicity: All operations commit together or none do
  • Consistency: Constraints validated before commit
  • Isolation: Changes invisible until commit
  • Durability: WAL ensures persistence even on crash

Usage

All engines (including MemoryEngine) use BadgerTransaction: 

engine := storage.NewMemoryEngine() // or NewBadgerEngine()
tx, err := engine.BeginTransaction()
if err != nil {
    return err
}
defer tx.Rollback() // Rollback if not committed

tx.CreateNode(&Node{ID: "n1", Labels: []string{"Person"}})
tx.CreateNode(&Node{ID: "n2", Labels: []string{"Person"}})

return tx.Commit() // Atomic - both succeed or both fail

Package storage provides the storage engine interface and implementations for NornicDB.

The storage layer is designed for Neo4j compatibility while adding NornicDB-specific extensions for memory decay, vector embeddings, and automatic relationship inference.

Design Principles:

  • Neo4j JSON export/import compatibility
  • Testability through dependency injection
  • Thread-safe implementations
  • Property graph model (labeled property graph)

Example Usage: 

// Create storage engine
engine := storage.NewMemoryEngine()
defer engine.Close()

// Create nodes
node := &storage.Node{
	ID:     storage.NodeID("user-123"),
	Labels: []string{"User", "Person"},
	Properties: map[string]any{
		"name":  "Alice",
		"email": "alice@example.com",
	},
	CreatedAt: time.Now(),
}
engine.CreateNode(node)

// Create relationships
edge := &storage.Edge{
	ID:        storage.EdgeID("follows-1"),
	StartNode: storage.NodeID("user-123"),
	EndNode:   storage.NodeID("user-456"),
	Type:      "FOLLOWS",
	CreatedAt: time.Now(),
}
engine.CreateEdge(edge)

// Export to Neo4j format
nodes, _ := engine.AllNodes()
edges, _ := engine.AllEdges()
export := storage.ToNeo4jExport(nodes, edges)

// Save as JSON
data, _ := json.MarshalIndent(export, "", "  ")
os.WriteFile("graph-export.json", data, 0644)

Package storage provides write-ahead logging for NornicDB durability.

WAL (Write-Ahead Logging) ensures crash recovery by logging all mutations before they are applied to the storage engine. Combined with periodic snapshots, this provides:

  • Durability: No data loss on crash
  • Recovery: Restore state from snapshot + WAL replay
  • Audit trail: Complete history of all mutations

Feature flag: NORNICDB_WAL_ENABLED (enabled by default)

Usage: 

// Create WAL-backed storage
engine := NewMemoryEngine()
wal, err := NewWAL("/path/to/wal", nil)
walEngine := NewWALEngine(engine, wal)

// Operations are logged before execution
walEngine.CreateNode(&Node{ID: "n1", ...})

// Create periodic snapshots
snapshot, err := wal.CreateSnapshot(engine)
wal.SaveSnapshot(snapshot, "/path/to/snapshot.json")

// Recovery after crash
engine, err = RecoverFromWAL("/path/to/wal", "/path/to/snapshot.json")

Package storage provides write-ahead logging for NornicDB durability.

Package storage provides write-ahead logging for NornicDB durability.

Index

Constants

View Source 
const (
	ConstraintContractKindPrimitiveNode         = "primitive-node"
	ConstraintContractKindPrimitiveRelationship = "primitive-relationship"
	ConstraintContractKindBooleanNode           = "boolean-node"
	ConstraintContractKindBooleanRelationship   = "boolean-relationship"
)
View Source 
const DefaultBytesMetricsInterval = 30 * time.Second

DefaultBytesMetricsInterval is the D-07 default sweep cadence.

View Source 
const DefaultRetentionPolicyMaxVersionsPerKey = 0

DefaultRetentionPolicyMaxVersionsPerKey — closed historical versions preserved per key by default. Zero means "no history, head-only" which collapses every write to a single primary-key write (no MVCC version archival). Operators who need audit/rollback raise NORNICDB_MVCC_RETENTION_MAX_VERSIONS via config.

Variables

View Source 
var (
	ErrNoTransaction       = errors.New("no active transaction")
	ErrTransactionActive   = errors.New("transaction already active")
	ErrTransactionClosed   = errors.New("transaction already closed")
	ErrTransactionRollback = errors.New("transaction rolled back")
)

Transaction errors

View Source 
var (
	ErrNotFound         = errors.New("not found")
	ErrAlreadyExists    = errors.New("already exists")
	ErrConflict         = errors.New("conflict")
	ErrExhausted        = errors.New("exhausted")
	ErrInvalidID        = errors.New("invalid id")
	ErrInvalidData      = errors.New("invalid data")
	ErrNotImplemented   = errors.New("not implemented")
	ErrInvalidEdge      = errors.New("invalid edge: start or end node not found")
	ErrStorageClosed    = errors.New("storage closed")
	ErrIterationStopped = errors.New("iteration stopped") // Sentinel to stop streaming early
	// ErrCrossNamespaceTransaction is returned when a single transaction
	// attempts to mix writes from multiple namespaces. The transaction layer
	// pins each transaction to one namespace at the first prefixed write, and
	// every subsequent write must share that prefix. Per-database MVCC
	// counters depend on this invariant.
	ErrCrossNamespaceTransaction = errors.New("transaction spans multiple namespaces")
	// ErrNotVisibleAtSnapshot signals that the entity exists (a head
	// record is present) but is not visible to the snapshot version
	// the caller passed. Distinct from ErrNotFound, which means the
	// entity has no head at all. Callers that maintain transaction
	// snapshot isolation MUST treat this as a hard miss rather than
	// falling back to a fresh-view read of the primary key, which
	// would expose peer commits that landed between the reader's
	// begin and the read.
	ErrNotVisibleAtSnapshot = errors.New("entity not visible at snapshot")
)

Common errors

View Source 
var (
	ErrMVCCSnapshotGracefulCancel = errors.New("mvcc: snapshot cancelled due to resource pressure")
	ErrMVCCSnapshotHardExpired    = errors.New("mvcc: snapshot forcibly expired due to critical resource pressure")
)

Snapshot expiration errors surface when an already-admitted reader is forced to stop.

View Source 
var (
	ErrWALClosed         = errors.New("wal: closed")
	ErrWALCorrupted      = errors.New("wal: corrupted entry")
	ErrWALPartialWrite   = errors.New("wal: partial write detected")
	ErrWALChecksumFailed = errors.New("wal: checksum verification failed")
	ErrWALMissingTrailer = errors.New("wal: missing or invalid trailer (incomplete write)")
	ErrSnapshotFailed    = errors.New("wal: snapshot creation failed")
	ErrRecoveryFailed    = errors.New("wal: recovery failed")
)

Common WAL errors

View Source 
var ErrMVCCResourcePressure = errors.New("mvcc: resource pressure exceeded snapshot lifetime")

ErrMVCCResourcePressure is returned when MVCC resource pressure exceeds snapshot lifetime.

View Source 
var ErrNoUndoData = errors.New("wal: entry has no undo data")

UndoWALEntry reverses a WAL entry using its stored "before image". This is used for transaction rollback on crash recovery. Returns ErrNoUndoData if the entry lacks undo information.

Functions

func BulkCreateEdgesForRecovery added in v1.1.2 

func BulkCreateEdgesForRecovery(engine Engine, edges []*Edge) error

BulkCreateEdgesForRecovery creates recovered edges while adapting to backend transaction limits. A single oversized edge still returns the original error.

func BulkCreateNodesForRecovery added in v1.1.2 

func BulkCreateNodesForRecovery(engine Engine, nodes []*Node) error

BulkCreateNodesForRecovery creates recovered nodes while adapting to backend transaction limits. It preserves the normal bulk path unless the backend says the requested transaction is too large.

func ClassifyIndexName added in v1.1.0 

func ClassifyIndexName(internal string) string

ClassifyIndexName is the exported alias used by the observation sites in pkg/cypher / cmd/nornicdb / future plan 04-05 search-engine plumbing. The internal-only `classifyIndexName` is the single source of truth; this alias keeps the public API surface uppercase per Go convention without duplicating the closed-enum logic.

func CollectEdgeTypes 

func CollectEdgeTypes(ctx context.Context, engine Engine) ([]string, error)

CollectEdgeTypes collects all unique edge types using streaming.

func CollectLabels 

func CollectLabels(ctx context.Context, engine Engine) ([]string, error)

CollectLabels collects all unique labels using streaming.

func CountNodesWithLabel 

func CountNodesWithLabel(ctx context.Context, engine Engine, label string) (int64, error)

CountNodesWithLabel counts nodes with a specific label using streaming.

func DecayScoringTime added in v1.1.0 

func DecayScoringTime() int64

DecayScoringTime returns a frozen nanosecond timestamp for use as the scoring time across a single query. Call once per query and pass to all filter calls.

func EnsureDatabasePrefix 

func EnsureDatabasePrefix(dbName, id string) string

EnsureDatabasePrefix adds "<dbName>:" to id if id has no existing valid prefix.

If id already has a prefix (even a different database), id is returned unchanged to avoid accidentally rewriting cross-database IDs.

func ExtractNamespaceFromID added in v1.1.0 

func ExtractNamespaceFromID(id string) string

ExtractNamespaceFromID returns the namespace prefix of an entity ID.

func FromNeo4jExport 

func FromNeo4jExport(export *Neo4jExport) ([]*Node, []*Edge)

FromNeo4jExport converts Neo4j JSON export format to NornicDB nodes and edges.

This function imports data exported from Neo4j, extracting NornicDB-specific properties (those with "_" prefix) back into their dedicated fields.

Supports both export formats:

  • neo4j-admin database dump (flat format)
  • apoc.export.json (nested format)

Example: 

// Load Neo4j export file
data, _ := os.ReadFile("neo4j-export.json")

var export storage.Neo4jExport
json.Unmarshal(data, &export)

// Convert to NornicDB format
nodes, edges := storage.FromNeo4jExport(&export)

// Import into NornicDB
if err := engine.BulkCreateNodes(nodes); err != nil {
	log.Fatal(err)
}
if err := engine.BulkCreateEdges(edges); err != nil {
	log.Fatal(err)
}

// emit "Imported N nodes, M edges" via the configured logger

Returns nodes and edges ready for storage engine insertion.

func GenericSaveToNeo4jExport 

func GenericSaveToNeo4jExport(engine ExportableEngine, path string) error

GenericSaveToNeo4jExport works with any ExportableEngine.

func GetEntryTxID 

func GetEntryTxID(entry WALEntry) string

GetEntryTxID extracts the transaction ID from a WAL entry, if present.

func LoadFromNeo4jExport 

func LoadFromNeo4jExport(engine Engine, path string) error

LoadFromNeo4jExport loads data from a combined Neo4j export file.

This function loads data from a single JSON file containing both nodes and relationships in a combined format. This is the format produced by SaveToNeo4jExport() and provides a more compact representation.

Parameters:

  • engine: Storage engine to load data into
  • path: Path to the combined JSON export file

Returns:

  • Error if file cannot be read or data is invalid

Example: 

// Load from combined export file
engine := storage.NewMemoryEngine()
err := storage.LoadFromNeo4jExport(engine, "./data-export.json")
if err != nil {
	log.Fatalf("Failed to load export: %v", err)
}

// emit "Successfully loaded data from export file" via the configured logger

File Format: 

{
  "nodes": [
    {"id":"0","labels":["Person"],"properties":{"name":"Alice"}},
    {"id":"1","labels":["Person"],"properties":{"name":"Bob"}}
  ],
  "relationships": [
    {"id":"0","type":"KNOWS","startNode":"0","endNode":"1","properties":{}}
  ]
}

Performance:

  • Loads entire file into memory for parsing
  • Uses bulk operations for efficient storage
  • Suitable for moderate-sized datasets

Use Cases:

  • Restoring NornicDB backups
  • Migrating data between environments
  • Loading test datasets
  • Data exchange with Neo4j systems

func LoadFromNeo4jJSON 

func LoadFromNeo4jJSON(engine Engine, dir string) error

LoadFromNeo4jJSON loads nodes and edges from Neo4j APOC JSON export format.

This function reads the standard Neo4j APOC export format consisting of nodes.json and relationships.json files. This format is produced by Neo4j's APOC library using `apoc.export.json.all()` or similar procedures.

File Format:

  • nodes.json: One JSON object per line (JSONL format)
  • relationships.json: One JSON object per line (JSONL format)

Parameters:

  • engine: Storage engine to load data into
  • dir: Directory containing nodes.json and relationships.json files

Returns:

  • Error if files cannot be read or data is invalid

Example: 

// Load from Neo4j APOC export directory
engine := storage.NewMemoryEngine()
err := storage.LoadFromNeo4jJSON(engine, "./neo4j-export/")
if err != nil {
	log.Fatalf("Failed to load Neo4j data: %v", err)
}

// emit "Successfully loaded Neo4j data into NornicDB" via the configured logger

Expected Directory Structure: 

neo4j-export/
├── nodes.json
└── relationships.json

Node Format (nodes.json): 

{"id":"0","labels":["Person"],"properties":{"name":"Alice","age":30}}
{"id":"1","labels":["Company"],"properties":{"name":"Acme Corp"}}

Relationship Format (relationships.json): 

{"id":"0","type":"WORKS_FOR","startNode":"0","endNode":"1","properties":{"since":2020}}

Performance:

  • Processes files sequentially (nodes first, then relationships)
  • Uses streaming JSON parsing for memory efficiency
  • Bulk operations for optimal storage performance

func NodeNeedsEmbedding 

func NodeNeedsEmbedding(node *Node) bool

NodeNeedsEmbedding checks if a node needs an embedding to be generated. Returns true if the node should have an embedding generated, false if it should be skipped.

A node is skipped (returns false) if:

  • It has an internal label (starts with '_')
  • It already has an embedding
  • It has the "embedding_skipped" property set
  • It has "has_embedding" property explicitly set to false

Example: 

for _, node := range nodes {
    if storage.NodeNeedsEmbedding(node) {
        generateEmbedding(node)
    }
}

func ParseDatabasePrefix 

func ParseDatabasePrefix(id string) (db string, unprefixed string, ok bool)

ParseDatabasePrefix splits an ID formatted as "<db>:<id>" into database and unprefixed ID.

Returns ok=false if there is no valid prefix (no ':', empty db, or empty id).

func PruneOldSnapshotFiles 

func PruneOldSnapshotFiles(dir string, cfg *WALConfig) error

PruneOldSnapshotFiles removes old snapshot files in dir according to cfg retention. Keeps at most SnapshotRetentionMaxCount snapshots (newest by mtime) and deletes any older than SnapshotRetentionMaxAge. Idempotent; safe to call after each save.

func RecoverFromWALWithResult 

func RecoverFromWALWithResult(walDir, snapshotPath string) (*MemoryEngine, ReplayResult, error)

RecoverFromWALWithResult recovers database state and returns detailed results. Use this for programmatic access to replay statistics and errors.

func RecoverWithTransactions 

func RecoverWithTransactions(walDir, snapshotPath string) (*MemoryEngine, *TransactionRecoveryResult, error)

RecoverWithTransactions performs transaction-aware WAL recovery. Incomplete transactions (no commit/abort) are rolled back using undo data. Returns the engine state and recovery statistics.

func RefreshUniqueConstraintValuesForEngine added in v1.0.43 

func RefreshUniqueConstraintValuesForEngine(engine Engine, schema *SchemaManager) error

RefreshUniqueConstraintValuesForEngine rebuilds single-property UNIQUE value caches from the engine after a schema mutation has been admitted.

func ReplayWALEntry 

func ReplayWALEntry(engine Engine, entry WALEntry) error

ReplayWALEntry applies a single WAL entry to the engine. Uses the database name from the entry if present, otherwise wraps with default database.

func ResetGlobalEdgeMetaStore 

func ResetGlobalEdgeMetaStore()

ResetGlobalEdgeMetaStore resets the global edge meta store. Primarily for testing.

func ResetGlobalNodeConfigStore 

func ResetGlobalNodeConfigStore()

ResetGlobalNodeConfigStore resets the global node config store. Primarily for testing.

func SaveSnapshot 

func SaveSnapshot(snapshot *Snapshot, path string) error

SaveSnapshot writes a snapshot to disk with full durability guarantees. Uses write-to-temp + atomic-rename pattern for crash safety.

func SaveToNeo4jExport 

func SaveToNeo4jExport(engine Engine, path string) error

SaveToNeo4jExport exports all data from the storage engine to a Neo4j-compatible JSON file.

This function creates a complete export of all nodes and relationships in the storage engine, formatted for compatibility with Neo4j tools and for backup/restore operations.

Parameters:

  • engine: Storage engine to export data from
  • path: Output file path for the JSON export

Returns:

  • Error if export fails or file cannot be written

Example: 

// Export all data to JSON file
err := storage.SaveToNeo4jExport(engine, "./backup.json")
if err != nil {
	log.Fatalf("Export failed: %v", err)
}

// emit "Data exported successfully" via the configured logger

// The exported file can be loaded back with:
// storage.LoadFromNeo4jExport(newEngine, "./backup.json")

Output Format: 

{
  "nodes": [
    {
      "id": "node-123",
      "labels": ["Person", "Employee"],
      "properties": {
        "name": "Alice Smith",
        "age": 30,
        "email": "alice@example.com"
      }
    }
  ],
  "relationships": [
    {
      "id": "rel-456",
      "type": "WORKS_FOR",
      "startNode": "node-123",
      "endNode": "node-789",
      "properties": {
        "since": "2020-01-15",
        "role": "Developer"
      }
    }
  ]
}

Use Cases:

  • Creating backups of NornicDB data
  • Migrating data to Neo4j
  • Data analysis with Neo4j tools
  • Sharing datasets in a standard format

Performance:

  • Reads all data into memory before export
  • Uses pretty-printed JSON for readability
  • Suitable for datasets that fit in memory

Note: Currently only supports MemoryEngine. Other storage engines will return an error indicating unsupported engine type.

func StreamEdgesWithFallback 

func StreamEdgesWithFallback(ctx context.Context, engine Engine, chunkSize int, fn EdgeVisitor) error

StreamEdgesWithFallback provides streaming iteration with fallback.

func StreamNodesWithFallback 

func StreamNodesWithFallback(ctx context.Context, engine Engine, chunkSize int, fn NodeVisitor) error

StreamNodesWithFallback provides streaming iteration with fallback. If the engine supports StreamingEngine, it uses that. Otherwise, it loads all nodes but processes them in chunks.

func StripDatabasePrefix 

func StripDatabasePrefix(dbName, id string) string

StripDatabasePrefix removes "<db>:" from id only if it matches dbName. If dbName is empty, or id does not have the matching prefix, id is returned unchanged.

func UndoWALEntry 

func UndoWALEntry(engine Engine, entry WALEntry) error

func ValidateConstraintContractOnCreationForEngine 

func ValidateConstraintContractOnCreationForEngine(engine Engine, contract ConstraintContract) error

func ValidateConstraintOnCreationForEngine 

func ValidateConstraintOnCreationForEngine(engine Engine, c Constraint) error

ValidateConstraintOnCreationForEngine validates constraints using the Engine interface. This allows callers (like Cypher) to validate through wrapper engines (namespaced, WAL, etc.).

func ValidatePropertyType 

func ValidatePropertyType(value interface{}, expectedType PropertyType) error

ValidatePropertyType checks if a value matches the expected type. Handles JSON/MessagePack serialization quirks where integers become float64.

func ValidatePropertyTypeConstraintOnCreationForEngine 

func ValidatePropertyTypeConstraintOnCreationForEngine(engine Engine, ptc PropertyTypeConstraint) error

ValidatePropertyTypeConstraintOnCreationForEngine validates type constraints using Engine.

Types

type AdjacentEdgesEngine added in v1.1.2 

type AdjacentEdgesEngine interface {
	GetAdjacentEdges(nodeID NodeID) (outgoing, incoming []*Edge, err error)
}

AdjacentEdgesEngine is an optional extension interface for fetching both directions of edges incident to a node in a single underlying transaction.

BFS-style traversals (shortestPath, variable-length MATCH) historically called GetOutgoingEdges + GetIncomingEdges per frontier node, opening two fresh Badger view transactions for each. With ~500-1000 frontier nodes per shortestPath request, that adds 1000-2000 transaction opens — the dominant per-request fixed cost the profile flagged. Engines that can fold both directions into a single view should implement this; callers fall back to the pair of single-direction calls when an engine doesn't.

type AsyncEngine 

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

AsyncEngine wraps a storage engine with write-behind caching. Writes return immediately after updating the cache, and are flushed to the underlying engine asynchronously.

func NewAsyncEngine 

func NewAsyncEngine(engine Engine, config *AsyncEngineConfig) *AsyncEngine

NewAsyncEngine wraps an engine with write-behind caching. 

func (ae *AsyncEngine) AddSpanLink(sc trace.SpanContext)

AddSpanLink records a request span context so the next flush span can link back to it (TRC-23). Called by TracedEngine on each write operation.

func (*AsyncEngine) AddToPendingEmbeddings 

func (ae *AsyncEngine) AddToPendingEmbeddings(nodeID NodeID)

AddToPendingEmbeddings delegates to the underlying engine, if supported. Call this to re-queue a node for embedding after a failed attempt (e.g. so another worker can retry).

func (*AsyncEngine) AllEdges 

func (ae *AsyncEngine) AllEdges() ([]*Edge, error)

AllEdges returns merged view of cache and engine. It snapshots async cache state quickly under lock, then releases the lock before engine I/O to avoid holding ae.mu across potentially slow scans.

func (*AsyncEngine) AllNodes 

func (ae *AsyncEngine) AllNodes() ([]*Node, error)

AllNodes returns merged view of cache, in-flight nodes, and engine. It snapshots async cache state quickly under lock, then releases the lock before engine I/O to avoid holding ae.mu across potentially slow scans.

func (*AsyncEngine) BatchGetNodes 

func (ae *AsyncEngine) BatchGetNodes(ids []NodeID) (map[NodeID]*Node, error)

BatchGetNodes fetches multiple nodes, checking cache first then engine. Returns a map for O(1) lookup. Missing nodes are not included.

func (*AsyncEngine) BulkCreateEdges 

func (ae *AsyncEngine) BulkCreateEdges(edges []*Edge) error

BulkCreateEdges creates edges in batch (async).

func (*AsyncEngine) BulkCreateNodes 

func (ae *AsyncEngine) BulkCreateNodes(nodes []*Node) error

BulkCreateNodes creates nodes in batch (async).

func (*AsyncEngine) BulkDeleteEdges 

func (ae *AsyncEngine) BulkDeleteEdges(ids []EdgeID) error

BulkDeleteEdges marks multiple edges for deletion (async).

func (*AsyncEngine) BulkDeleteNodes 

func (ae *AsyncEngine) BulkDeleteNodes(ids []NodeID) error

BulkDeleteNodes marks multiple nodes for deletion (async).

func (*AsyncEngine) Close 

func (ae *AsyncEngine) Close() error

Close stops the background flush goroutine and flushes all pending data. Returns an error if the final flush fails or if data remains unflushed.

func (*AsyncEngine) CreateEdge 

func (ae *AsyncEngine) CreateEdge(edge *Edge) error

CreateEdge adds to cache and returns immediately.

func (*AsyncEngine) CreateNode 

func (ae *AsyncEngine) CreateNode(node *Node) (NodeID, error)

CreateNode adds to cache and returns immediately.

func (*AsyncEngine) DeleteByPrefix 

func (ae *AsyncEngine) DeleteByPrefix(prefix string) (nodesDeleted int64, edgesDeleted int64, err error)

DeleteByPrefix delegates to the underlying engine.

func (*AsyncEngine) DeleteEdge 

func (ae *AsyncEngine) DeleteEdge(id EdgeID) error

DeleteEdge marks for deletion and returns immediately. Optimized: if edge was created in this transaction (still in cache), just remove it from cache - no need to delete from underlying engine. CRITICAL: If edge is in-flight (being flushed), we must also mark for deletion because the flush will write it to the underlying engine.

func (*AsyncEngine) DeleteNode 

func (ae *AsyncEngine) DeleteNode(id NodeID) error

DeleteNode marks for deletion and returns immediately. Optimized: if node was created in this transaction (still in cache), just remove it from cache - no need to delete from underlying engine. CRITICAL: If node is in-flight (being flushed), we must also mark for deletion because the flush will write it to the underlying engine.

func (*AsyncEngine) EdgeCount 

func (ae *AsyncEngine) EdgeCount() (int64, error)

func (*AsyncEngine) EdgeCountByPrefix 

func (ae *AsyncEngine) EdgeCountByPrefix(prefix string) (int64, error)

func (*AsyncEngine) FindNodeNeedingEmbedding 

func (ae *AsyncEngine) FindNodeNeedingEmbedding() *Node

FindNodeNeedingEmbedding returns a node that needs embedding. IMPORTANT: This checks the in-memory cache first to ensure we don't re-process nodes that have embeddings pending flush to the underlying engine.

The algorithm: 1. Build set of node IDs that have embeddings in cache (pending flush) 2. First check nodes in our cache that need embedding 3. Then check underlying engine, skipping nodes we have in cache with embeddings

func (*AsyncEngine) Flush 

func (ae *AsyncEngine) Flush() error

Flush writes all pending changes to the underlying engine. Uses batched operations for better performance - all deletes in one transaction.

CRITICAL FIX: Failed items are NOT removed from cache - they will be retried on the next flush. This prevents silent data loss.

Design: Snapshot caches, clear them, UNLOCK, then write to engine. Reads during write see engine data (consistent since cache is empty). This avoids blocking reads during I/O which kills Mac M-series performance.

Thread-safe: Uses flushMu to prevent concurrent flushes which can cause race conditions when cache limit is reached during concurrent writes.

Fast path: if there's nothing pending, avoid taking the exclusive flushMu.Lock() entirely. In seed-heavy workloads the implicit-txn path flushes the cache inline before starting each transaction, so by the time the background ticker fires the cache is almost always empty. Skipping the lock here keeps transaction-path HoldFlush RLockers from queueing behind a no-op write lock acquisition.

func (*AsyncEngine) FlushWithResult 

func (ae *AsyncEngine) FlushWithResult() FlushResult

FlushWithResult writes pending changes and returns detailed results. Use this for programmatic access to flush statistics.

func (*AsyncEngine) ForEachNodeIDByLabel 

func (ae *AsyncEngine) ForEachNodeIDByLabel(label string, visit func(NodeID) bool) error

ForEachNodeIDByLabel streams node IDs for a label, combining cache + engine. Stops early when visit returns false.

func (*AsyncEngine) GetAdjacentEdges added in v1.1.2 

func (ae *AsyncEngine) GetAdjacentEdges(nodeID NodeID) ([]*Edge, []*Edge, error)

GetAdjacentEdges fetches outgoing+incoming edges for nodeID, folding the async cache with a single inner-engine call. Mirrors the merge logic of the per-direction methods; the win is one transaction at the inner engine instead of two per BFS frontier expansion.

func (*AsyncEngine) GetAllNodes 

func (ae *AsyncEngine) GetAllNodes() []*Node

func (*AsyncEngine) GetEdge 

func (ae *AsyncEngine) GetEdge(id EdgeID) (*Edge, error)

GetEdge checks cache first, then underlying engine.

func (*AsyncEngine) GetEdgeBetween 

func (ae *AsyncEngine) GetEdgeBetween(startID, endID NodeID, edgeType string) *Edge

func (*AsyncEngine) GetEdgeCurrentHead 

func (ae *AsyncEngine) GetEdgeCurrentHead(id EdgeID) (MVCCHead, error)

GetEdgeCurrentHead delegates edge head lookup to the wrapped engine when supported.

func (*AsyncEngine) GetEdgeLatestEffective 

func (ae *AsyncEngine) GetEdgeLatestEffective(id EdgeID) (*Edge, error)

GetEdgeLatestEffective returns the merged latest-visible edge across pending, in-flight, and persisted state.

func (*AsyncEngine) GetEdgeLatestVisible 

func (ae *AsyncEngine) GetEdgeLatestVisible(id EdgeID) (*Edge, error)

GetEdgeLatestVisible resolves the latest persisted-or-effective edge.

func (*AsyncEngine) GetEdgeVisibleAt 

func (ae *AsyncEngine) GetEdgeVisibleAt(id EdgeID, version MVCCVersion) (*Edge, error)

GetEdgeVisibleAt delegates snapshot-visible edge reads to the wrapped engine when supported.

func (*AsyncEngine) GetEdgesBetween 

func (ae *AsyncEngine) GetEdgesBetween(startID, endID NodeID) ([]*Edge, error)

func (*AsyncEngine) GetEdgesBetweenVisibleAt 

func (ae *AsyncEngine) GetEdgesBetweenVisibleAt(startID, endID NodeID, version MVCCVersion) ([]*Edge, error)

GetEdgesBetweenVisibleAt delegates snapshot-visible topology queries to the wrapped engine when supported.

func (*AsyncEngine) GetEdgesByType 

func (ae *AsyncEngine) GetEdgesByType(edgeType string) ([]*Edge, error)

GetEdgesByType returns all edges of a specific type, merging cache and engine.

func (*AsyncEngine) GetEdgesByTypeVisibleAt 

func (ae *AsyncEngine) GetEdgesByTypeVisibleAt(edgeType string, version MVCCVersion) ([]*Edge, error)

GetEdgesByTypeVisibleAt delegates snapshot-visible edge-type queries to the wrapped engine when supported.

func (*AsyncEngine) GetEngine 

func (ae *AsyncEngine) GetEngine() Engine

GetEngine returns the underlying storage engine. Used for transaction support which needs direct access.

func (*AsyncEngine) GetFirstNodeByLabel 

func (ae *AsyncEngine) GetFirstNodeByLabel(label string) (*Node, error)

GetNodesByLabel checks cache and merges with engine results. Uses case-insensitive label matching for Neo4j compatibility. Snapshots cache state quickly, then releases lock before engine I/O. GetFirstNodeByLabel returns the first node with the specified label. Optimized for MATCH...LIMIT 1 patterns - uses label index for O(1) lookup.

func (*AsyncEngine) GetInDegree 

func (ae *AsyncEngine) GetInDegree(nodeID NodeID) int

func (*AsyncEngine) GetIncomingEdges 

func (ae *AsyncEngine) GetIncomingEdges(nodeID NodeID) ([]*Edge, error)

func (*AsyncEngine) GetInnerEngine 

func (e *AsyncEngine) GetInnerEngine() Engine

GetInnerEngine returns the wrapped storage engine.

func (*AsyncEngine) GetNode 

func (ae *AsyncEngine) GetNode(id NodeID) (*Node, error)

GetNode checks cache first, then underlying engine.

func (*AsyncEngine) GetNodeCurrentHead 

func (ae *AsyncEngine) GetNodeCurrentHead(id NodeID) (MVCCHead, error)

GetNodeCurrentHead delegates node head lookup to the wrapped engine when supported.

func (*AsyncEngine) GetNodeLatestEffective 

func (ae *AsyncEngine) GetNodeLatestEffective(id NodeID) (*Node, error)

GetNodeLatestEffective returns the merged latest-visible node across pending, in-flight, and persisted state.

func (*AsyncEngine) GetNodeLatestVisible 

func (ae *AsyncEngine) GetNodeLatestVisible(id NodeID) (*Node, error)

GetNodeLatestVisible resolves the latest persisted-or-effective node.

func (*AsyncEngine) GetNodeVisibleAt 

func (ae *AsyncEngine) GetNodeVisibleAt(id NodeID, version MVCCVersion) (*Node, error)

GetNodeVisibleAt delegates snapshot-visible node reads to the wrapped engine when supported.

func (*AsyncEngine) GetNodesByLabel 

func (ae *AsyncEngine) GetNodesByLabel(label string) ([]*Node, error)

func (*AsyncEngine) GetNodesByLabelVisibleAt 

func (ae *AsyncEngine) GetNodesByLabelVisibleAt(label string, version MVCCVersion) ([]*Node, error)

GetNodesByLabelVisibleAt delegates snapshot-visible label queries to the wrapped engine when supported.

func (*AsyncEngine) GetOutDegree 

func (ae *AsyncEngine) GetOutDegree(nodeID NodeID) int

func (*AsyncEngine) GetOutgoingEdges 

func (ae *AsyncEngine) GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)

func (*AsyncEngine) GetSchema 

func (ae *AsyncEngine) GetSchema() *SchemaManager

func (*AsyncEngine) GetSchemaForNamespace 

func (ae *AsyncEngine) GetSchemaForNamespace(namespace string) *SchemaManager

GetSchemaForNamespace implements NamespaceSchemaProvider when the underlying engine supports it.

func (*AsyncEngine) GetUnderlying 

func (ae *AsyncEngine) GetUnderlying() Engine

GetUnderlying returns the underlying storage engine. This is used for transaction support when the underlying engine supports ACID transactions (e.g., BadgerEngine).

func (*AsyncEngine) HasPendingWrites 

func (ae *AsyncEngine) HasPendingWrites() bool

HasPendingWrites returns true if there are unflushed writes. This is a cheap check that can be used to avoid unnecessary flush calls.

func (*AsyncEngine) HoldFlush added in v1.0.42 

func (ae *AsyncEngine) HoldFlush() func()

HoldFlush acquires a shared flush guard and returns a release function.

While held, background or manual Flush calls block on flushMu.Lock(), which is useful when a higher-level transaction needs a stable committed view for its full lifetime. Regular async writes still queue into memory and will flush once the returned release function is called.

func (*AsyncEngine) IsCurrentTemporalNode 

func (ae *AsyncEngine) IsCurrentTemporalNode(node *Node, asOf time.Time) (bool, error)

IsCurrentTemporalNode delegates current-version checks to the wrapped engine when supported.

func (*AsyncEngine) IterateNodes 

func (ae *AsyncEngine) IterateNodes(fn func(*Node) bool) error

IterateNodes iterates through all nodes, checking cache first.

func (*AsyncEngine) LastWriteTime 

func (ae *AsyncEngine) LastWriteTime() time.Time

LastWriteTime returns the last known write time from the underlying engine, if available.

func (*AsyncEngine) LifecycleStatus 

func (ae *AsyncEngine) LifecycleStatus() map[string]interface{}

LifecycleStatus delegates lifecycle status when supported.

func (*AsyncEngine) ListNamespaces 

func (ae *AsyncEngine) ListNamespaces() []string

ListNamespaces returns known namespaces from the wrapped engine, if supported.

func (*AsyncEngine) MarkNodeEmbedded 

func (ae *AsyncEngine) MarkNodeEmbedded(nodeID NodeID)

MarkNodeEmbedded delegates to the underlying engine, if supported. This removes a node from the pending-embeddings secondary index once embedded.

func (*AsyncEngine) NodeCount 

func (ae *AsyncEngine) NodeCount() (int64, error)

func (*AsyncEngine) NodeCountByPrefix 

func (ae *AsyncEngine) NodeCountByPrefix(prefix string) (int64, error)

func (*AsyncEngine) OnEdgeCreated 

func (ae *AsyncEngine) OnEdgeCreated(callback EdgeEventCallback)

OnEdgeCreated sets a callback to be invoked when edges are created.

func (*AsyncEngine) OnEdgeDeleted 

func (ae *AsyncEngine) OnEdgeDeleted(callback EdgeDeleteCallback)

OnEdgeDeleted sets a callback to be invoked when edges are deleted.

func (*AsyncEngine) OnEdgeUpdated 

func (ae *AsyncEngine) OnEdgeUpdated(callback EdgeEventCallback)

OnEdgeUpdated sets a callback to be invoked when edges are updated.

func (*AsyncEngine) OnNodeCreated 

func (ae *AsyncEngine) OnNodeCreated(callback NodeEventCallback)

OnNodeCreated sets a callback to be invoked when nodes are created.

func (*AsyncEngine) OnNodeDeleted 

func (ae *AsyncEngine) OnNodeDeleted(callback NodeDeleteCallback)

OnNodeDeleted sets a callback to be invoked when nodes are deleted.

func (*AsyncEngine) OnNodeUpdated 

func (ae *AsyncEngine) OnNodeUpdated(callback NodeEventCallback)

OnNodeUpdated sets a callback to be invoked when nodes are updated.

func (*AsyncEngine) PauseLifecycle 

func (ae *AsyncEngine) PauseLifecycle()

PauseLifecycle delegates lifecycle pause when supported.

func (*AsyncEngine) PendingEmbeddingsCount 

func (ae *AsyncEngine) PendingEmbeddingsCount() int

PendingEmbeddingsCount delegates to the underlying engine, if supported.

func (*AsyncEngine) PruneMVCCVersions 

func (ae *AsyncEngine) PruneMVCCVersions(ctx context.Context, opts MVCCPruneOptions) (int64, error)

PruneMVCCVersions delegates MVCC pruning to the wrapped engine when supported.

func (*AsyncEngine) PruneTemporalHistory 

func (ae *AsyncEngine) PruneTemporalHistory(ctx context.Context, opts TemporalPruneOptions) (int64, error)

PruneTemporalHistory delegates temporal pruning to the wrapped engine when supported.

func (*AsyncEngine) RebuildMVCCHeads 

func (ae *AsyncEngine) RebuildMVCCHeads(ctx context.Context) error

RebuildMVCCHeads delegates MVCC head rebuild to the wrapped engine when supported.

func (*AsyncEngine) RebuildTemporalIndexes 

func (ae *AsyncEngine) RebuildTemporalIndexes(ctx context.Context) error

RebuildTemporalIndexes delegates temporal index rebuild to the wrapped engine when supported.

func (*AsyncEngine) RecordMaterializedAccess added in v1.1.0 

func (ae *AsyncEngine) RecordMaterializedAccess(entityID string)

RecordMaterializedAccess delegates result-materialization access recording to the underlying engine, if supported.

func (*AsyncEngine) RefreshPendingEmbeddingsIndex 

func (ae *AsyncEngine) RefreshPendingEmbeddingsIndex() int

RefreshPendingEmbeddingsIndex delegates to the underlying engine, if supported. This keeps the pending-embeddings secondary index consistent even when AsyncEngine is the outer-most storage layer.

func (*AsyncEngine) RegisterSnapshotReader 

func (ae *AsyncEngine) RegisterSnapshotReader(info SnapshotReaderInfo) func()

RegisterSnapshotReader delegates snapshot-reader registration when supported.

func (*AsyncEngine) ResumeLifecycle 

func (ae *AsyncEngine) ResumeLifecycle()

ResumeLifecycle delegates lifecycle resume when supported.

func (*AsyncEngine) SetLifecycleSchedule 

func (ae *AsyncEngine) SetLifecycleSchedule(interval time.Duration) error

SetLifecycleSchedule delegates lifecycle cadence updates when supported.

func (*AsyncEngine) Stats 

func (ae *AsyncEngine) Stats() (pendingWrites, totalFlushes int64)

Stats returns async engine statistics.

func (*AsyncEngine) StreamEdges 

func (ae *AsyncEngine) StreamEdges(ctx context.Context, fn func(edge *Edge) error) error

StreamEdges implements StreamingEngine.StreamEdges by delegating to the underlying engine.

func (*AsyncEngine) StreamNodeChunks 

func (ae *AsyncEngine) StreamNodeChunks(ctx context.Context, chunkSize int, fn func(nodes []*Node) error) error

StreamNodeChunks implements StreamingEngine.StreamNodeChunks by using StreamNodes. We always use StreamNodes (not delegate) to properly merge cache + underlying engine.

func (*AsyncEngine) StreamNodes 

func (ae *AsyncEngine) StreamNodes(ctx context.Context, fn func(node *Node) error) error

StreamNodes implements StreamingEngine.StreamNodes by delegating to the underlying engine. It merges cached nodes with the underlying stream for consistency.

func (*AsyncEngine) StreamNodesByPrefix 

func (ae *AsyncEngine) StreamNodesByPrefix(ctx context.Context, prefix string, fn func(node *Node) error) error

StreamNodesByPrefix implements PrefixStreamingEngine by merging pending cache entries with prefix-scoped streaming from the underlying engine.

func (*AsyncEngine) TopLifecycleDebtKeys 

func (ae *AsyncEngine) TopLifecycleDebtKeys(limit int) []MVCCLifecycleDebtKey

TopLifecycleDebtKeys delegates lifecycle debt inspection when supported.

func (*AsyncEngine) TriggerPruneNow 

func (ae *AsyncEngine) TriggerPruneNow(ctx context.Context) error

TriggerPruneNow delegates lifecycle prune-now when supported.

func (*AsyncEngine) UpdateEdge 

func (ae *AsyncEngine) UpdateEdge(edge *Edge) error

UpdateEdge adds to cache and returns immediately.

func (*AsyncEngine) UpdateNode 

func (ae *AsyncEngine) UpdateNode(node *Node) error

UpdateNode adds to cache and returns immediately.

func (*AsyncEngine) UpdateNodeEmbedding 

func (ae *AsyncEngine) UpdateNodeEmbedding(node *Node) error

UpdateNodeEmbedding updates an existing node with its embedding. Unlike UpdateNode, this MUST NOT create a new node; it returns ErrNotFound if the node does not exist (in cache, in-flight, or in the underlying engine).

type AsyncEngineConfig 

type AsyncEngineConfig struct {
	// FlushInterval controls how often pending writes are flushed.
	// Smaller = more consistent, larger = better throughput.
	// Default: 50ms
	FlushInterval time.Duration

	// AdaptiveFlush enables volume-based flush timing.
	// When enabled, the flush loop ticks at MinFlushInterval and only
	// flushes when the adaptive interval has elapsed.
	AdaptiveFlush bool

	// MinFlushInterval is the shortest interval between flushes when adaptive flush is enabled.
	// Default: 10ms
	MinFlushInterval time.Duration

	// MaxFlushInterval is the longest interval between flushes when adaptive flush is enabled.
	// Default: 200ms
	MaxFlushInterval time.Duration

	// TargetFlushSize is the pending write count at which we reach MaxFlushInterval.
	// Smaller batches flush more frequently; larger batches flush less frequently.
	// Default: 1000 pending writes
	TargetFlushSize int

	// MaxNodeCacheSize is the maximum number of nodes to buffer before forcing a flush.
	// When this limit is reached, CreateNode will block and flush synchronously.
	// This prevents unbounded memory growth during bulk inserts.
	// Set to 0 for unlimited (not recommended for bulk operations).
	// Default: 50000 (50K nodes, ~35MB assuming 700 bytes/node)
	MaxNodeCacheSize int

	// MaxEdgeCacheSize is the maximum number of edges to buffer before forcing a flush.
	// When this limit is reached, CreateEdge will block and flush synchronously.
	// This prevents unbounded memory growth during bulk inserts.
	// Set to 0 for unlimited (not recommended for bulk operations).
	// Default: 100000 (100K edges, ~50MB assuming 500 bytes/edge)
	MaxEdgeCacheSize int

	// Logger is the structured *slog.Logger threaded into the AsyncEngine.
	// D-01 logger DI: optional; nil falls back to a discard handler at ctor
	// entry per D-01a so existing callers (tests, scripts) compile unchanged.
	// D-06: the flush goroutine derives a single-allocation child logger from
	// this field at goroutine start.
	Logger *slog.Logger
}

AsyncEngineConfig configures the async engine behavior.

func DefaultAsyncEngineConfig 

func DefaultAsyncEngineConfig() *AsyncEngineConfig

DefaultAsyncEngineConfig returns sensible defaults.

type BadgerEngine 

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

BadgerEngine provides persistent storage using BadgerDB.

Features:

  • ACID transactions for all operations
  • Persistent storage to disk
  • Secondary indexes for efficient queries
  • Thread-safe concurrent access
  • Automatic crash recovery

Key Structure:

  • Nodes: 0x01 + nodeID -> JSON(Node)
  • Edges: 0x02 + edgeID -> JSON(Edge)
  • Label Index: 0x03 + label + 0x00 + nodeID -> empty
  • Outgoing Index: 0x04 + nodeID + 0x00 + edgeID -> empty
  • Incoming Index: 0x05 + nodeID + 0x00 + edgeID -> empty
  • Edge-Between Set: 0x18 + startID + 0x00 + endID + 0x00 + type + 0x00 + edgeID -> empty
  • Edge-Between Head: 0x19 + startID + 0x00 + endID + 0x00 + type -> edgeID

Example: 

engine, err := storage.NewBadgerEngine("/path/to/data")
if err != nil {
	log.Fatal(err)
}
defer engine.Close()

node := &storage.Node{
	ID:     "user-123",
	Labels: []string{"User"},
	Properties: map[string]any{"name": "Alice"},
}
engine.CreateNode(node)

func NewBadgerEngine 

func NewBadgerEngine(dataDir string) (*BadgerEngine, error)

NewBadgerEngine creates a new persistent storage engine with default settings.

This is the simplest way to create a storage engine. The engine uses BadgerDB for persistent disk storage with ACID transaction guarantees. All data is stored in the specified directory and persists across restarts.

Parameters:

  • dataDir: Directory path for storing data files. Created if it doesn't exist.

Returns:

  • *BadgerEngine on success
  • error if database cannot be opened (e.g., permissions, disk space)

Example 1 - Basic Usage: 

engine, err := storage.NewBadgerEngine("./data/nornicdb")
if err != nil {
	log.Fatal(err)
}
defer engine.Close()

// Engine is ready - create nodes
node := &storage.Node{
	ID:     "user-1",
	Labels: []string{"User"},
	Properties: map[string]any{"name": "Alice"},
}
engine.CreateNode(node)

Example 2 - Production Application: 

// Use absolute path for production
dataDir := filepath.Join(os.Getenv("APP_HOME"), "data", "nornicdb")
engine, err := storage.NewBadgerEngine(dataDir)
if err != nil {
	return fmt.Errorf("failed to open database: %w", err)
}
defer engine.Close()

Example 3 - Multiple Databases: 

// Main application database
mainDB, _ := storage.NewBadgerEngine("./data/main")
defer mainDB.Close()

// Test database
testDB, _ := storage.NewBadgerEngine("./data/test")
defer testDB.Close()

// Cache database
cacheDB, _ := storage.NewBadgerEngine("./data/cache")
defer cacheDB.Close()

ELI12:

Think of NewBadgerEngine like setting up a filing cabinet in your room. You tell it "put the cabinet here" (the dataDir), and it creates folders and organizes everything. Even if you turn off your computer, the cabinet stays there with all your files inside. Next time you start up, all your data is still there!

Disk Usage:

  • Approximately 2-3x the size of your actual data
  • Includes write-ahead log and compaction overhead

Thread Safety: 

Safe for concurrent use from multiple goroutines.

func NewBadgerEngineInMemory 

func NewBadgerEngineInMemory() (*BadgerEngine, error)

NewBadgerEngineInMemory creates an in-memory BadgerDB for testing.

Data is not persisted and is lost when the engine is closed. Useful for unit tests that need persistent storage semantics without actual disk I/O.

Example: 

engine, err := storage.NewBadgerEngineInMemory()
if err != nil {
	t.Fatal(err)
}
defer engine.Close()

// Use engine for testing...

func NewBadgerEngineWithOptions 

func NewBadgerEngineWithOptions(opts BadgerOptions) (*BadgerEngine, error)

NewBadgerEngineWithOptions creates a BadgerEngine with custom configuration.

Use this function when you need fine-grained control over the storage engine behavior, such as enabling in-memory mode for testing, forcing synchronous writes for maximum durability, or reducing memory usage.

Parameters:

  • opts: BadgerOptions struct with configuration settings

Returns:

  • *BadgerEngine on success
  • error if database cannot be opened

Example 1 - In-Memory Database for Testing: 

engine, err := storage.NewBadgerEngineWithOptions(storage.BadgerOptions{
	DataDir:  "./test", // Still needs a path but won't be used
	InMemory: true,     // All data in RAM, lost on shutdown
})
defer engine.Close()

// Perfect for unit tests - fast and clean
testCreateNodes(engine)

Example 2 - Maximum Durability for Financial Data: 

engine, err := storage.NewBadgerEngineWithOptions(storage.BadgerOptions{
	DataDir:    "./data/transactions",
	SyncWrites: true, // Force fsync after each write (slower but safer)
})
// Guaranteed data persistence even if power fails

Example 3 - Low Memory Mode for Embedded Devices: 

engine, err := storage.NewBadgerEngineWithOptions(storage.BadgerOptions{
	DataDir:   "./data/nornicdb",
	LowMemory: true, // Reduces RAM usage by 50-70%
})
// Uses ~50MB instead of ~150MB for typical workloads

Example 4 - Custom Logger Integration: 

logger := zerolog.New(os.Stdout).With().Timestamp().Logger()
engine, err := storage.NewBadgerEngineWithOptions(storage.BadgerOptions{
	DataDir: "./data/nornicdb",
	Logger:  &BadgerLogger{zlog: logger}, // Custom logging
})

ELI12:

NewBadgerEngine is like getting a basic backpack for school. NewBadgerEngineWithOptions is like customizing your backpack - you can:

  • Make it waterproof (SyncWrites = true)
  • Make it lighter but less storage (LowMemory = true)
  • Use it as a temporary bag (InMemory = true)
  • Add custom labels (Logger)

Configuration Trade-offs:

  • SyncWrites=true: Slower writes (2-5x) but maximum safety
  • LowMemory=true: Less RAM but slightly slower
  • InMemory=true: Fastest but data lost on shutdown

Thread Safety: 

Safe for concurrent use from multiple goroutines.

func (*BadgerEngine) ActiveReaders added in v1.1.0 

func (b *BadgerEngine) ActiveReaders() int64

ActiveReaders returns the count of currently-open MVCC reader snapshots. Always non-nil; returns 0 when the engine has no active readers.

Plan 04-04-01 RISK-2 fix — used by the observability MVCC bag's active_readers GaugeFunc callback.

func (*BadgerEngine) AddToPendingEmbeddings 

func (b *BadgerEngine) AddToPendingEmbeddings(nodeID NodeID)

AddToPendingEmbeddings adds a node to the pending embeddings index. Call this when creating a node that needs embedding.

func (*BadgerEngine) AllEdges 

func (b *BadgerEngine) AllEdges() ([]*Edge, error)

AllEdges returns all edges (implements Engine interface).

func (*BadgerEngine) AllNodes 

func (b *BadgerEngine) AllNodes() ([]*Node, error)

AllNodes returns all nodes (implements Engine interface).

func (*BadgerEngine) AppendEdgeTombstone 

func (b *BadgerEngine) AppendEdgeTombstone(id EdgeID, version MVCCVersion) error

func (*BadgerEngine) AppendEdgeVersion 

func (b *BadgerEngine) AppendEdgeVersion(edge *Edge, version MVCCVersion) error

func (*BadgerEngine) AppendNodeTombstone 

func (b *BadgerEngine) AppendNodeTombstone(id NodeID, version MVCCVersion) error

func (*BadgerEngine) AppendNodeVersion 

func (b *BadgerEngine) AppendNodeVersion(node *Node, version MVCCVersion) error

func (*BadgerEngine) AttachMetrics added in v1.1.0 

func (b *BadgerEngine) AttachMetrics(storage *observability.StorageMetrics, mvcc *observability.MVCCMetrics)

AttachMetrics injects the observability bags into the engine and pre-binds the per-op observers. Idempotent: subsequent calls overwrite the previously-bound observers, which is safe because BoundLatency Observer is value-typed and concurrent Observe calls on either the new or old observer are race-clean (client_golang HistogramVec promise).

Plan 04-04-07 calls AttachMetrics from cmd/nornicdb startup AFTER constructing the bags but BEFORE starting the supervisor.

Ordering discipline (D-02c): metricsAttached flag is set LAST to prevent race where metricsAttached=true but observer fields are still being initialized. This ensures that once metricsAttached is true, all observer fields are guaranteed to be bound.

func (*BadgerEngine) Backup 

func (b *BadgerEngine) Backup(path string) error

Backup creates a backup of the database to the specified file path. Uses BadgerDB's streaming backup which creates a consistent snapshot. The backup file is a self-contained, portable copy of the database.

func (*BadgerEngine) BatchGetNodes 

func (b *BadgerEngine) BatchGetNodes(ids []NodeID) (map[NodeID]*Node, error)

BatchGetNodes fetches multiple nodes in a single transaction. Returns a map for O(1) lookup by ID. Missing nodes are not included in the result. This is optimized for traversal operations that need to fetch many nodes.

func (*BadgerEngine) BatchGetNodesLatestVisible 

func (b *BadgerEngine) BatchGetNodesLatestVisible(ids []NodeID) (map[NodeID]*Node, error)

func (*BadgerEngine) BeginQueryRevealScope added in v1.1.0 

func (b *BadgerEngine) BeginQueryRevealScope(reveal bool) func()

BeginQueryRevealScope guards a query's reveal mode against concurrent queries on the same engine. Reveal queries take an exclusive scope while normal queries take a shared scope so they cannot observe revealAll=true.

func (*BadgerEngine) BeginTransaction 

func (b *BadgerEngine) BeginTransaction() (*BadgerTransaction, error)

BeginTransaction starts a new Badger transaction with ACID guarantees.

At begin time the namespace is unknown; readTS is populated with a wall-clock-only sample (no sequence component) and rebound the moment the transaction's namespace is pinned via the first prefixed write or SetNamespace. Pre-pin reads see a version that does not constrain against any namespace's commit sequence, which is the correct behavior: a transaction that has not yet identified its database has not made any per-database isolation claims.

func (*BadgerEngine) BulkCreateEdges 

func (b *BadgerEngine) BulkCreateEdges(edges []*Edge) error

BulkCreateEdges creates multiple edges in a single transaction.

func (*BadgerEngine) BulkCreateNodes 

func (b *BadgerEngine) BulkCreateNodes(nodes []*Node) error

BulkCreateNodes creates multiple nodes in a single transaction.

func (*BadgerEngine) BulkDeleteEdges 

func (b *BadgerEngine) BulkDeleteEdges(ids []EdgeID) error

BulkDeleteEdges removes multiple edges in a single transaction. This is much faster than calling DeleteEdge repeatedly.

func (*BadgerEngine) BulkDeleteNodes 

func (b *BadgerEngine) BulkDeleteNodes(ids []NodeID) error

BulkDeleteNodes removes multiple nodes in a single transaction. This is much faster than calling DeleteNode repeatedly. IMPORTANT: This also deletes all edges connected to the deleted nodes and updates edge counts.

func (*BadgerEngine) ClearAllEmbeddings 

func (b *BadgerEngine) ClearAllEmbeddings() (int, error)

ClearAllEmbeddings removes embeddings from all nodes, allowing them to be regenerated. Returns the number of nodes that had their embeddings cleared.

func (*BadgerEngine) ClearAllEmbeddingsForPrefix 

func (b *BadgerEngine) ClearAllEmbeddingsForPrefix(idPrefix string) (int, error)

ClearAllEmbeddingsForPrefix removes embeddings from nodes whose IDs start with the given prefix. This is used to clear embeddings for a single logical database namespace (e.g., "nornic:").

If idPrefix is empty, clears embeddings for all nodes.

func (*BadgerEngine) Close 

func (b *BadgerEngine) Close() error

Close closes the BadgerDB database.

func (*BadgerEngine) CreateEdge 

func (b *BadgerEngine) CreateEdge(edge *Edge) error

CreateEdge creates a new edge between two nodes.

func (*BadgerEngine) CreateNode 

func (b *BadgerEngine) CreateNode(node *Node) (NodeID, error)

CreateNode creates a new node in persistent storage. REQUIRES: node.ID must be prefixed with namespace (e.g., "nornic:node-123"). This enforces that all nodes are namespaced at the storage layer.

func (*BadgerEngine) DB added in v1.1.0 

func (b *BadgerEngine) DB() *badger.DB

DB returns the underlying *badger.DB handle. Used by Plan 04-04-04 bytes_metrics_sweeper to call EstimateSize(prefix). Read-only handle — callers must not invoke lifecycle methods (Close/DropAll) on the returned pointer; the engine owns lifecycle.

func (*BadgerEngine) DataDirFreeSpace 

func (b *BadgerEngine) DataDirFreeSpace() (int64, error)

DataDirFreeSpace returns free bytes available to the current process for the underlying Badger data directory.

func (*BadgerEngine) DeleteAccessMeta added in v1.1.0 

func (b *BadgerEngine) DeleteAccessMeta(entityID string) error

func (*BadgerEngine) DeleteByPrefix 

func (b *BadgerEngine) DeleteByPrefix(prefix string) (nodesDeleted int64, edgesDeleted int64, err error)

DeleteByPrefix deletes all nodes and edges with IDs starting with the given prefix. Used for DROP DATABASE operations to delete all data in a namespace.

This uses Badger's native prefix drop for db-scoped keyspaces to avoid per-node decoding and DeleteNode/DeleteEdge loops. Secondary indexes that don't begin with the node/edge ID (label and edge-type indexes) are cleaned up by scanning those index keyspaces and deleting entries whose suffix IDs match the prefix.

func (*BadgerEngine) DeleteDeindexWorkItem added in v1.1.0 

func (b *BadgerEngine) DeleteDeindexWorkItem(workItemID string) error

func (*BadgerEngine) DeleteEdge 

func (b *BadgerEngine) DeleteEdge(id EdgeID) error

DeleteEdge removes an edge.

func (*BadgerEngine) DeleteIndexEntryCatalog added in v1.1.0 

func (b *BadgerEngine) DeleteIndexEntryCatalog(entityID string) error

func (*BadgerEngine) DeleteIndexTombstones added in v1.1.0 

func (b *BadgerEngine) DeleteIndexTombstones(keys [][]byte) error

DeleteIndexTombstones removes tombstones for the given original index keys. Used when an entity recovers visibility (score rises above threshold) or when reveal() restores an entity.

func (*BadgerEngine) DeleteMVCCVersion 

func (b *BadgerEngine) DeleteMVCCVersion(ctx context.Context, logicalKey []byte, version MVCCVersion) error

DeleteMVCCVersion deletes a single MVCC version for a logical key.

func (*BadgerEngine) DeleteNode 

func (b *BadgerEngine) DeleteNode(id NodeID) error

DeleteNode removes a node and all its edges.

func (*BadgerEngine) EdgeCount 

func (b *BadgerEngine) EdgeCount() (int64, error)

EdgeCount returns the total number of valid, decodable edges. This is consistent with AllEdges() - only counts edges that can be successfully decoded.

func (*BadgerEngine) EdgeCountByPrefix 

func (b *BadgerEngine) EdgeCountByPrefix(prefix string) (int64, error)

EdgeCountByPrefix counts edges whose EdgeID begins with the provided prefix. The prefix refers to the EdgeID string prefix (e.g., database namespace "nornic:").

func (*BadgerEngine) EnqueueDeindexIfSuppressed added in v1.1.0 

func (b *BadgerEngine) EnqueueDeindexIfSuppressed(entityID string, isEdge bool) (bool, error)

EnqueueDeindexIfSuppressed evaluates the entity's current decay score. If below the visibility threshold, it marks the entity as suppressed and creates a pending deindex work item. If above threshold and currently suppressed, it clears the suppression and deletes any tombstones. The returned bool is true only when the entity transitioned into the suppressed state during this call.

func (*BadgerEngine) FilterEdgePropertyByDecay added in v1.1.0 

func (b *BadgerEngine) FilterEdgePropertyByDecay(edgeID EdgeID, edgeType, propKey string, createdAtNanos, versionAtNanos, nowNanos int64) bool

FilterEdgePropertyByDecay returns true if the edge property should be hidden from results.

func (*BadgerEngine) FilterPropertyByDecay added in v1.1.0 

func (b *BadgerEngine) FilterPropertyByDecay(nodeID NodeID, labels []string, propKey string, createdAtNanos, versionAtNanos, nowNanos int64) bool

FilterPropertyByDecay returns true if the property should be hidden from results.

func (*BadgerEngine) FindNodeNeedingEmbedding 

func (b *BadgerEngine) FindNodeNeedingEmbedding() *Node

FindNodeNeedingEmbedding returns a node that needs embedding. Uses Badger's secondary index (prefixPendingEmbed) for O(1) lookup.

This is highly optimized: - O(1) to find next node (just seek to prefix and get first key) - No in-memory index needed - Persistent across restarts - Atomic with node operations

CRITICAL: This method aggressively cleans up stale entries to prevent processing non-existent nodes. It will skip up to 100 stale entries before giving up to prevent infinite loops.

func (*BadgerEngine) ForEachNodeIDByLabel 

func (b *BadgerEngine) ForEachNodeIDByLabel(label string, visit func(NodeID) bool) error

ForEachNodeIDByLabel streams node IDs for a label without decoding nodes. Stops early when visit returns false.

func (*BadgerEngine) GetAccessMeta added in v1.1.0 

func (b *BadgerEngine) GetAccessMeta(entityID string) (*knowledgepolicy.AccessMetaEntry, error)

func (*BadgerEngine) GetAdjacentEdges added in v1.1.2 

func (b *BadgerEngine) GetAdjacentEdges(nodeID NodeID) ([]*Edge, []*Edge, error)

GetAdjacentEdges fetches both outgoing and incoming edges for nodeID. On a hot path the per-node adjacency cache short-circuits the Badger iterator entirely, falling back to a single view transaction when either direction misses.

func (*BadgerEngine) GetAllNodes 

func (b *BadgerEngine) GetAllNodes() []*Node

GetAllNodes returns all nodes in the storage.

func (*BadgerEngine) GetDeindexWorkItem added in v1.1.0 

func (b *BadgerEngine) GetDeindexWorkItem(workItemID string) (*DeindexWorkItem, error)

func (*BadgerEngine) GetEdge 

func (b *BadgerEngine) GetEdge(id EdgeID) (*Edge, error)

GetEdge retrieves an edge by ID.

func (*BadgerEngine) GetEdgeBetween 

func (b *BadgerEngine) GetEdgeBetween(source, target NodeID, edgeType string) *Edge

GetEdgeBetween returns an edge between two nodes with the given type.

func (*BadgerEngine) GetEdgeCurrentHead 

func (b *BadgerEngine) GetEdgeCurrentHead(id EdgeID) (MVCCHead, error)

func (*BadgerEngine) GetEdgeLatestVisible 

func (b *BadgerEngine) GetEdgeLatestVisible(id EdgeID) (*Edge, error)

func (*BadgerEngine) GetEdgeVisibleAt 

func (b *BadgerEngine) GetEdgeVisibleAt(id EdgeID, version MVCCVersion) (*Edge, error)

func (*BadgerEngine) GetEdgesBetween 

func (b *BadgerEngine) GetEdgesBetween(startID, endID NodeID) ([]*Edge, error)

GetEdgesBetween returns all edges between two nodes.

func (*BadgerEngine) GetEdgesBetweenVisibleAt 

func (b *BadgerEngine) GetEdgesBetweenVisibleAt(startID, endID NodeID, version MVCCVersion) ([]*Edge, error)

func (*BadgerEngine) GetEdgesByType 

func (b *BadgerEngine) GetEdgesByType(edgeType string) ([]*Edge, error)

GetEdgesByType returns all edges of a specific type using the edge type index. This is MUCH faster than AllEdges() for queries like mutual follows. Edge types are matched case-insensitively (Neo4j compatible). Results are cached per type to speed up repeated queries.

func (*BadgerEngine) GetEdgesByTypeVisibleAt 

func (b *BadgerEngine) GetEdgesByTypeVisibleAt(edgeType string, version MVCCVersion) ([]*Edge, error)

func (*BadgerEngine) GetEntityMeta added in v1.1.0 

func (b *BadgerEngine) GetEntityMeta(entityID string) (knowledgepolicy.EntityMeta, error)

GetEntityMeta returns the metadata needed by the access flusher to resolve ON ACCESS policies and property visibility for either nodes or edges.

func (*BadgerEngine) GetFirstNodeByLabel 

func (b *BadgerEngine) GetFirstNodeByLabel(label string) (*Node, error)

GetFirstNodeByLabel returns the first node with the specified label. This is optimized for MATCH...LIMIT 1 patterns - stops after first match.

func (*BadgerEngine) GetInDegree 

func (b *BadgerEngine) GetInDegree(nodeID NodeID) int

GetInDegree returns the number of incoming edges to a node.

func (*BadgerEngine) GetIncomingEdges 

func (b *BadgerEngine) GetIncomingEdges(nodeID NodeID) ([]*Edge, error)

GetIncomingEdges returns all edges where the given node is the target.

func (*BadgerEngine) GetIndexEntryCatalog added in v1.1.0 

func (b *BadgerEngine) GetIndexEntryCatalog(entityID string) (*IndexEntryCatalog, error)

func (*BadgerEngine) GetNode 

func (b *BadgerEngine) GetNode(id NodeID) (*Node, error)

GetNode retrieves a node by ID.

func (*BadgerEngine) GetNodeCurrentHead 

func (b *BadgerEngine) GetNodeCurrentHead(id NodeID) (MVCCHead, error)

func (*BadgerEngine) GetNodeLatestVisible 

func (b *BadgerEngine) GetNodeLatestVisible(id NodeID) (*Node, error)

func (*BadgerEngine) GetNodeVisibleAt 

func (b *BadgerEngine) GetNodeVisibleAt(id NodeID, version MVCCVersion) (*Node, error)

func (*BadgerEngine) GetNodesByLabel 

func (b *BadgerEngine) GetNodesByLabel(label string) ([]*Node, error)

GetNodesByLabel returns all nodes with the specified label.

func (*BadgerEngine) GetNodesByLabelVisibleAt 

func (b *BadgerEngine) GetNodesByLabelVisibleAt(label string, version MVCCVersion) ([]*Node, error)

func (*BadgerEngine) GetOutDegree 

func (b *BadgerEngine) GetOutDegree(nodeID NodeID) int

GetOutDegree returns the number of outgoing edges from a node.

func (*BadgerEngine) GetOutgoingEdges 

func (b *BadgerEngine) GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)

GetOutgoingEdges returns all edges where the given node is the source.

func (*BadgerEngine) GetSchema 

func (b *BadgerEngine) GetSchema() *SchemaManager

GetSchema returns the schema manager.

func (*BadgerEngine) GetSchemaForNamespace 

func (b *BadgerEngine) GetSchemaForNamespace(namespace string) *SchemaManager

GetSchemaForNamespace returns the schema for a specific database namespace. If no schema exists yet, an empty schema is created (and will be persisted once mutated).

func (*BadgerEngine) GetTemporalNodeAsOfInNamespace 

func (b *BadgerEngine) GetTemporalNodeAsOfInNamespace(namespace, label, keyProp string, keyValue interface{}, validFromProp, validToProp string, asOf time.Time) (*Node, error)

func (*BadgerEngine) HasLabelBatch 

func (b *BadgerEngine) HasLabelBatch(ids []NodeID, label string) (map[NodeID]bool, error)

HasLabelBatch checks label membership for a batch of node IDs using the label index. This avoids decoding node records and is significantly faster for large batches.

func (*BadgerEngine) IDDictCounters added in v1.1.0 

func (b *BadgerEngine) IDDictCounters() (nodes, edges uint64)

IDDictCounters exposes the dict's counter state for observability probes. Package-level method on BadgerEngine routes through here.

func (*BadgerEngine) IDDictFreelistPending added in v1.1.0 

func (b *BadgerEngine) IDDictFreelistPending() (nodes, edges int64)

IDDictFreelistPending exposes the freelist pending counters for observability probes.

func (*BadgerEngine) InvalidateEdgeTypeCache 

func (b *BadgerEngine) InvalidateEdgeTypeCache()

InvalidateEdgeTypeCache clears the entire edge type cache. Called after bulk edge mutations to ensure cache consistency.

func (*BadgerEngine) InvalidateEdgeTypeCacheForType 

func (b *BadgerEngine) InvalidateEdgeTypeCacheForType(edgeType string)

InvalidateEdgeTypeCacheForType removes only the specified edge type from cache. Much faster than full invalidation for single-edge operations.

func (*BadgerEngine) InvalidatePendingEmbeddingsIndex 

func (b *BadgerEngine) InvalidatePendingEmbeddingsIndex()

InvalidatePendingEmbeddingsIndex is a no-op for Badger index. The index is persistent and doesn't need invalidation.

func (*BadgerEngine) IsCurrentTemporalNode 

func (b *BadgerEngine) IsCurrentTemporalNode(node *Node, asOf time.Time) (bool, error)

IsCurrentTemporalNode reports whether the given temporal node is the live/current version.

func (*BadgerEngine) IsCurrentTemporalNodeInNamespace 

func (b *BadgerEngine) IsCurrentTemporalNodeInNamespace(namespace string, node *Node, asOf time.Time) (bool, error)

IsCurrentTemporalNodeInNamespace reports whether the given temporal node is the live/current version.

func (*BadgerEngine) IsDecayEnabled added in v1.1.0 

func (b *BadgerEngine) IsDecayEnabled() bool

IsDecayEnabled reports whether knowledge-layer decay scoring is active.

func (*BadgerEngine) IsEmbeddingsEnabled added in v1.1.0 

func (b *BadgerEngine) IsEmbeddingsEnabled() bool

IsEmbeddingsEnabled reports whether the engine should maintain the pending- embed index.

func (*BadgerEngine) IsInMemory 

func (b *BadgerEngine) IsInMemory() bool

IsInMemory returns true if the engine is running in memory-only mode. In-memory mode is used for testing - there's no disk to fsync to.

func (*BadgerEngine) IterateLatestVisibleEdges 

func (b *BadgerEngine) IterateLatestVisibleEdges(yield func(*Edge) error) error

func (*BadgerEngine) IterateLatestVisibleNodes 

func (b *BadgerEngine) IterateLatestVisibleNodes(yield func(*Node) error) error

func (*BadgerEngine) IterateMVCCHeads 

func (b *BadgerEngine) IterateMVCCHeads(ctx context.Context, yield func(logicalKey []byte, head MVCCHead) error) error

IterateMVCCHeads iterates all persisted MVCC heads.

func (*BadgerEngine) IterateMVCCVersions 

func (b *BadgerEngine) IterateMVCCVersions(ctx context.Context, logicalKey []byte, yield func(version MVCCVersion, tombstoned bool, sizeBytes int64) error) error

IterateMVCCVersions iterates all versions for a logical key.

func (*BadgerEngine) IterateNodes 

func (b *BadgerEngine) IterateNodes(fn func(*Node) bool) error

IterateNodes iterates through all nodes one at a time without loading all into memory. The callback returns true to continue, false to stop.

func (*BadgerEngine) LifecycleStatus 

func (b *BadgerEngine) LifecycleStatus() map[string]interface{}

LifecycleStatus reports lifecycle status when enabled.

func (*BadgerEngine) ListNamespaces 

func (b *BadgerEngine) ListNamespaces() []string

ListNamespaces returns the set of database namespaces currently present in this Badger engine, derived from the cached per-namespace node/edge counters.

Namespaces are returned without the trailing ':' (e.g., "nornic", "db2").

func (*BadgerEngine) Logger added in v1.1.0 

func (b *BadgerEngine) Logger() *slog.Logger

Logger returns the BadgerEngine's structured logger (D-01 accessor). Used by NamespacedEngine and other wrappers to derive child loggers without an additional ctor parameter. Returns a discard logger if no logger was supplied at construction.

func (*BadgerEngine) MarkNodeEmbedded 

func (b *BadgerEngine) MarkNodeEmbedded(nodeID NodeID)

MarkNodeEmbedded removes a node from the pending embeddings index. Call this after successfully embedding a node.

func (*BadgerEngine) NodeCount 

func (b *BadgerEngine) NodeCount() (int64, error)

func (*BadgerEngine) NodeCountByPrefix 

func (b *BadgerEngine) NodeCountByPrefix(prefix string) (int64, error)

NodeCountByPrefix counts nodes whose NodeID begins with the provided prefix. The prefix refers to the NodeID string prefix (e.g., database namespace "nornic:").

This is an optional fast-path used by NamespacedEngine to provide accurate per-database counts without decoding values.

func (*BadgerEngine) OldestReaderAgeSeconds added in v1.1.0 

func (b *BadgerEngine) OldestReaderAgeSeconds() float64

OldestReaderAgeSeconds returns the wall-clock age (in seconds) of the oldest still-open MVCC reader snapshot. Returns 0 when no readers are active or no controller is wired (atomic-counter fallback does not track per-reader StartTime).

Plan 04-04-01 RISK-2 fix — used by the observability MVCC bag's oldest_reader_age_seconds GaugeFunc callback.

func (*BadgerEngine) OnEdgeCreated 

func (b *BadgerEngine) OnEdgeCreated(callback EdgeEventCallback)

OnEdgeCreated sets a callback to be invoked when edges are created. Implements StorageEventNotifier interface.

func (*BadgerEngine) OnEdgeDeleted 

func (b *BadgerEngine) OnEdgeDeleted(callback EdgeDeleteCallback)

OnEdgeDeleted sets a callback to be invoked when edges are deleted. Implements StorageEventNotifier interface.

func (*BadgerEngine) OnEdgeUpdated 

func (b *BadgerEngine) OnEdgeUpdated(callback EdgeEventCallback)

OnEdgeUpdated sets a callback to be invoked when edges are updated. Implements StorageEventNotifier interface.

func (*BadgerEngine) OnNodeCreated 

func (b *BadgerEngine) OnNodeCreated(callback NodeEventCallback)

OnNodeCreated sets a callback to be invoked when nodes are created. Implements StorageEventNotifier interface.

func (*BadgerEngine) OnNodeDeleted 

func (b *BadgerEngine) OnNodeDeleted(callback NodeDeleteCallback)

OnNodeDeleted sets a callback to be invoked when nodes are deleted. Implements StorageEventNotifier interface.

func (*BadgerEngine) OnNodeUpdated 

func (b *BadgerEngine) OnNodeUpdated(callback NodeEventCallback)

OnNodeUpdated sets a callback to be invoked when nodes are updated. Implements StorageEventNotifier interface.

func (*BadgerEngine) PauseLifecycle 

func (b *BadgerEngine) PauseLifecycle()

PauseLifecycle pauses lifecycle work.

func (*BadgerEngine) PendingEmbeddingsCount 

func (b *BadgerEngine) PendingEmbeddingsCount() int

PendingEmbeddingsCount returns the number of nodes waiting for embedding. Note: This requires a scan of the pending index, so use sparingly.

func (*BadgerEngine) PinnedBytes added in v1.1.0 

func (b *BadgerEngine) PinnedBytes() int64

PinnedBytes returns the cumulative byte count pinned by all active MVCC readers. Returns 0 when no controller is wired (fallback path tracks counts only, not bytes). Always non-nil.

Plan 04-04-01 RISK-2 fix — used by the observability MVCC bag's pinned_bytes GaugeFunc callback.

func (*BadgerEngine) PropKeyDictCounters added in v1.1.0 

func (b *BadgerEngine) PropKeyDictCounters() map[string]uint64

PropKeyDictCounters exposes per-namespace counter usage for observability probes.

func (*BadgerEngine) PruneMVCCVersions 

func (b *BadgerEngine) PruneMVCCVersions(ctx context.Context, opts MVCCPruneOptions) (int64, error)

func (*BadgerEngine) PruneTemporalHistory 

func (b *BadgerEngine) PruneTemporalHistory(ctx context.Context, opts TemporalPruneOptions) (int64, error)

PruneTemporalHistory removes older closed temporal versions according to opts.

func (*BadgerEngine) PutAccessMeta added in v1.1.0 

func (b *BadgerEngine) PutAccessMeta(entityID string, entry *knowledgepolicy.AccessMetaEntry) error

func (*BadgerEngine) PutDeindexWorkItem added in v1.1.0 

func (b *BadgerEngine) PutDeindexWorkItem(item *DeindexWorkItem) error

func (*BadgerEngine) PutIndexEntryCatalog added in v1.1.0 

func (b *BadgerEngine) PutIndexEntryCatalog(entityID string, cat *IndexEntryCatalog) error

func (*BadgerEngine) ReadMVCCHead 

func (b *BadgerEngine) ReadMVCCHead(ctx context.Context, logicalKey []byte) (MVCCHead, error)

ReadMVCCHead loads the current MVCC head for a logical key.

func (*BadgerEngine) RebuildMVCCHeads 

func (b *BadgerEngine) RebuildMVCCHeads(ctx context.Context) error

func (*BadgerEngine) RebuildTemporalIndexes 

func (b *BadgerEngine) RebuildTemporalIndexes(ctx context.Context) error

RebuildTemporalIndexes recreates the temporal history and current-pointer indexes from stored nodes.

func (*BadgerEngine) ReconcileDecaySuppression added in v1.1.0 

func (b *BadgerEngine) ReconcileDecaySuppression(namespace string) error

ReconcileDecaySuppression re-evaluates current suppression state for all entities in a namespace after knowledge-policy changes.

func (*BadgerEngine) ReconcileDecaySuppressionWithChanges added in v1.1.0 

func (b *BadgerEngine) ReconcileDecaySuppressionWithChanges(namespace string) ([]SuppressionStateChange, error)

ReconcileDecaySuppressionWithChanges re-evaluates suppression state and returns the entities whose visibility-suppressed state changed.

func (*BadgerEngine) RecordMaterializedAccess added in v1.1.0 

func (b *BadgerEngine) RecordMaterializedAccess(entityID string)

RecordMaterializedAccess records an access only after the query executor has fully materialized the entity into a result row.

func (*BadgerEngine) RefreshPendingEmbeddingsIndex 

func (b *BadgerEngine) RefreshPendingEmbeddingsIndex() int

RefreshPendingEmbeddingsIndex rebuilds the pending embeddings index. This scans all nodes and adds any missing ones to the index. It also removes stale entries for nodes that no longer exist or already have embeddings. Use this on startup or after bulk imports.

func (*BadgerEngine) RegisterSnapshotReader 

func (b *BadgerEngine) RegisterSnapshotReader(info SnapshotReaderInfo) func()

RegisterSnapshotReader delegates reader registration when lifecycle is enabled.

func (*BadgerEngine) ResumeLifecycle 

func (b *BadgerEngine) ResumeLifecycle()

ResumeLifecycle resumes lifecycle work.

func (*BadgerEngine) RunGC 

func (b *BadgerEngine) RunGC() error

RunGC runs garbage collection on the BadgerDB value log. Should be called periodically for long-running applications.

func (*BadgerEngine) RunOnStartMigrations added in v1.1.0 

func (b *BadgerEngine) RunOnStartMigrations(allowUpgrade bool) error

RunOnStartMigrations advances the on-disk schema version to storageVersionCurrent if (and only if) the operator has authorized the upgrade via allowUpgrade. Without authorization, an out-of-date store causes RunOnStartMigrations to return ErrStorageUpgradeRequired and the engine refuses to open.

Migrations run in order based on the on-disk version, each preserving the prior step's invariants: 

V0 → V1: extracts legacy access state into AccessMeta records.
V1 → V2: eager rewrite of every node and edge body to the tokenized
         property-key codec; bumps the version after a clean pass.

Sets engine.storageVersion to the post-migration version so the encode path can deterministically pick codecs from it.

func (*BadgerEngine) ScanAccessMeta added in v1.1.0 

func (b *BadgerEngine) ScanAccessMeta() ([]*knowledgepolicy.AccessMetaEntry, error)

func (*BadgerEngine) ScanPendingDeindexWorkItems added in v1.1.0 

func (b *BadgerEngine) ScanPendingDeindexWorkItems() ([]*DeindexWorkItem, error)

ScanPendingDeindexWorkItems returns all work items with status "pending".

func (*BadgerEngine) ScorerForNamespace added in v1.1.0 

func (b *BadgerEngine) ScorerForNamespace(namespace string) *knowledgepolicy.Scorer

ScorerForNamespace returns a Scorer for the given namespace, or nil.

func (*BadgerEngine) SetAccessAccumulator added in v1.1.0 

func (b *BadgerEngine) SetAccessAccumulator(acc accessIncrementor)

SetAccessAccumulator wires the P-local accumulator so read paths can record access events for nodes that survive visibility filtering.

func (*BadgerEngine) SetDecayEnabled added in v1.1.0 

func (b *BadgerEngine) SetDecayEnabled(enabled bool)

SetDecayEnabled enables or disables knowledge-layer decay scoring on read paths.

func (*BadgerEngine) SetEmbeddingsEnabled added in v1.1.0 

func (b *BadgerEngine) SetEmbeddingsEnabled(enabled bool)

SetEmbeddingsEnabled toggles the pending-embed index. When false, CreateNode and UpsertNode skip the pendingEmbedKey write on user nodes — no embed worker will consume the marker when embeddings are globally disabled, so the Set is pure write amplification. Wired from the server at startup alongside the decay flag (see pkg/nornicdb/db.go).

func (*BadgerEngine) SetLifecycleController 

func (b *BadgerEngine) SetLifecycleController(controller MVCCLifecycleController)

SetLifecycleController injects the MVCC lifecycle controller.

func (*BadgerEngine) SetLifecycleSchedule 

func (b *BadgerEngine) SetLifecycleSchedule(interval time.Duration) error

SetLifecycleSchedule updates lifecycle cadence when supported by the controller.

func (*BadgerEngine) SetRevealAll added in v1.1.0 

func (b *BadgerEngine) SetRevealAll(reveal bool)

SetRevealAll disables decay suppression for all entities in the current query. Call ClearRevealAll after the query completes.

func (*BadgerEngine) Size 

func (b *BadgerEngine) Size() (lsm, vlog int64)

Size returns the approximate size of the database in bytes.

func (*BadgerEngine) StartLifecycleManager 

func (b *BadgerEngine) StartLifecycleManager(ctx context.Context)

StartLifecycleManager starts the injected lifecycle manager.

func (*BadgerEngine) StreamEdges 

func (b *BadgerEngine) StreamEdges(ctx context.Context, fn func(edge *Edge) error) error

StreamEdges implements StreamingEngine.StreamEdges for memory-efficient iteration. Iterates through all edges one at a time without loading all into memory.

func (*BadgerEngine) StreamNodeChunks 

func (b *BadgerEngine) StreamNodeChunks(ctx context.Context, chunkSize int, fn func(nodes []*Node) error) error

StreamNodeChunks implements StreamingEngine.StreamNodeChunks for batch processing. Iterates through nodes in chunks, more efficient for batch operations.

func (*BadgerEngine) StreamNodes 

func (b *BadgerEngine) StreamNodes(ctx context.Context, fn func(node *Node) error) error

StreamNodes implements StreamingEngine.StreamNodes for memory-efficient iteration. Iterates through all nodes one at a time without loading all into memory.

func (*BadgerEngine) StreamNodesByPrefix 

func (b *BadgerEngine) StreamNodesByPrefix(ctx context.Context, prefix string, fn func(node *Node) error) error

StreamNodesByPrefix streams nodes whose IDs start with prefix. This is significantly faster than full StreamNodes + callback filtering when tenants/databases share a physical store.

func (*BadgerEngine) Sync 

func (b *BadgerEngine) Sync() error

Sync forces a sync of all data to disk. This is useful for ensuring durability before a crash.

func (*BadgerEngine) TopLifecycleDebtKeys 

func (b *BadgerEngine) TopLifecycleDebtKeys(limit int) []MVCCLifecycleDebtKey

TopLifecycleDebtKeys returns the highest-debt keys when supported by the controller.

func (*BadgerEngine) TriggerPruneNow 

func (b *BadgerEngine) TriggerPruneNow(ctx context.Context) error

TriggerPruneNow runs an immediate lifecycle prune.

func (*BadgerEngine) UpdateEdge 

func (b *BadgerEngine) UpdateEdge(edge *Edge) error

UpdateEdge updates an existing edge.

func (*BadgerEngine) UpdateEdgeCurrentHead 

func (b *BadgerEngine) UpdateEdgeCurrentHead(id EdgeID, version MVCCVersion, tombstoned bool) error

func (*BadgerEngine) UpdateNode 

func (b *BadgerEngine) UpdateNode(node *Node) error

UpdateNode updates an existing node or creates it if it doesn't exist (upsert).

func (*BadgerEngine) UpdateNodeCurrentHead 

func (b *BadgerEngine) UpdateNodeCurrentHead(id NodeID, version MVCCVersion, tombstoned bool) error

func (*BadgerEngine) UpdateNodeEmbedding 

func (b *BadgerEngine) UpdateNodeEmbedding(node *Node) error

UpdateNodeEmbedding updates only the embedding field of an existing node. Returns ErrNotFound if the node doesn't exist (does NOT create the node). This is used by the embedding queue to prevent creating orphaned nodes. REQUIRES: node.ID must be prefixed with namespace (e.g., "nornic:node-123").

func (*BadgerEngine) ValidateConstraintOnCreation 

func (b *BadgerEngine) ValidateConstraintOnCreation(c Constraint) error

ValidateConstraintOnCreation validates that all existing data satisfies the constraint. This is called when CREATE CONSTRAINT is executed, matching Neo4j behavior.

func (*BadgerEngine) ValidatePropertyTypeConstraintOnCreation 

func (b *BadgerEngine) ValidatePropertyTypeConstraintOnCreation(ptc PropertyTypeConstraint) error

ValidatePropertyTypeConstraintOnCreation validates existing data against type constraint.

func (*BadgerEngine) ValidateRelationshipConstraint 

func (b *BadgerEngine) ValidateRelationshipConstraint(rc RelationshipConstraint) error

ValidateRelationshipConstraint validates relationship property constraints.

func (*BadgerEngine) WriteIndexTombstones added in v1.1.0 

func (b *BadgerEngine) WriteIndexTombstones(keys [][]byte) error

WriteIndexTombstones writes zero-length presence markers for all given original index keys in a single batched transaction.

func (*BadgerEngine) WriteMVCCHead 

func (b *BadgerEngine) WriteMVCCHead(ctx context.Context, logicalKey []byte, head MVCCHead) error

WriteMVCCHead writes an updated MVCC head for a logical key.

type BadgerOptions 

type BadgerOptions struct {
	// DataDir is the directory for storing data files.
	// Required.
	DataDir string

	// InMemory runs BadgerDB in memory-only mode.
	// Useful for testing. Data is not persisted.
	InMemory bool

	// SyncWrites forces fsync after each write.
	// Slower but more durable.
	SyncWrites bool

	// BadgerInternalLogger is the logger handed to BadgerDB itself for its
	// own internal logging (compaction, value-log GC, etc.). If nil, BadgerDB's
	// quiet/default logger is used. This field replaces the previous
	// BadgerOptions.Logger which collided with the new structured *slog.Logger
	// field below; rename is internal-only (no external setter ever existed
	// outside doc comments — verified 2026-05-01).
	BadgerInternalLogger badger.Logger

	// Logger is the structured *slog.Logger threaded into the storage engine.
	// D-01 logger DI: optional; nil falls back to a discard handler at ctor
	// entry per D-01a so existing callers (and tests) compile unchanged.
	// Once set, BadgerEngine emits all operational diagnostics through this
	// logger tagged with component=storage, engine=badger.
	Logger *slog.Logger

	// LowMemory enables memory-constrained settings.
	// Reduces MemTableSize and other buffers to use less RAM.
	LowMemory bool

	// HighPerformance enables aggressive caching and larger buffers.
	// Uses more RAM but significantly faster writes/reads.
	HighPerformance bool

	// EncryptionKey is the 16, 24, or 32 byte key for AES encryption.
	// If provided, all data will be encrypted at rest using AES-CTR.
	// WARNING: If you lose this key, your data is irrecoverable!
	// Leave empty to disable encryption.
	EncryptionKey []byte

	// AllowStorageUpgrade authorizes the engine to advance the on-disk
	// storage version through whichever migration arms this binary
	// understands. Without it, a binary that opens a data directory
	// older than its own version refuses to start. The upgrade is
	// one-way; operators should back up before passing this flag.
	AllowStorageUpgrade bool

	// NodeCacheMaxEntries is the maximum number of nodes held in the in-process
	// hot node cache (used by GetNode). When exceeded, the cache is cleared.
	// Set to 0 to use the default.
	NodeCacheMaxEntries int

	// EdgeTypeCacheMaxTypes is the maximum number of distinct edge types cached
	// for GetEdgesByType. When exceeded, the cache is cleared.
	// Set to 0 to use the default.
	EdgeTypeCacheMaxTypes int

	// LabelFirstNodeCacheMaxEntries is the maximum number of labels cached for
	// fast label-first lookups. When exceeded, the cache is cleared.
	// Set to 0 to use the default.
	LabelFirstNodeCacheMaxEntries int

	// EngineOptions holds engine-wide MVCC retention defaults and similar policies.
	EngineOptions EngineOptions
}

BadgerOptions configures the BadgerDB engine.

type BadgerTransaction 

type BadgerTransaction struct {

	// Transaction identity
	ID        string
	StartTime time.Time
	Status    TransactionStatus

	// CommitVersion is assigned once for a successful commit that mutates storage.
	CommitVersion MVCCVersion

	// Transaction metadata (for logging/debugging)
	Metadata map[string]interface{}
	// contains filtered or unexported fields
}

BadgerTransaction wraps Badger's native transaction with constraint validation.

Provides ACID guarantees:

  • Atomicity: All operations commit together or none do
  • Consistency: Constraints are validated before commit
  • Isolation: Changes invisible until commit
  • Durability: Badger's WAL ensures persistence

func (*BadgerTransaction) AllNodes added in v1.0.42 

func (tx *BadgerTransaction) AllNodes() ([]*Node, error)

AllNodes returns all visible nodes including pending transaction writes.

func (*BadgerTransaction) BulkCreateEdges added in v1.1.0 

func (tx *BadgerTransaction) BulkCreateEdges(edges []*Edge) error

BulkCreateEdges buffers many edges in one pass, amortizing the lock, the lifecycle check, and — most importantly — the committed-node existence probes. A batch of N edges that share K distinct endpoint nodes (K ≪ 2N in practice for graph fan-in/fan-out) now costs K Badger point-reads instead of up to 2N.

func (*BadgerTransaction) Commit 

func (tx *BadgerTransaction) Commit() error

Commit applies all changes atomically with full constraint validation. Explicit transactions get strict ACID durability with immediate fsync.

func (*BadgerTransaction) CreateEdge 

func (tx *BadgerTransaction) CreateEdge(edge *Edge) error

CreateEdge adds an edge to the transaction.

func (*BadgerTransaction) CreateNode 

func (tx *BadgerTransaction) CreateNode(node *Node) (NodeID, error)

CreateNode adds a node to the transaction with constraint validation. REQUIRES: node.ID must be prefixed with namespace (e.g., "nornic:node-123"). This enforces that all nodes are namespaced at the storage layer.

func (*BadgerTransaction) DeleteEdge 

func (tx *BadgerTransaction) DeleteEdge(edgeID EdgeID) error

DeleteEdge deletes an edge from the transaction.

func (*BadgerTransaction) DeleteNode 

func (tx *BadgerTransaction) DeleteNode(nodeID NodeID) error

DeleteNode deletes a node from the transaction.

func (*BadgerTransaction) GetAllNodes added in v1.0.42 

func (tx *BadgerTransaction) GetAllNodes() []*Node

func (*BadgerTransaction) GetEdge 

func (tx *BadgerTransaction) GetEdge(edgeID EdgeID) (*Edge, error)

GetEdge retrieves an edge with read-your-writes semantics.

Like GetNode, reads pin the transaction to the edge's namespace so the readTS is bound to the namespace's actual sequence rather than the pre-pin sequence-0 placeholder.

func (*BadgerTransaction) GetEdgeBetween 

func (tx *BadgerTransaction) GetEdgeBetween(startID, endID NodeID, edgeType string) *Edge

GetEdgeBetween returns a matching edge including pending transaction writes.

func (*BadgerTransaction) GetEdgesBetween 

func (tx *BadgerTransaction) GetEdgesBetween(startID, endID NodeID) ([]*Edge, error)

GetEdgesBetween returns edges between two nodes including pending transaction writes.

func (*BadgerTransaction) GetEdgesByType 

func (tx *BadgerTransaction) GetEdgesByType(edgeType string) ([]*Edge, error)

GetEdgesByType returns edges of a given type including pending transaction writes.

func (*BadgerTransaction) GetFirstNodeByLabel added in v1.0.42 

func (tx *BadgerTransaction) GetFirstNodeByLabel(label string) (*Node, error)

GetFirstNodeByLabel returns the first visible node with the given label.

func (*BadgerTransaction) GetIncomingEdges 

func (tx *BadgerTransaction) GetIncomingEdges(nodeID NodeID) ([]*Edge, error)

GetIncomingEdges returns incoming edges including pending transaction writes.

func (*BadgerTransaction) GetMetadata 

func (tx *BadgerTransaction) GetMetadata() map[string]interface{}

GetMetadata returns transaction metadata copy.

func (*BadgerTransaction) GetNode 

func (tx *BadgerTransaction) GetNode(nodeID NodeID) (*Node, error)

GetNode retrieves a node (read-your-writes).

Reads pin the transaction to the node's namespace if it isn't already pinned. Without this, a transaction's pre-pin readTS sits at sequence 0 and visibility checks against a head whose FloorVersion is non-zero reject every committed value as "not yet visible" — which manifests as lost updates under the Read-Modify-Write retry loop in DB.Update.

func (*BadgerTransaction) GetNodesByLabel added in v1.0.42 

func (tx *BadgerTransaction) GetNodesByLabel(label string) ([]*Node, error)

GetNodesByLabel returns nodes with the given label including pending transaction writes.

func (*BadgerTransaction) GetOutgoingEdges 

func (tx *BadgerTransaction) GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)

GetOutgoingEdges returns outgoing edges including pending transaction writes.

func (*BadgerTransaction) HasPendingNodeMutations added in v1.0.42 

func (tx *BadgerTransaction) HasPendingNodeMutations() bool

func (*BadgerTransaction) IsActive 

func (tx *BadgerTransaction) IsActive() bool

IsActive returns true if the transaction is still active.

func (*BadgerTransaction) Namespace added in v1.1.1 

func (tx *BadgerTransaction) Namespace() string

Namespace returns the database namespace this transaction is pinned to, or "" if no namespaced write has been recorded yet. Once set, every subsequent write must share this namespace.

func (*BadgerTransaction) OperationCount 

func (tx *BadgerTransaction) OperationCount() int

OperationCount returns the number of buffered operations.

func (*BadgerTransaction) Rollback 

func (tx *BadgerTransaction) Rollback() error

Rollback discards all changes.

func (*BadgerTransaction) SetDeferredConstraintValidation 

func (tx *BadgerTransaction) SetDeferredConstraintValidation(deferValidation bool) error

SetDeferredConstraintValidation controls per-operation constraint checks. When enabled, constraints are enforced at commit time only.

func (*BadgerTransaction) SetImplicit added in v1.1.0 

func (tx *BadgerTransaction) SetImplicit(implicit bool) error

SetImplicit marks this transaction as implicit (auto-opened by the executor for a single Cypher statement, no user BEGIN). Implicit transactions skip the per-Commit engine.Sync() because the Bolt session end and the async flush loop coalesce durability for them.

func (*BadgerTransaction) SetMetadata 

func (tx *BadgerTransaction) SetMetadata(metadata map[string]interface{}) error

SetMetadata sets transaction metadata (same as Transaction).

func (*BadgerTransaction) SetNamespace added in v1.1.1 

func (tx *BadgerTransaction) SetNamespace(ns string) error

SetNamespace eagerly pins the transaction to ns. Callers that already know the target namespace at BeginTransaction time (the cypher executor's transactionStorageWrapper, for example) use this to fail fast on misrouted writes instead of waiting for the first prefixed mutation. Returns an error if the transaction is already pinned to a different namespace.

SetNamespace deliberately does not call ensureLifecycleActiveLocked before the namespace bind: lifecycle expiration is meaningless for a transaction that has not yet registered against any namespace. We only require Status==active.

func (*BadgerTransaction) SetSkipCreateExistenceCheck 

func (tx *BadgerTransaction) SetSkipCreateExistenceCheck(skip bool) error

SetSkipCreateExistenceCheck controls read-before-write checks for CREATE. When enabled, CREATE skips the storage existence read for UUID IDs.

func (*BadgerTransaction) UpdateEdge 

func (tx *BadgerTransaction) UpdateEdge(edge *Edge) error

UpdateEdge updates an existing edge within the transaction.

This is required so Cypher can do CREATE ... SET r.prop = ... in a single query while using implicit/explicit transactions (writes must remain isolated until commit).

func (*BadgerTransaction) UpdateNode 

func (tx *BadgerTransaction) UpdateNode(node *Node) error

UpdateNode updates a node in the transaction.

type BatchWriter 

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

BatchWriter provides explicit batch commit control for bulk operations. Instead of syncing after each write (even in batch mode), BatchWriter buffers all writes and only syncs when Commit() is called.

This dramatically improves bulk write throughput by reducing fsync calls. Use for imports, migrations, or any operation writing many records.

Example: 

batch := wal.NewBatch()
for _, node := range nodes {
    batch.AppendNode(OpCreateNode, node)
}
if err := batch.Commit(); err != nil {
    batch.Rollback() // Discard uncommitted entries
}

Performance:

  • Single fsync at end instead of per-write
  • 10-100x faster for bulk operations
  • Memory usage proportional to batch size

IMPORTANT: Sequence numbers are assigned at commit time, not append time. This ensures entries are written in sequence order, preventing replay issues when mixing batch and non-batch operations.

func (*BatchWriter) AppendDelete 

func (b *BatchWriter) AppendDelete(op OperationType, id string) error

AppendDelete adds a delete operation to the batch.

func (*BatchWriter) AppendEdge 

func (b *BatchWriter) AppendEdge(op OperationType, edge *Edge) error

AppendEdge adds an edge operation to the batch.

func (*BatchWriter) AppendNode 

func (b *BatchWriter) AppendNode(op OperationType, node *Node) error

AppendNode adds a node operation to the batch.

func (*BatchWriter) Commit 

func (b *BatchWriter) Commit() error

Commit writes all batched entries and syncs to disk. Sequence numbers are assigned here to ensure proper ordering. This is the only fsync in the batch - much faster than per-write sync.

func (*BatchWriter) CommitWithSeq 

func (b *BatchWriter) CommitWithSeq() (uint64, uint64, error)

CommitWithSeq writes all batched entries, syncs to disk, and returns the sequence range assigned to the batch. If the batch is empty or WAL is disabled, returns (0, 0, nil).

func (*BatchWriter) Len 

func (b *BatchWriter) Len() int

Len returns the number of pending entries.

func (*BatchWriter) Rollback 

func (b *BatchWriter) Rollback()

Rollback discards all uncommitted entries.

type BytesMetricsSweeper added in v1.1.0 

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

BytesMetricsSweeper is the lifecycle.Component that populates the nornicdb_storage_bytes{kind} gauges every interval via badger.DB.EstimateSize(prefix).

func NewBytesMetricsSweeper added in v1.1.0 

func NewBytesMetricsSweeper(metrics *observability.StorageMetrics, db *badger.DB, searchSize SearchSizeFn, interval time.Duration) *BytesMetricsSweeper

NewBytesMetricsSweeper constructs the sweeper. interval ≤ 0 falls back to DefaultBytesMetricsInterval. metrics or db nil disables the sweep (Start returns nil immediately) — defensive for partial-init.

func (*BytesMetricsSweeper) Name added in v1.1.0 

func (s *BytesMetricsSweeper) Name() string

Name implements lifecycle.Component.

func (*BytesMetricsSweeper) Shutdown added in v1.1.0 

func (s *BytesMetricsSweeper) Shutdown(ctx context.Context) error

Shutdown implements lifecycle.Component. Idempotent; subsequent calls are no-ops.

func (*BytesMetricsSweeper) Start added in v1.1.0 

func (s *BytesMetricsSweeper) Start(ctx context.Context) error

Start implements lifecycle.Component. Blocks until ctx is cancelled or Shutdown is called. Performs an initial sweep immediately so the gauges have non-zero values by the first scrape.

type CompositeEngine 

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

CompositeEngine is a storage engine that spans multiple constituent databases. It implements the Engine interface by routing operations to constituents and merging results.

This enables composite databases (similar to Neo4j Fabric) where a single database view spans multiple physical databases.

func NewCompositeEngine 

func NewCompositeEngine(
	constituents map[string]Engine,
	constituentNames map[string]string,
	accessModes map[string]string,
) *CompositeEngine

NewCompositeEngine creates a new composite engine that spans multiple constituent databases.

Parameters:

  • constituents: Map of alias -> storage engine for each constituent
  • constituentNames: Map of alias -> actual database name
  • accessModes: Map of alias -> access mode ("read", "write", "read_write")

func (*CompositeEngine) AllEdges 

func (c *CompositeEngine) AllEdges() ([]*Edge, error)

AllEdges returns all edges from all constituents. Duplicate edges (same ID) are deduplicated - only the first occurrence is kept.

func (*CompositeEngine) AllNodes 

func (c *CompositeEngine) AllNodes() ([]*Node, error)

AllNodes returns all nodes from all constituents. Duplicate nodes (same ID) are deduplicated - only the first occurrence is kept.

func (*CompositeEngine) BatchGetNodes 

func (c *CompositeEngine) BatchGetNodes(ids []NodeID) (map[NodeID]*Node, error)

BatchGetNodes retrieves multiple nodes from all constituents.

func (*CompositeEngine) BulkCreateEdges 

func (c *CompositeEngine) BulkCreateEdges(edges []*Edge) error

BulkCreateEdges creates edges in bulk. Routes edges to constituents containing their start nodes.

func (*CompositeEngine) BulkCreateNodes 

func (c *CompositeEngine) BulkCreateNodes(nodes []*Node) error

BulkCreateNodes creates nodes in bulk. Routes nodes to appropriate constituents based on routing rules.

func (*CompositeEngine) BulkDeleteEdges 

func (c *CompositeEngine) BulkDeleteEdges(ids []EdgeID) error

BulkDeleteEdges deletes edges in bulk from all constituents.

func (*CompositeEngine) BulkDeleteNodes 

func (c *CompositeEngine) BulkDeleteNodes(ids []NodeID) error

BulkDeleteNodes deletes nodes in bulk from all constituents.

func (*CompositeEngine) Close 

func (c *CompositeEngine) Close() error

Close closes all constituent engines.

func (*CompositeEngine) CreateEdge 

func (c *CompositeEngine) CreateEdge(edge *Edge) error

CreateEdge creates an edge. Routes to the constituent containing the start node. Following Neo4j TransactionState pattern: check if nodes were created in current transaction, and if so, try that constituent first. Otherwise try all constituents.

func (*CompositeEngine) CreateNode 

func (c *CompositeEngine) CreateNode(node *Node) (NodeID, error)

CreateNode creates a node. Routes to the appropriate constituent based on routing rules.

func (*CompositeEngine) DeleteByPrefix 

func (c *CompositeEngine) DeleteByPrefix(prefix string) (nodesDeleted int64, edgesDeleted int64, err error)

DeleteByPrefix is not supported for composite databases. Composite databases don't have their own data - they're virtual views.

func (*CompositeEngine) DeleteEdge 

func (c *CompositeEngine) DeleteEdge(id EdgeID) error

DeleteEdge deletes an edge from the constituent containing it.

func (*CompositeEngine) DeleteNode 

func (c *CompositeEngine) DeleteNode(id NodeID) error

DeleteNode deletes a node from the constituent that contains it.

func (*CompositeEngine) EdgeCount 

func (c *CompositeEngine) EdgeCount() (int64, error)

EdgeCount returns the total edge count across all constituents.

func (*CompositeEngine) GetAllNodes 

func (c *CompositeEngine) GetAllNodes() []*Node

GetAllNodes returns all nodes from all constituents (non-error version).

func (*CompositeEngine) GetConstituentByAlias 

func (c *CompositeEngine) GetConstituentByAlias(alias string) (Engine, error)

GetConstituentByAlias returns the storage engine for a specific constituent within this composite database. This is used by the Cypher executor to resolve USE composite.alias references.

func (*CompositeEngine) GetEdge 

func (c *CompositeEngine) GetEdge(id EdgeID) (*Edge, error)

GetEdge retrieves an edge. Searches all constituents.

func (*CompositeEngine) GetEdgeBetween 

func (c *CompositeEngine) GetEdgeBetween(startID, endID NodeID, edgeType string) *Edge

GetEdgeBetween returns an edge between two nodes from any constituent.

func (*CompositeEngine) GetEdgesBetween 

func (c *CompositeEngine) GetEdgesBetween(startID, endID NodeID) ([]*Edge, error)

GetEdgesBetween returns edges between two nodes from all constituents. Duplicate edges (same ID) are deduplicated - only the first occurrence is kept.

func (*CompositeEngine) GetEdgesByType 

func (c *CompositeEngine) GetEdgesByType(edgeType string) ([]*Edge, error)

GetEdgesByType returns edges of the given type from all constituents. Duplicate edges (same ID) are deduplicated - only the first occurrence is kept.

func (*CompositeEngine) GetFirstNodeByLabel 

func (c *CompositeEngine) GetFirstNodeByLabel(label string) (*Node, error)

GetFirstNodeByLabel returns the first node with the given label from any constituent.

func (*CompositeEngine) GetInDegree 

func (c *CompositeEngine) GetInDegree(nodeID NodeID) int

GetInDegree returns the in-degree of a node across all constituents.

func (*CompositeEngine) GetIncomingEdges 

func (c *CompositeEngine) GetIncomingEdges(nodeID NodeID) ([]*Edge, error)

GetIncomingEdges returns incoming edges from all constituents. Duplicate edges (same ID) are deduplicated - only the first occurrence is kept.

func (*CompositeEngine) GetNode 

func (c *CompositeEngine) GetNode(id NodeID) (*Node, error)

GetNode retrieves a node. Searches all readable constituents.

func (*CompositeEngine) GetNodesByLabel 

func (c *CompositeEngine) GetNodesByLabel(label string) ([]*Node, error)

GetNodesByLabel returns nodes with the given label from all constituents. Duplicate nodes (same ID) are deduplicated - only the first occurrence is kept. Returns empty slice if no readable constituents are available.

func (*CompositeEngine) GetOutDegree 

func (c *CompositeEngine) GetOutDegree(nodeID NodeID) int

GetOutDegree returns the out-degree of a node across all constituents.

func (*CompositeEngine) GetOutgoingEdges 

func (c *CompositeEngine) GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)

GetOutgoingEdges returns outgoing edges from all constituents. Duplicate edges (same ID) are deduplicated - only the first occurrence is kept.

func (*CompositeEngine) GetSchema 

func (c *CompositeEngine) GetSchema() *SchemaManager

GetSchema returns a merged schema from all constituents. Merges constraints and indexes from all constituent databases.

func (*CompositeEngine) IsComposite 

func (c *CompositeEngine) IsComposite() bool

IsComposite returns true, identifying this engine as a composite database. This enables type-assertion-free composite detection via interface check: 

type compositeChecker interface { IsComposite() bool }
if cc, ok := engine.(compositeChecker); ok && cc.IsComposite() { ... }

func (*CompositeEngine) NodeCount 

func (c *CompositeEngine) NodeCount() (int64, error)

NodeCount returns the total node count across all constituents.

func (*CompositeEngine) SetLabelRouting 

func (c *CompositeEngine) SetLabelRouting(label string, constituents []string)

SetLabelRouting configures label-based routing for a specific label. This allows explicit configuration of which constituents should handle nodes with a given label.

func (*CompositeEngine) SetPropertyDefault 

func (c *CompositeEngine) SetPropertyDefault(propertyName string, constituent string)

SetPropertyDefault sets the default constituent for a property when the value is not found in routing rules.

func (*CompositeEngine) SetPropertyRouting 

func (c *CompositeEngine) SetPropertyRouting(propertyName string, value interface{}, constituent string)

SetPropertyRouting configures property-based routing for a specific property value. This enables routing based on property values (e.g., database_id).

func (*CompositeEngine) StreamEdges 

func (c *CompositeEngine) StreamEdges(ctx context.Context, fn func(edge *Edge) error) error

StreamEdges streams edges from all constituents.

func (*CompositeEngine) StreamNodeChunks 

func (c *CompositeEngine) StreamNodeChunks(ctx context.Context, chunkSize int, fn func(nodes []*Node) error) error

StreamNodeChunks streams nodes in chunks from all constituents.

func (*CompositeEngine) StreamNodes 

func (c *CompositeEngine) StreamNodes(ctx context.Context, fn func(node *Node) error) error

StreamNodes streams nodes from all constituents.

func (*CompositeEngine) UpdateEdge 

func (c *CompositeEngine) UpdateEdge(edge *Edge) error

UpdateEdge updates an edge. Routes to the constituent containing it.

func (*CompositeEngine) UpdateNode 

func (c *CompositeEngine) UpdateNode(node *Node) error

UpdateNode updates a node. Routes to the constituent that contains it.

type CompositeIndex 

type CompositeIndex struct {
	Name       string
	Label      string
	Properties []string // Ordered list of property names
	// contains filtered or unexported fields
}

CompositeIndex represents an index on multiple properties for efficient multi-property queries. This is Neo4j's composite index equivalent.

Composite indexes support:

  • Full key lookups (all properties specified)
  • Prefix lookups (leading properties specified, for ordered access)
  • Range queries on the last property in a prefix

func (*CompositeIndex) IndexNode 

func (idx *CompositeIndex) IndexNode(nodeID NodeID, properties map[string]interface{}) error

IndexNodeComposite indexes a node in a composite index. Call this when creating or updating a node with the indexed properties.

func (*CompositeIndex) LookupFull 

func (idx *CompositeIndex) LookupFull(values ...interface{}) []NodeID

LookupFull finds nodes matching all property values exactly. All properties in the composite index must be specified.

func (*CompositeIndex) LookupPrefix 

func (idx *CompositeIndex) LookupPrefix(values ...interface{}) []NodeID

LookupPrefix finds nodes matching a prefix of property values. Specify 1 to N-1 property values (where N is total properties in index). Returns all nodes that match the prefix.

Example: For index on (country, city, zipcode)

  • LookupPrefix("US") returns all nodes in the US
  • LookupPrefix("US", "NYC") returns all nodes in NYC, US

func (*CompositeIndex) LookupWithFilter 

func (idx *CompositeIndex) LookupWithFilter(filter func(NodeID) bool, values ...interface{}) []NodeID

LookupWithFilter finds nodes using a prefix and applies a filter function. This enables more complex queries like range queries on the last property.

Example: Find all users in "US", "NYC" with zipcode > "10000" 

idx.LookupWithFilter(func(n NodeID, props map[string]interface{}) bool {
    zip := props["zipcode"].(string)
    return zip > "10000"
}, "US", "NYC")

func (*CompositeIndex) RemoveNode 

func (idx *CompositeIndex) RemoveNode(nodeID NodeID, properties map[string]interface{})

RemoveNode removes a node from the composite index. Call this when deleting a node or updating its indexed properties.

func (*CompositeIndex) Stats 

func (idx *CompositeIndex) Stats() map[string]interface{}

Stats returns statistics about the composite index.

type CompositeKey 

type CompositeKey struct {
	Hash   string        // SHA256 hash of encoded values (for map lookup)
	Values []interface{} // Original values (for debugging/display)
}

CompositeKey represents a key composed of multiple property values. The key is a hash of all property values in order for efficient lookup.

func NewCompositeKey 

func NewCompositeKey(values ...interface{}) CompositeKey

NewCompositeKey creates a composite key from multiple property values.

Composite keys enable uniqueness constraints and indexes on multiple properties together (e.g., unique combination of firstName + lastName). The key is hashed using SHA-256 for efficient map lookups while preserving the original values.

Parameters:

  • values: Variable number of property values to combine

Returns:

  • CompositeKey with hash for lookup and original values

Example 1 - Unique Person Name: 

// Ensure no two people have the same first AND last name combination
key := storage.NewCompositeKey("Alice", "Johnson")
// key.Hash = "a1b2c3..." (SHA-256)
// key.Values = ["Alice", "Johnson"]

// Can store in map for O(1) lookup
uniqueKeys := make(map[string]bool)
uniqueKeys[key.Hash] = true

Example 2 - Multi-Column Unique Constraint: 

// Email + domain must be unique together
key1 := storage.NewCompositeKey("user", "example.com")
key2 := storage.NewCompositeKey("user", "different.com")
// key1.Hash != key2.Hash (different combinations)

key3 := storage.NewCompositeKey("user", "example.com")
// key3.Hash == key1.Hash (same combination)

Example 3 - Geographic Uniqueness: 

// Store locations - no duplicate (lat, lon) pairs
locations := make(map[string]storage.NodeID)

loc1 := storage.NewCompositeKey(40.7128, -74.0060) // NYC
locations[loc1.Hash] = storage.NodeID("loc-nyc")

loc2 := storage.NewCompositeKey(40.7128, -74.0060) // Same coords
if _, exists := locations[loc2.Hash]; exists {
	// the configured logger should emit "Location already exists!"
	_ = exists
}

ELI12:

Imagine you're making sure no two people in your class have the SAME full name (first + last together):

  • Alice Smith → Create a "fingerprint" (hash) from "Alice" + "Smith"
  • Bob Johnson → Different fingerprint
  • Alice Smith → SAME fingerprint as the first Alice Smith!

The hash is like a unique barcode for the combination. If two combinations have the same barcode, they're duplicates!

Why hash instead of just combining strings?

  • Fast lookups (constant time)
  • Handles any data types (numbers, strings, booleans)
  • Consistent length (SHA-256 always 64 chars)

Use Cases:

  • Composite unique constraints (email + database_id)
  • Multi-column indexes
  • Deduplication of complex records

func (CompositeKey) String 

func (ck CompositeKey) String() string

String returns a human-readable representation of the composite key.

type Constraint 

type Constraint struct {
	Name          string               `json:"name"`
	Type          ConstraintType       `json:"type"`
	EntityType    ConstraintEntityType `json:"entity_type,omitempty"` // defaults to NODE when empty
	Label         string               `json:"label"`                 // label for nodes, relationship type for relationships
	Properties    []string             `json:"properties,omitempty"`
	OwnedIndex    string               `json:"owned_index,omitempty"`    // name of the owned backing index (for uniqueness/key)
	AllowedValues []interface{}        `json:"allowed_values,omitempty"` // for DOMAIN constraints: list of allowed values
	MaxCount      int                  `json:"max_count,omitempty"`      // for CARDINALITY constraints: maximum edge count per node
	Direction     string               `json:"direction,omitempty"`      // for CARDINALITY constraints: "OUTGOING" or "INCOMING"
	SourceLabel   string               `json:"source_label,omitempty"`   // for RELATIONSHIP_POLICY constraints: required source node label
	TargetLabel   string               `json:"target_label,omitempty"`   // for RELATIONSHIP_POLICY constraints: required target node label
	PolicyMode    string               `json:"policy_mode,omitempty"`    // for RELATIONSHIP_POLICY constraints: "ALLOWED" or "DISALLOWED"
}

Constraint represents a Neo4j-compatible schema constraint.

func (Constraint) EffectiveEntityType 

func (c Constraint) EffectiveEntityType() ConstraintEntityType

EffectiveEntityType returns the entity type, defaulting to NODE for backward compatibility.

type ConstraintContract 

type ConstraintContract struct {
	Name              string                    `json:"name"`
	TargetEntityType  string                    `json:"target_entity_type"`
	TargetLabelOrType string                    `json:"target_label_or_type"`
	Definition        string                    `json:"definition"`
	Entries           []ConstraintContractEntry `json:"entries,omitempty"`
}

type ConstraintContractEntry 

type ConstraintContractEntry struct {
	Kind          string   `json:"kind"`
	PrimitiveType string   `json:"primitive_type,omitempty"`
	Properties    []string `json:"properties,omitempty"`
	Property      string   `json:"property,omitempty"`
	ExpectedType  string   `json:"expected_type,omitempty"`
	Expression    string   `json:"expression,omitempty"`
}

type ConstraintEntityType 

type ConstraintEntityType string

ConstraintEntityType distinguishes node constraints from relationship constraints. 

const (
	ConstraintEntityNode         ConstraintEntityType = "NODE"
	ConstraintEntityRelationship ConstraintEntityType = "RELATIONSHIP"
)

type ConstraintType 

type ConstraintType string

ConstraintType represents the type of constraint. 

const (
	ConstraintUnique          ConstraintType = "UNIQUE"
	ConstraintNodeKey         ConstraintType = "NODE_KEY"
	ConstraintExists          ConstraintType = "EXISTS"
	ConstraintPropertyType    ConstraintType = "PROPERTY_TYPE"
	ConstraintTemporal        ConstraintType = "TEMPORAL_NO_OVERLAP"
	ConstraintRelationshipKey ConstraintType = "RELATIONSHIP_KEY"
	ConstraintDomain          ConstraintType = "DOMAIN"
	ConstraintCardinality     ConstraintType = "CARDINALITY"
	ConstraintPolicy          ConstraintType = "RELATIONSHIP_POLICY"
)

type ConstraintViolationError 

type ConstraintViolationError struct {
	Type       ConstraintType
	Label      string
	Properties []string
	Message    string
	Cause      error
}

ConstraintViolationError is returned when a constraint is violated.

Cause is set when the violation is detected at a non-validation layer — currently used by the snapshot-isolation check when a MERGE's UpdateNode hits a peer-committed node that already carries the matching unique value. Surfacing the underlying ErrConflict via Unwrap keeps errors.Is(err, ErrConflict) and the transient classifier working so retry-aware drivers still see the transient sentinel even though the visible message is the consumer-pinned constraint shape.

func (*ConstraintViolationError) Error 

func (e *ConstraintViolationError) Error() string

func (*ConstraintViolationError) Unwrap added in v1.1.1 

func (e *ConstraintViolationError) Unwrap() error

type CorruptionDiagnostics 

type CorruptionDiagnostics struct {
	Timestamp      time.Time `json:"timestamp"`
	WALPath        string    `json:"wal_path"`
	CorruptedSeq   uint64    `json:"corrupted_seq"`
	Operation      string    `json:"operation"`
	ExpectedCRC    uint32    `json:"expected_crc"`
	ActualCRC      uint32    `json:"actual_crc"`
	FileSize       int64     `json:"file_size"`
	LastGoodSeq    uint64    `json:"last_good_seq"`
	SuspectedCause string    `json:"suspected_cause"`
	BackupPath     string    `json:"backup_path,omitempty"`
	RecoveryAction string    `json:"recovery_action"`
}

CorruptionDiagnostics captures detailed information about WAL corruption to help diagnose root causes (disk failure, split-brain, bugs, etc.)

type DeindexCleanupJob added in v1.1.0 

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

DeindexCleanupJob periodically drains pending deindex work items, writes tombstones for their secondary-index keys, and marks them completed.

func NewDeindexCleanupJob added in v1.1.0 

func NewDeindexCleanupJob(engine *BadgerEngine, interval time.Duration) *DeindexCleanupJob

NewDeindexCleanupJob creates a cleanup job. Default interval is 24h.

func (*DeindexCleanupJob) RunOnce added in v1.1.0 

func (j *DeindexCleanupJob) RunOnce(ctx context.Context) (int, error)

RunOnce processes all pending deindex work items. Returns the number of entities successfully deindexed.

func (*DeindexCleanupJob) Start added in v1.1.0 

func (j *DeindexCleanupJob) Start(ctx context.Context)

func (*DeindexCleanupJob) Stop added in v1.1.0 

func (j *DeindexCleanupJob) Stop()

type DeindexWorkItem added in v1.1.0 

type DeindexWorkItem struct {
	WorkItemID    string `msgpack:"workItemId"`
	TargetID      string `msgpack:"targetId"`
	TargetScope   string `msgpack:"targetScope"`
	EnqueuedAt    int64  `msgpack:"enqueuedAt"`
	NextAttemptAt int64  `msgpack:"nextAttemptAt"`
	RetryCount    int    `msgpack:"retryCount"`
	Status        string `msgpack:"status"`
}

DeindexWorkItem is a pending deindex task for an entity whose visibility score has dropped below the threshold. The background cleanup job drains these items and writes tombstones for the entity's secondary-index keys.

type Edge 

type Edge struct {
	ID         EdgeID         `json:"id"`
	StartNode  NodeID         `json:"startNode"`
	EndNode    NodeID         `json:"endNode"`
	Type       string         `json:"type"`
	Properties map[string]any `json:"properties"`

	// NornicDB extensions
	CreatedAt            time.Time `json:"-"`
	UpdatedAt            time.Time `json:"-"`
	Confidence           float64   `json:"-"`
	AutoGenerated        bool      `json:"-"`
	VisibilitySuppressed bool      `json:"-"`
}

Edge represents a directed graph relationship (arc) between two nodes.

Edges are directed connections that link nodes together, representing relationships like "Alice KNOWS Bob" or "Document CITES Paper". They follow the Neo4j relationship model with NornicDB extensions for automatic relationship inference and confidence scoring.

Core Neo4j Fields:

  • ID: Unique identifier for the relationship
  • StartNode: Source node ID (where the arrow starts)
  • EndNode: Target node ID (where the arrow points)
  • Type: Relationship type (e.g., "KNOWS", "FOLLOWS", "CONTAINS")
  • Properties: Key-value data about the relationship

NornicDB Extensions:

  • CreatedAt: When the relationship was created
  • Confidence: How certain we are this relationship exists (0.0-1.0)
  • AutoGenerated: True if detected by ML/inference, false if manually created

Example 1 - Social Network Relationship: 

edge := &storage.Edge{
	ID:         storage.EdgeID("friendship-123"),
	StartNode:  storage.NodeID("alice"),
	EndNode:    storage.NodeID("bob"),
	Type:       "KNOWS",
	Properties: map[string]any{
		"since":    "2020-01-15",
		"strength": "close_friend",
		"mutuality": true,
	},
	CreatedAt:     time.Now(),
	Confidence:    1.0,  // Manually created = 100% certain
	AutoGenerated: false,
}
engine.CreateEdge(edge)

Example 2 - Document Citation: 

citation := &storage.Edge{
	ID:        storage.EdgeID("cite-paper-5"),
	StartNode: storage.NodeID("paper-123"),
	EndNode:   storage.NodeID("paper-456"),
	Type:      "CITES",
	Properties: map[string]any{
		"context":    "Methods section",
		"page":       12,
		"importance": "high",
	},
	CreatedAt: time.Now(),
	Confidence: 1.0,
}

Example 3 - Auto-Detected Semantic Relationship: 

// NornicDB inference engine detected similarity
autoEdge := &storage.Edge{
	ID:            storage.EdgeID("similar-42"),
	StartNode:     storage.NodeID("note-1"),
	EndNode:       storage.NodeID("note-2"),
	Type:          "SIMILAR_TO",
	Confidence:    0.87,  // 87% confidence from embedding similarity
	AutoGenerated: true,  // Created automatically
	Properties: map[string]any{
		"similarity":  0.87,
		"method":      "cosine_similarity",
		"detected_at": time.Now(),
		"reason":      "High semantic similarity in embeddings",
	},
}

ELI12:

Think of an Edge like a string connecting two beads (nodes):

  • StartNode: The first bead (where you start)
  • EndNode: The second bead (where the string goes to)
  • Type: What kind of string? ("FRIENDS_WITH", "PARENT_OF", "LIKES")
  • Properties: Info about the connection ("since when?", "how strong?")

The arrow matters! "Alice KNOWS Bob" is different from "Bob KNOWS Alice" (they could both be true, but they're separate relationships).

NornicDB's cool additions:

  • Confidence: "I'm 85% sure these two things are related"
  • AutoGenerated: "I found this connection myself!" vs "A human told me"

Imagine your brain automatically connecting ideas: "Oh, these two notes seem related!" That's what AutoGenerated edges do - the system notices patterns and creates connections for you!

Neo4j Compatibility:

  • Type maps to Neo4j relationship type (e.g., -[:KNOWS]->)
  • StartNode/EndNode map to Neo4j node IDs
  • Properties map to Neo4j relationship properties
  • Direction is always preserved (Neo4j requirement)

Thread Safety: 

Edge structs are NOT thread-safe. The storage engine handles concurrency.

func CopyEdge 

func CopyEdge(edge *Edge) *Edge

CopyEdge is the exported version of copyEdge for use by other packages.

type EdgeDeleteCallback 

type EdgeDeleteCallback func(edgeID EdgeID)

EdgeDeleteCallback is called when an edge is successfully deleted from storage.

type EdgeEventCallback 

type EdgeEventCallback func(edge *Edge)

EdgeEventCallback is called when edge storage operations complete successfully.

type EdgeID 

type EdgeID string

EdgeID is a strongly-typed unique identifier for graph edges (relationships).

Similar to NodeID, provides type safety and API clarity.

Example: 

id := storage.EdgeID("follows-456")
edge, err := engine.GetEdge(id)

type EdgeMeta 

type EdgeMeta struct {
	// Edge identification
	EdgeID string `json:"edge_id"`
	Src    string `json:"src"`
	Dst    string `json:"dst"`
	Label  string `json:"label"`

	// Confidence and scoring
	Score         float64 `json:"score"`
	SignalType    string  `json:"signal_type"` // "coaccess", "similarity", "topology", "llm-infer", "manual"
	EvidenceCount int     `json:"evidence_count"`
	DecayState    float64 `json:"decay_state"`

	// Temporal
	Timestamp time.Time `json:"timestamp"`
	SessionID string    `json:"session_id,omitempty"`

	// Lifecycle
	Materialized bool   `json:"materialized"`
	Origin       string `json:"origin"` // agent/commit ID that created this

	// TLP-specific (topology link prediction)
	TopologyAlgorithm string  `json:"topology_algorithm,omitempty"`
	TopologyScore     float64 `json:"topology_score,omitempty"`
	SemanticScore     float64 `json:"semantic_score,omitempty"`

	// Method that generated this edge (detailed)
	Method string `json:"method,omitempty"`
	Reason string `json:"reason,omitempty"`

	// Additional metadata
	Metadata map[string]interface{} `json:"metadata,omitempty"`
}

EdgeMeta stores provenance for auto-generated edges. This is append-only for auditability.

type EdgeMetaKey 

type EdgeMetaKey struct {
	Src   string
	Dst   string
	Label string
}

EdgeMetaKey uniquely identifies an edge for provenance lookup.

func (EdgeMetaKey) String 

func (k EdgeMetaKey) String() string

String returns a string representation of the key.

type EdgeMetaStats 

type EdgeMetaStats struct {
	TotalRecords      int64
	TotalMaterialized int64
	UniqueEdges       int64
	BySignalType      map[string]int64
}

EdgeMetaStats provides observability into provenance tracking.

type EdgeMetaStore 

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

EdgeMetaStore manages edge provenance. Thread-safe and append-only for auditability.

func GlobalEdgeMetaStore 

func GlobalEdgeMetaStore() *EdgeMetaStore

GlobalEdgeMetaStore returns the global edge meta store singleton.

func NewEdgeMetaStore 

func NewEdgeMetaStore() *EdgeMetaStore

NewEdgeMetaStore creates a new edge metadata store.

func NewEdgeMetaStoreWithOptions 

func NewEdgeMetaStoreWithOptions(opts ...EdgeMetaStoreOption) *EdgeMetaStore

NewEdgeMetaStoreWithOptions creates a store with functional options.

func (*EdgeMetaStore) Append 

func (s *EdgeMetaStore) Append(ctx context.Context, meta EdgeMeta) error

Append adds a new evidence record (immutable). This is the primary method for recording edge provenance.

func (*EdgeMetaStore) AppendFromSuggestion 

func (s *EdgeMetaStore) AppendFromSuggestion(
	ctx context.Context,
	src, dst, label string,
	score float64,
	signalType, method, reason, sessionID, origin string,
	materialized bool,
) error

AppendFromSuggestion creates an EdgeMeta from inference suggestion data. Convenience method for inference engine integration.

func (*EdgeMetaStore) Cleanup 

func (s *EdgeMetaStore) Cleanup(maxAge time.Duration) int

Cleanup removes records older than maxAge. Returns the number of records removed.

func (*EdgeMetaStore) Clear 

func (s *EdgeMetaStore) Clear()

Clear removes all provenance records. Use with caution - audit trail will be lost.

func (*EdgeMetaStore) CountHistory 

func (s *EdgeMetaStore) CountHistory(src, dst, label string) int

CountHistory returns the number of provenance records for an edge.

func (*EdgeMetaStore) Export 

func (s *EdgeMetaStore) Export() []*EdgeMeta

Export returns all provenance records for backup/export.

func (*EdgeMetaStore) GetByOrigin 

func (s *EdgeMetaStore) GetByOrigin(ctx context.Context, origin string, limit int) ([]*EdgeMeta, error)

GetByOrigin returns provenance records by origin (agent/commit ID).

func (*EdgeMetaStore) GetBySession 

func (s *EdgeMetaStore) GetBySession(ctx context.Context, sessionID string, limit int) ([]*EdgeMeta, error)

GetBySession returns provenance records by session.

func (*EdgeMetaStore) GetBySignalType 

func (s *EdgeMetaStore) GetBySignalType(ctx context.Context, signalType string, limit int) ([]*EdgeMeta, error)

GetBySignalType filters by how edges were created.

func (*EdgeMetaStore) GetByTimeRange 

func (s *EdgeMetaStore) GetByTimeRange(ctx context.Context, start, end time.Time, limit int) ([]*EdgeMeta, error)

GetByTimeRange returns provenance records within a time range.

func (*EdgeMetaStore) GetHistory 

func (s *EdgeMetaStore) GetHistory(ctx context.Context, src, dst, label string) ([]*EdgeMeta, error)

GetHistory returns all evidence for an edge.

func (*EdgeMetaStore) GetLatest 

func (s *EdgeMetaStore) GetLatest(ctx context.Context, src, dst, label string) (*EdgeMeta, error)

GetLatest returns the most recent provenance record for an edge.

func (*EdgeMetaStore) GetMaterialized 

func (s *EdgeMetaStore) GetMaterialized(ctx context.Context, limit int) ([]*EdgeMeta, error)

GetMaterialized returns all materialized edges.

func (*EdgeMetaStore) HasHistory 

func (s *EdgeMetaStore) HasHistory(src, dst, label string) bool

HasHistory returns true if any provenance records exist for this edge.

func (*EdgeMetaStore) Import 

func (s *EdgeMetaStore) Import(records []*EdgeMeta) int

Import loads provenance records from backup/import. Existing records are NOT cleared - use Clear() first if needed.

func (*EdgeMetaStore) MarkMaterialized 

func (s *EdgeMetaStore) MarkMaterialized(ctx context.Context, src, dst, label, origin string) error

MarkMaterialized marks an edge as materialized (actually created). Appends a new record with Materialized=true.

func (*EdgeMetaStore) Size 

func (s *EdgeMetaStore) Size() int

Size returns the total number of provenance records.

func (*EdgeMetaStore) Stats 

func (s *EdgeMetaStore) Stats() EdgeMetaStats

Stats returns current provenance store statistics.

func (*EdgeMetaStore) UniqueEdgeCount 

func (s *EdgeMetaStore) UniqueEdgeCount() int

UniqueEdgeCount returns the number of unique edges with provenance.

type EdgeMetaStoreOption 

type EdgeMetaStoreOption func(*EdgeMetaStore)

EdgeMetaStoreOption configures an EdgeMetaStore.

type EdgeVisitor 

type EdgeVisitor func(edge *Edge) error

EdgeVisitor is a function called for each edge during streaming.

type Engine 

type Engine interface {
	// Node operations
	CreateNode(node *Node) (NodeID, error) // Returns the actual stored ID (may be prefixed for namespaced engines)
	GetNode(id NodeID) (*Node, error)
	UpdateNode(node *Node) error
	DeleteNode(id NodeID) error

	// Edge operations
	CreateEdge(edge *Edge) error
	GetEdge(id EdgeID) (*Edge, error)
	UpdateEdge(edge *Edge) error
	DeleteEdge(id EdgeID) error

	// Query operations
	GetNodesByLabel(label string) ([]*Node, error)
	GetFirstNodeByLabel(label string) (*Node, error) // Optimized for LIMIT 1
	GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)
	GetIncomingEdges(nodeID NodeID) ([]*Edge, error)
	GetEdgesBetween(startID, endID NodeID) ([]*Edge, error)
	GetEdgeBetween(startID, endID NodeID, edgeType string) *Edge
	GetEdgesByType(edgeType string) ([]*Edge, error) // Fast lookup by edge type
	AllNodes() ([]*Node, error)
	AllEdges() ([]*Edge, error)
	GetAllNodes() []*Node

	// Degree operations (for graph algorithms)
	GetInDegree(nodeID NodeID) int
	GetOutDegree(nodeID NodeID) int

	// Schema operations
	GetSchema() *SchemaManager

	// Bulk operations (for import)
	BulkCreateNodes(nodes []*Node) error
	BulkCreateEdges(edges []*Edge) error

	// Bulk delete operations (for async flush performance)
	BulkDeleteNodes(ids []NodeID) error
	BulkDeleteEdges(ids []EdgeID) error

	// Batch read operations (for traversal performance)
	// BatchGetNodes fetches multiple nodes in a single operation
	// Returns a map for O(1) lookup by ID
	BatchGetNodes(ids []NodeID) (map[NodeID]*Node, error)

	// Lifecycle
	Close() error

	// Stats
	NodeCount() (int64, error)
	EdgeCount() (int64, error)

	// DeleteByPrefix deletes all nodes and edges with IDs starting with the given prefix.
	// Used for DROP DATABASE operations to delete all data in a namespace.
	// Returns the number of nodes and edges deleted.
	//
	// This is an optional interface - engines that don't support it will return an error.
	// For multi-database support, this must be implemented.
	DeleteByPrefix(prefix string) (nodesDeleted int64, edgesDeleted int64, err error)
}

Engine defines the storage engine interface for graph database operations.

All Engine implementations MUST be:

  • Thread-safe: Safe for concurrent access from multiple goroutines
  • ACID-like: Operations are atomic within their scope
  • Idempotent where appropriate: CreateNode fails if ID exists

The interface provides standard graph database operations:

  • CRUD for nodes and edges
  • Label-based queries
  • Graph traversal (outgoing/incoming edges)
  • Bulk operations for import/export
  • Statistics

Implementations:

  • MemoryEngine: In-memory storage for testing and small datasets
  • BadgerEngine: Persistent disk storage (planned)

Example Usage: 

var engine storage.Engine
engine = storage.NewMemoryEngine()
defer engine.Close()

// Create data
node := &storage.Node{
	ID:     "n1",
	Labels: []string{"Person"},
	Properties: map[string]any{"name": "Alice"},
}
if _, err := engine.CreateNode(node); err != nil {
	log.Fatal(err)
}

// Query
people, _ := engine.GetNodesByLabel("Person")
// emit "Found N people" via the configured logger

// Traversal
outgoing, _ := engine.GetOutgoingEdges("n1")
for _, edge := range outgoing {
	// emit "<start> -> <end> [<type>]" via the configured logger
	_ = edge
}

type EngineOptions 

type EngineOptions struct {
	RetentionPolicy RetentionPolicy
	// IDFreelistTTL debounces numID recycling: after a node/edge is
	// pruned, its numID stays parked for this long before allocations
	// can reclaim it. Zero = engine default (30 seconds).
	IDFreelistTTL time.Duration
}

EngineOptions contains storage-engine-wide options shared across engine implementations.

type ErrStorageUpgradeRequired added in v1.1.0 

type ErrStorageUpgradeRequired struct {
	OnDisk, Current int
}

ErrStorageUpgradeRequired is returned by RunOnStartMigrations when the on-disk schema version is older than the binary's current version and the operator has not authorized an upgrade. Callers should surface this directly to the operator with the recommended remediation: back up the data directory, then restart with the --upgrade-storage flag.

func (*ErrStorageUpgradeRequired) Error added in v1.1.0 

func (e *ErrStorageUpgradeRequired) Error() string

type ExportableEngine 

type ExportableEngine interface {
	Engine
	AllNodes() ([]*Node, error)
	AllEdges() ([]*Edge, error)
}

ExportableEngine extends Engine with export capabilities.

type FlushResult 

type FlushResult struct {
	NodesWritten     int
	NodesFailed      int
	EdgesWritten     int
	EdgesFailed      int
	NodesDeleted     int
	EdgesDeleted     int
	DeletesFailed    int
	FailedNodeIDs    []NodeID // IDs that failed - still in cache for retry
	FailedEdgeIDs    []EdgeID // IDs that failed - still in cache for retry
	FirstNodeError   string
	FirstEdgeError   string
	FirstDeleteError string
}

FlushResult tracks the outcome of a flush operation for observability.

func (FlushResult) HasErrors 

func (r FlushResult) HasErrors() bool

HasErrors returns true if any flush operations failed.

type FulltextIndex 

type FulltextIndex struct {
	Name              string   `json:"name"`
	Labels            []string `json:"labels,omitempty"`
	RelationshipTypes []string `json:"relationship_types,omitempty"`
	Properties        []string `json:"properties"`
}

FulltextIndex represents a full-text search index.

An index is scoped by EITHER Labels (a node-scoped index, declared via CREATE FULLTEXT INDEX <name> FOR (n:Label) ON EACH [n.prop]) OR RelationshipTypes (a relationship-scoped index, declared via CREATE FULLTEXT INDEX <name> FOR ()-[r:Type]-() ON EACH [r.prop]). Exactly one of those slices is populated for any well-formed index; the runtime uses the populated slice to decide which storage scan to drive.

Both Labels and RelationshipTypes use omitempty so an index that only carries one kind serializes without a stray empty array for the other. RelationshipTypes was added after the initial release; older databases serialize without it and load cleanly because JSON unmarshal treats missing fields as the zero value.

type IndexEntryCatalog added in v1.1.0 

type IndexEntryCatalog struct {
	TargetID    string   `msgpack:"targetId"`
	TargetScope string   `msgpack:"targetScope"`
	IndexKeys   [][]byte `msgpack:"indexKeys"`
	Deindexed   bool     `msgpack:"deindexed,omitempty"`
}

IndexEntryCatalog tracks the exact secondary-index Badger keys written for an entity. The deindex cleanup job uses this to write tombstones without scanning the full index keyspace.

type IndexStats 

type IndexStats struct {
	Name         string   `json:"name"`
	Type         string   `json:"type"`
	Label        string   `json:"label"`
	Property     string   `json:"property,omitempty"`
	Properties   []string `json:"properties,omitempty"`
	TotalEntries int64    `json:"totalEntries"`
	UniqueValues int64    `json:"uniqueValues"`
	Selectivity  float64  `json:"selectivity"` // uniqueValues / totalEntries
}

IndexStats represents statistics about an index.

type LabelConfig 

type LabelConfig struct {
	Label         string        `json:"label"`
	MaxEdges      int           `json:"max_edges"`      // 0 = unlimited
	MinConfidence float64       `json:"min_confidence"` // Override default threshold
	Cooldown      time.Duration `json:"cooldown"`       // Override default cooldown
	Disabled      bool          `json:"disabled"`       // Completely disable this label
}

LabelConfig defines per-label edge limits.

type LabelIndexEngine 

type LabelIndexEngine interface {
	// HasLabelBatch returns a map of node IDs that have the given label.
	// Implementations should treat labels case-insensitively (Neo4j compatible).
	// Missing nodes are treated as not having the label.
	HasLabelBatch(ids []NodeID, label string) (map[NodeID]bool, error)
}

LabelIndexEngine is an optional interface implemented by engines that can answer label-membership queries via an index, without decoding full nodes.

This is used by performance-sensitive query executors that need to enforce `(n:Label)` semantics while avoiding per-node GetNode/BatchGetNodes overhead.

type LabelNodeIDLookupEngine 

type LabelNodeIDLookupEngine interface {
	ForEachNodeIDByLabel(label string, visit func(NodeID) bool) error
}

LabelNodeIDLookupEngine is an optional interface for engines that can return node IDs for a label without decoding full nodes.

Implementations must treat labels case-insensitively (Neo4j compatible). The visit function should return true to continue iteration, false to stop.

type MVCCAppendEngine 

type MVCCAppendEngine interface {
	AppendNodeVersion(node *Node, version MVCCVersion) error
	AppendNodeTombstone(id NodeID, version MVCCVersion) error
	AppendEdgeVersion(edge *Edge, version MVCCVersion) error
	AppendEdgeTombstone(id EdgeID, version MVCCVersion) error
	UpdateNodeCurrentHead(id NodeID, version MVCCVersion, tombstoned bool) error
	UpdateEdgeCurrentHead(id EdgeID, version MVCCVersion, tombstoned bool) error
}

MVCCAppendEngine is an optional extension interface for immutable MVCC writes.

type MVCCEnumerationEngine 

type MVCCEnumerationEngine interface {
	BatchGetNodesLatestVisible(ids []NodeID) (map[NodeID]*Node, error)
	IterateLatestVisibleNodes(yield func(*Node) error) error
	IterateLatestVisibleEdges(yield func(*Edge) error) error
}

MVCCEnumerationEngine is an optional extension interface for latest-visible iteration.

type MVCCHead 

type MVCCHead struct {
	Version      MVCCVersion
	Tombstoned   bool
	FloorVersion MVCCVersion
}

MVCCHead stores the current persisted head for a logical record.

type MVCCHeadEngine 

type MVCCHeadEngine interface {
	GetNodeCurrentHead(id NodeID) (MVCCHead, error)
	GetEdgeCurrentHead(id EdgeID) (MVCCHead, error)
}

MVCCHeadEngine is an optional extension interface for persisted head lookups.

type MVCCIndexedVisibilityEngine 

type MVCCIndexedVisibilityEngine interface {
	GetNodesByLabelVisibleAt(label string, version MVCCVersion) ([]*Node, error)
	GetEdgesByTypeVisibleAt(edgeType string, version MVCCVersion) ([]*Edge, error)
	GetEdgesBetweenVisibleAt(startID, endID NodeID, version MVCCVersion) ([]*Edge, error)
}

MVCCIndexedVisibilityEngine is an optional extension interface for snapshot-visible graph queries that resolve label, type, and topology against MVCC history instead of only the current materialized indexes.

type MVCCLatestEffectiveEngine 

type MVCCLatestEffectiveEngine interface {
	GetNodeLatestEffective(id NodeID) (*Node, error)
	GetEdgeLatestEffective(id EdgeID) (*Edge, error)
}

MVCCLatestEffectiveEngine is an optional extension interface for wrapper-level latest reads that merge pending, in-flight, and persisted state.

type MVCCLifecycleController 

type MVCCLifecycleController interface {
	MVCCLifecycleEngine
	AcquireSnapshotReader(info SnapshotReaderInfo) (func(), error)
	EvaluateSnapshotReader(info SnapshotReaderInfo) (graceful bool, hard bool)
	RunPruneNow(ctx context.Context, opts MVCCPruneOptions) (int64, error)
	StartLifecycle(ctx context.Context)
	StopLifecycle()
	IsLifecycleEnabled() bool
	IsLifecycleRunning() bool
	ReaderRegistry() SnapshotReaderRegistry
}

MVCCLifecycleController is the storage-facing control interface used by engines. A concrete implementation can live outside the storage package and be injected.

type MVCCLifecycleDebtEngine 

type MVCCLifecycleDebtEngine interface {
	TopLifecycleDebtKeys(limit int) []MVCCLifecycleDebtKey
}

MVCCLifecycleDebtEngine is an optional extension for inspecting the highest-debt logical keys.

type MVCCLifecycleDebtKey 

type MVCCLifecycleDebtKey struct {
	LogicalKey       string `json:"logical_key"`
	Namespace        string `json:"namespace,omitempty"`
	DebtBytes        int64  `json:"debt_bytes"`
	TombstoneDepth   int    `json:"tombstone_depth"`
	FloorLagVersions int    `json:"floor_lag_versions"`
	VersionsToDelete int    `json:"versions_to_delete"`
}

MVCCLifecycleDebtKey describes one logical key contributing lifecycle debt.

type MVCCLifecycleEngine 

type MVCCLifecycleEngine interface {
	RegisterSnapshotReader(info SnapshotReaderInfo) func()
	LifecycleStatus() map[string]interface{}
	TriggerPruneNow(ctx context.Context) error
	PauseLifecycle()
	ResumeLifecycle()
}

MVCCLifecycleEngine is an optional extension interface for lifecycle management.

type MVCCLifecycleScheduleEngine 

type MVCCLifecycleScheduleEngine interface {
	SetLifecycleSchedule(interval time.Duration) error
}

MVCCLifecycleScheduleEngine is an optional extension for runtime lifecycle cadence control.

type MVCCMaintenanceEngine 

type MVCCMaintenanceEngine interface {
	RebuildMVCCHeads(ctx context.Context) error
	PruneMVCCVersions(ctx context.Context, opts MVCCPruneOptions) (int64, error)
}

MVCCMaintenanceEngine is an optional extension interface for rebuild and prune operations.

type MVCCPruneOptions 

type MVCCPruneOptions struct {
	MaxVersionsPerKey int
	MinRetentionAge   time.Duration
}

MVCCPruneOptions controls pruning of older MVCC versions. Zero values inherit the engine's configured RetentionPolicy.

type MVCCReadMode 

type MVCCReadMode string

MVCCReadMode selects latest-visible versus snapshot-visible reads. 

const (
	MVCCReadLatest   MVCCReadMode = "latest"
	MVCCReadSnapshot MVCCReadMode = "snapshot"
)

type MVCCReadSelector 

type MVCCReadSelector struct {
	Mode    MVCCReadMode
	Version MVCCVersion
}

MVCCReadSelector describes which committed version a read should resolve.

type MVCCVersion 

type MVCCVersion struct {
	CommitTimestamp time.Time
	CommitSequence  uint64
}

MVCCVersion identifies one committed storage version.

Versions are ordered first by committed timestamp, then by a monotonic sequence allocated at commit time to break same-timestamp ties.

func (MVCCVersion) Compare 

func (v MVCCVersion) Compare(other MVCCVersion) int

Compare returns -1, 0, or 1 using MVCC ordering semantics.

func (MVCCVersion) IsZero 

func (v MVCCVersion) IsZero() bool

IsZero reports whether the version is uninitialized.

func (MVCCVersion) String 

func (v MVCCVersion) String() string

String renders a stable debug form for logs and diagnostics.

type MVCCVisibilityEngine 

type MVCCVisibilityEngine interface {
	GetNodeLatestVisible(id NodeID) (*Node, error)
	GetNodeVisibleAt(id NodeID, version MVCCVersion) (*Node, error)
	GetEdgeLatestVisible(id EdgeID) (*Edge, error)
	GetEdgeVisibleAt(id EdgeID, version MVCCVersion) (*Edge, error)
}

MVCCVisibilityEngine is an optional extension interface for latest and snapshot-visible node and edge reads.

type MemoryEngine 

type MemoryEngine struct {
	*BadgerEngine
}

MemoryEngine is a thread-safe in-memory graph storage implementation. It wraps BadgerDB's in-memory mode for testing purposes.

Use Cases:

  • Unit testing (no disk I/O, fast cleanup)
  • Loading Neo4j exports into memory for analysis
  • Small datasets that fit entirely in RAM
  • Development and prototyping

Implementation Note: 

MemoryEngine is a thin wrapper around BadgerEngine with InMemory=true.
This ensures tests use the exact same code path as production.

func NewMemoryEngine 

func NewMemoryEngine() *MemoryEngine

NewMemoryEngine creates a new in-memory storage engine for testing. It uses BadgerDB's in-memory mode internally.

Example: 

engine := storage.NewMemoryEngine()
defer engine.Close()

Note: For tests, prefer storage.NewTestEngine(t) which handles cleanup.

func NewMemoryEngineWithMVCCHistory added in v1.1.0 

func NewMemoryEngineWithMVCCHistory() *MemoryEngine

NewMemoryEngineWithMVCCHistory creates an in-memory engine that retains historical MVCC versions. Use this for tests and callers that exercise the multi-version surface (temporal.asOf, time-travel reads, prior-version snapshots). The default NewMemoryEngine runs head-only, matching the production default.

func RecoverFromWAL 

func RecoverFromWAL(walDir, snapshotPath string) (*MemoryEngine, error)

RecoverFromWAL recovers database state from a snapshot and WAL. Returns a new MemoryEngine with the recovered state.

Uses a discard *slog.Logger for diagnostics. Callers wanting structured recovery visibility should use RecoverFromWALWithLogger.

func RecoverFromWALWithLogger added in v1.1.0 

func RecoverFromWALWithLogger(walDir, snapshotPath string, logger *slog.Logger) (*MemoryEngine, error)

RecoverFromWALWithLogger is the slog-aware variant of RecoverFromWAL. D-07: pass a child logger tagged subsystem=wal_recovery so completion- with-errors records land in the structured stream. Operator-actionable fields: failed, errors, summary; per-error attributes seq, operation, error.

func (*MemoryEngine) BeginTransaction 

func (m *MemoryEngine) BeginTransaction() (*BadgerTransaction, error)

BeginTransaction starts a new transaction. Returns *BadgerTransaction for compatibility with the executor.

func (*MemoryEngine) DeleteByPrefix 

func (m *MemoryEngine) DeleteByPrefix(prefix string) (nodesDeleted int64, edgesDeleted int64, err error)

DeleteByPrefix delegates to the underlying BadgerEngine.

type NamespaceLister 

type NamespaceLister interface {
	ListNamespaces() []string
}

NamespaceLister is an optional extension interface that reports the known database namespaces stored in an engine.

Returned values are unqualified namespace names (e.g., "nornic", "db2"), not ID prefixes (e.g., "nornic:").

type NamespaceSchemaProvider 

type NamespaceSchemaProvider interface {
	GetSchemaForNamespace(namespace string) *SchemaManager
}

NamespaceSchemaProvider is an optional extension interface that provides per-namespace schema.

This enables multi-database deployments to maintain isolated constraints/indexes per database, matching Neo4j’s per-database schema model.

type NamespaceTemporalCurrentNodeProvider 

type NamespaceTemporalCurrentNodeProvider interface {
	IsCurrentTemporalNodeInNamespace(namespace string, node *Node, asOf time.Time) (bool, error)
}

NamespaceTemporalCurrentNodeProvider is an optional extension interface that evaluates current/live temporal versions within a namespace.

type NamespaceTemporalLookupProvider 

type NamespaceTemporalLookupProvider interface {
	GetTemporalNodeAsOfInNamespace(namespace, label, keyProp string, keyValue interface{}, validFromProp, validToProp string, asOf time.Time) (*Node, error)
}

NamespaceTemporalLookupProvider is an optional extension interface that provides efficient temporal lookups scoped to a storage namespace.

type NamespacedEngine 

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

NamespacedEngine wraps a storage engine with database namespace isolation. All node and edge IDs are automatically prefixed with the namespace.

This provides logical database separation within a single physical storage:

  • Keys are prefixed: "tenant_a:node:123" instead of "node:123"
  • Queries only see data in the current namespace
  • DROP DATABASE = delete all keys with prefix

Thread-safe: delegates to underlying engine's thread safety.

func NewNamespacedEngine 

func NewNamespacedEngine(inner Engine, namespace string) *NamespacedEngine

NewNamespacedEngine creates a namespaced view of the storage engine.

Parameters:

  • inner: The underlying storage engine (shared across all namespaces)
  • namespace: The database name (e.g., "tenant_a", "nornic")

The namespace is used as a key prefix for all operations.

func (*NamespacedEngine) AddToPendingEmbeddings 

func (n *NamespacedEngine) AddToPendingEmbeddings(nodeID NodeID)

AddToPendingEmbeddings adds a node back to the pending embeddings index (e.g. after a failed embed so it can be retried).

func (*NamespacedEngine) AllEdges 

func (n *NamespacedEngine) AllEdges() ([]*Edge, error)

func (*NamespacedEngine) AllNodes 

func (n *NamespacedEngine) AllNodes() ([]*Node, error)

func (*NamespacedEngine) BatchGetNodes 

func (n *NamespacedEngine) BatchGetNodes(ids []NodeID) (map[NodeID]*Node, error)

func (*NamespacedEngine) BulkCreateEdges 

func (n *NamespacedEngine) BulkCreateEdges(edges []*Edge) error

func (*NamespacedEngine) BulkCreateNodes 

func (n *NamespacedEngine) BulkCreateNodes(nodes []*Node) error

func (*NamespacedEngine) BulkDeleteEdges 

func (n *NamespacedEngine) BulkDeleteEdges(ids []EdgeID) error

func (*NamespacedEngine) BulkDeleteNodes 

func (n *NamespacedEngine) BulkDeleteNodes(ids []NodeID) error

func (*NamespacedEngine) Close 

func (n *NamespacedEngine) Close() error

func (*NamespacedEngine) CreateEdge 

func (n *NamespacedEngine) CreateEdge(edge *Edge) error

func (*NamespacedEngine) CreateNode 

func (n *NamespacedEngine) CreateNode(node *Node) (NodeID, error)

func (*NamespacedEngine) DeleteByPrefix 

func (n *NamespacedEngine) DeleteByPrefix(prefix string) (nodesDeleted int64, edgesDeleted int64, err error)

DeleteByPrefix is not supported for NamespacedEngine. Use the underlying engine's DeleteByPrefix with the namespace prefix instead.

func (*NamespacedEngine) DeleteEdge 

func (n *NamespacedEngine) DeleteEdge(id EdgeID) error

func (*NamespacedEngine) DeleteNode 

func (n *NamespacedEngine) DeleteNode(id NodeID) error

func (*NamespacedEngine) EdgeCount 

func (n *NamespacedEngine) EdgeCount() (int64, error)

func (*NamespacedEngine) FindNodeNeedingEmbedding 

func (n *NamespacedEngine) FindNodeNeedingEmbedding() *Node

FindNodeNeedingEmbedding finds a node that needs embedding, but only from this namespace. It only looks for nodes with the current namespace prefix - all IDs must be prefixed.

func (*NamespacedEngine) ForEachNodeIDByLabel 

func (n *NamespacedEngine) ForEachNodeIDByLabel(label string, visit func(NodeID) bool) error

ForEachNodeIDByLabel streams node IDs for a label, filtered to the namespace. Stops early when visit returns false.

func (*NamespacedEngine) GetAdjacentEdges added in v1.1.2 

func (n *NamespacedEngine) GetAdjacentEdges(nodeID NodeID) ([]*Edge, []*Edge, error)

GetAdjacentEdges fetches both directions through a single inner call when the inner engine supports the AdjacentEdgesEngine capability. ID translation and namespace filtering mirror the per-direction methods.

func (*NamespacedEngine) GetAllNodes 

func (n *NamespacedEngine) GetAllNodes() []*Node

func (*NamespacedEngine) GetEdge 

func (n *NamespacedEngine) GetEdge(id EdgeID) (*Edge, error)

func (*NamespacedEngine) GetEdgeBetween 

func (n *NamespacedEngine) GetEdgeBetween(startID, endID NodeID, edgeType string) *Edge

func (*NamespacedEngine) GetEdgeCurrentHead 

func (n *NamespacedEngine) GetEdgeCurrentHead(id EdgeID) (MVCCHead, error)

GetEdgeCurrentHead resolves edge head metadata within the namespace.

func (*NamespacedEngine) GetEdgeLatestVisible 

func (n *NamespacedEngine) GetEdgeLatestVisible(id EdgeID) (*Edge, error)

GetEdgeLatestVisible resolves the latest visible edge within the namespace.

func (*NamespacedEngine) GetEdgeVisibleAt 

func (n *NamespacedEngine) GetEdgeVisibleAt(id EdgeID, version MVCCVersion) (*Edge, error)

GetEdgeVisibleAt resolves a snapshot-visible edge within the namespace.

func (*NamespacedEngine) GetEdgesBetween 

func (n *NamespacedEngine) GetEdgesBetween(startID, endID NodeID) ([]*Edge, error)

func (*NamespacedEngine) GetEdgesBetweenVisibleAt 

func (n *NamespacedEngine) GetEdgesBetweenVisibleAt(startID, endID NodeID, version MVCCVersion) ([]*Edge, error)

GetEdgesBetweenVisibleAt resolves snapshot-visible topology queries within the namespace.

func (*NamespacedEngine) GetEdgesByType 

func (n *NamespacedEngine) GetEdgesByType(edgeType string) ([]*Edge, error)

func (*NamespacedEngine) GetEdgesByTypeVisibleAt 

func (n *NamespacedEngine) GetEdgesByTypeVisibleAt(edgeType string, version MVCCVersion) ([]*Edge, error)

GetEdgesByTypeVisibleAt resolves snapshot-visible edge-type queries within the namespace.

func (*NamespacedEngine) GetFirstNodeByLabel 

func (n *NamespacedEngine) GetFirstNodeByLabel(label string) (*Node, error)

func (*NamespacedEngine) GetInDegree 

func (n *NamespacedEngine) GetInDegree(nodeID NodeID) int

func (*NamespacedEngine) GetIncomingEdges 

func (n *NamespacedEngine) GetIncomingEdges(nodeID NodeID) ([]*Edge, error)

func (*NamespacedEngine) GetInnerEngine 

func (n *NamespacedEngine) GetInnerEngine() Engine

GetInnerEngine returns the underlying storage engine (unwraps the namespace). This is used by DatabaseManager to create NamespacedEngines for other databases.

func (*NamespacedEngine) GetNode 

func (n *NamespacedEngine) GetNode(id NodeID) (*Node, error)

func (*NamespacedEngine) GetNodeCurrentHead 

func (n *NamespacedEngine) GetNodeCurrentHead(id NodeID) (MVCCHead, error)

GetNodeCurrentHead resolves node head metadata within the namespace.

func (*NamespacedEngine) GetNodeLatestVisible 

func (n *NamespacedEngine) GetNodeLatestVisible(id NodeID) (*Node, error)

GetNodeLatestVisible resolves the latest visible node within the namespace.

func (*NamespacedEngine) GetNodeVisibleAt 

func (n *NamespacedEngine) GetNodeVisibleAt(id NodeID, version MVCCVersion) (*Node, error)

GetNodeVisibleAt resolves a snapshot-visible node within the namespace.

func (*NamespacedEngine) GetNodesByLabel 

func (n *NamespacedEngine) GetNodesByLabel(label string) ([]*Node, error)

func (*NamespacedEngine) GetNodesByLabelVisibleAt 

func (n *NamespacedEngine) GetNodesByLabelVisibleAt(label string, version MVCCVersion) ([]*Node, error)

GetNodesByLabelVisibleAt resolves snapshot-visible label queries within the namespace.

func (*NamespacedEngine) GetOutDegree 

func (n *NamespacedEngine) GetOutDegree(nodeID NodeID) int

func (*NamespacedEngine) GetOutgoingEdges 

func (n *NamespacedEngine) GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)

func (*NamespacedEngine) GetSchema 

func (n *NamespacedEngine) GetSchema() *SchemaManager

func (*NamespacedEngine) GetTemporalNodeAsOf 

func (n *NamespacedEngine) GetTemporalNodeAsOf(label, keyProp string, keyValue interface{}, validFromProp, validToProp string, asOf time.Time) (*Node, error)

GetTemporalNodeAsOf performs an efficient temporal lookup when the inner engine supports it.

func (*NamespacedEngine) IsCurrentTemporalNode 

func (n *NamespacedEngine) IsCurrentTemporalNode(node *Node, asOf time.Time) (bool, error)

IsCurrentTemporalNode reports whether node is the current/live temporal version.

func (*NamespacedEngine) LastWriteTime 

func (n *NamespacedEngine) LastWriteTime() time.Time

LastWriteTime returns the last known write time from the underlying engine, if available.

func (*NamespacedEngine) LifecycleStatus 

func (n *NamespacedEngine) LifecycleStatus() map[string]interface{}

LifecycleStatus delegates lifecycle status when supported.

func (*NamespacedEngine) MarkNodeEmbedded 

func (n *NamespacedEngine) MarkNodeEmbedded(nodeID NodeID)

MarkNodeEmbedded marks a node as embedded (removes from pending index). The node ID should be unprefixed (without namespace).

func (*NamespacedEngine) Namespace 

func (n *NamespacedEngine) Namespace() string

Namespace returns the current database namespace.

func (*NamespacedEngine) NodeCount 

func (n *NamespacedEngine) NodeCount() (int64, error)

func (*NamespacedEngine) PauseLifecycle 

func (n *NamespacedEngine) PauseLifecycle()

PauseLifecycle delegates lifecycle pause when supported.

func (*NamespacedEngine) RecordMaterializedAccess added in v1.1.0 

func (n *NamespacedEngine) RecordMaterializedAccess(entityID string)

RecordMaterializedAccess records a result-materialization access against the fully qualified entity ID in the underlying engine.

func (*NamespacedEngine) RefreshPendingEmbeddingsIndex 

func (n *NamespacedEngine) RefreshPendingEmbeddingsIndex() int

RefreshPendingEmbeddingsIndex refreshes the pending embeddings index, but only for nodes in this namespace. Also cleans up stale entries from other namespaces in the underlying index.

func (*NamespacedEngine) RegisterSnapshotReader 

func (n *NamespacedEngine) RegisterSnapshotReader(info SnapshotReaderInfo) func()

RegisterSnapshotReader registers a reader scoped to the namespace when supported.

func (*NamespacedEngine) ResumeLifecycle 

func (n *NamespacedEngine) ResumeLifecycle()

ResumeLifecycle delegates lifecycle resume when supported.

func (*NamespacedEngine) SetLifecycleSchedule 

func (n *NamespacedEngine) SetLifecycleSchedule(interval time.Duration) error

SetLifecycleSchedule delegates lifecycle cadence updates when supported.

func (*NamespacedEngine) StreamEdges 

func (n *NamespacedEngine) StreamEdges(ctx context.Context, fn func(edge *Edge) error) error

StreamEdges streams edges in the namespace.

func (*NamespacedEngine) StreamNodeChunks 

func (n *NamespacedEngine) StreamNodeChunks(ctx context.Context, chunkSize int, fn func(nodes []*Node) error) error

StreamNodeChunks streams nodes in chunks.

func (*NamespacedEngine) StreamNodes 

func (n *NamespacedEngine) StreamNodes(ctx context.Context, fn func(node *Node) error) error

StreamNodes streams nodes in the namespace.

func (*NamespacedEngine) TopLifecycleDebtKeys 

func (n *NamespacedEngine) TopLifecycleDebtKeys(limit int) []MVCCLifecycleDebtKey

TopLifecycleDebtKeys delegates lifecycle debt inspection when supported.

func (*NamespacedEngine) TriggerPruneNow 

func (n *NamespacedEngine) TriggerPruneNow(ctx context.Context) error

TriggerPruneNow delegates lifecycle prune-now when supported.

func (*NamespacedEngine) UpdateEdge 

func (n *NamespacedEngine) UpdateEdge(edge *Edge) error

func (*NamespacedEngine) UpdateNode 

func (n *NamespacedEngine) UpdateNode(node *Node) error

type Neo4jExport 

type Neo4jExport struct {
	Nodes         []Neo4jNode         `json:"nodes"`
	Relationships []Neo4jRelationship `json:"relationships"`
}

Neo4jExport represents the Neo4j JSON export format. This is compatible with `neo4j-admin database dump` JSON output.

func ToNeo4jExport 

func ToNeo4jExport(nodes []*Node, edges []*Edge) *Neo4jExport

ToNeo4jExport converts NornicDB nodes and edges to Neo4j JSON export format.

This function prepares data for export that can be imported into Neo4j using neo4j-admin or APOC procedures. NornicDB-specific fields (decay score, embeddings, access counts) are stored with "_" prefix to mark them as system properties.

The output is compatible with:

  • `neo4j-admin database import`
  • `CALL apoc.import.json()`
  • Standard Neo4j JSON format

Example: 

// Get all data
nodes, _ := engine.GetNodesByLabel("") // All nodes
edges, _ := engine.AllEdges()

// Convert to Neo4j format
export := storage.ToNeo4jExport(nodes, edges)

// Save as JSON
data, _ := json.MarshalIndent(export, "", "  ")
err := os.WriteFile("neo4j-export.json", data, 0644)

// Import into Neo4j
// $ neo4j-admin database import --nodes=neo4j-export.json full
// Or in Cypher:
// CALL apoc.import.json("file:///neo4j-export.json")

NornicDB extensions are preserved as properties: 

_decayScore, _lastAccessed, _accessCount, _confidence, _autoGenerated

type Neo4jNode 

type Neo4jNode struct {
	ID         string         `json:"id"`
	Labels     []string       `json:"labels"`
	Properties map[string]any `json:"properties"`
}

Neo4jNode is the Neo4j JSON export format for nodes.

type Neo4jNodeRef 

type Neo4jNodeRef struct {
	ID     string   `json:"id"`
	Labels []string `json:"labels,omitempty"`
}

Neo4jNodeRef is a reference to a node in Neo4j relationship format.

type Neo4jRelationship 

type Neo4jRelationship struct {
	ID         string         `json:"id"`
	Type       string         `json:"type"`
	Properties map[string]any `json:"properties"`

	// Flat format (neo4j-admin dump)
	StartNode string `json:"startNode,omitempty"`
	EndNode   string `json:"endNode,omitempty"`

	// APOC format (apoc.export.json)
	Start Neo4jNodeRef `json:"start,omitempty"`
	End   Neo4jNodeRef `json:"end,omitempty"`
}

Neo4jRelationship is the Neo4j JSON export format for relationships. Supports both flat format (startNode/endNode strings) and APOC format (start/end objects).

func (*Neo4jRelationship) GetEndID 

func (r *Neo4jRelationship) GetEndID() string

GetEndID returns the end node ID regardless of format.

func (*Neo4jRelationship) GetStartID 

func (r *Neo4jRelationship) GetStartID() string

GetStartID returns the start node ID supporting both Neo4j export formats.

Neo4j exports can use two formats:

  1. Flat format: startNode/endNode as strings (neo4j-admin dump)
  2. APOC format: start/end as objects (apoc.export.json)

This method abstracts the difference, always returning the start node ID.

Example: 

// Flat format
rel := &Neo4jRelationship{
	StartNode: "user-123",
}
// rel.GetStartID() returns "user-123"

// APOC format
rel = &Neo4jRelationship{
	Start: Neo4jNodeRef{ID: "user-456"},
}
// rel.GetStartID() returns "user-456"

type Node 

type Node struct {
	ID         NodeID         `json:"id"`
	Labels     []string       `json:"labels"`
	Properties map[string]any `json:"properties"`

	// NornicDB extensions
	CreatedAt       time.Time            `json:"-"`
	UpdatedAt       time.Time            `json:"-"`
	NamedEmbeddings map[string][]float32 `json:"-"` // Named vector embeddings (e.g., "title", "content", "default")
	ChunkEmbeddings [][]float32          `json:"-"` // Chunked embeddings for long documents (legacy, migration support)

	// Embedding metadata (separate from user Properties to avoid namespace pollution)
	// Keys: embedding_model, embedding_dimensions, has_embedding, embedded_at, has_chunks, chunk_count
	EmbedMeta map[string]any `json:"-"`

	// Internal storage flags (not exposed to users, used during encode/decode)
	EmbeddingsStoredSeparately bool `json:"-"` // True when embeddings are stored in separate keys (large node optimization)
	VisibilitySuppressed       bool `json:"-"` // Set by deindex cleanup when decay score falls below visibility threshold
}

Node represents a graph node (vertex) in the labeled property graph.

Nodes follow the Neo4j data model with NornicDB-specific extensions for memory decay, semantic search, and access tracking. Nodes are the fundamental entities in the graph and can represent people, documents, concepts, or any other entity in your domain.

Core Neo4j Fields:

  • ID: Unique identifier (must be unique across all nodes)
  • Labels: Type tags like ["Person", "User"] (Neo4j :Person:User)
  • Properties: Key-value data (any JSON-serializable types) See docs/user-guides/property-data-types.md for complete type reference

NornicDB Extensions (not exported to Neo4j):

  • CreatedAt: When the node was first created
  • UpdatedAt: Last modification timestamp
  • NamedEmbeddings: Named vector embeddings (e.g., "title", "content", "default")
  • ChunkEmbeddings: Chunked embeddings for long documents (legacy, migration support)

Decay scoring is handled by the knowledge-layer scoring system (pkg/knowledgepolicy). Scores are computed at query time from AccessMeta and retention bindings, not stored on the node.

Example 1 - Basic User Node: 

node := &storage.Node{
	ID:     storage.NodeID("user-alice"),
	Labels: []string{"Person", "User"},
	Properties: map[string]any{
		"name":     "Alice Johnson",
		"age":      30,
		"email":    "alice@example.com",
		"verified": true,
	},
	CreatedAt: time.Now(),
}
engine.CreateNode(node)

Example 2 - Document Node with Metadata: 

doc := &storage.Node{
	ID:     storage.NodeID("doc-readme"),
	Labels: []string{"Document", "Markdown"},
	Properties: map[string]any{
		"title":    "README.md",
		"content":  "# Welcome to...",
		"path":     "./README.md",
		"size":     4096,
		"language": "markdown",
	},
	CreatedAt: time.Now(),
	Embedding: generateEmbedding("# Welcome to..."), // For semantic search
}

Example 3 - Concept Node for Knowledge Graph: 

concept := &storage.Node{
	ID:     storage.NodeID("concept-database"),
	Labels: []string{"Concept", "Technology"},
	Properties: map[string]any{
		"name":        "Database Systems",
		"definition":  "Systems for storing and retrieving data",
		"category":    "Software",
		"importance":  "high",
	},
	CreatedAt: time.Now(),
}

ELI12:

Think of a Node like a character card in a trading card game:

  • ID: The card's unique number (no two cards have the same)
  • Labels: The types on the card (["Hero", "Warrior"])
  • Properties: Stats on the card (name: "Alice", strength: 10, health: 100)

NornicDB adds extra info:

  • Embedding: A secret code that helps find similar cards
  • Decay scoring: How "fresh" the card is — computed automatically by knowledge-layer policies

Just like trading cards can be rare or common, frequently used or forgotten, Nodes track their usage and importance in the graph!

Neo4j Compatibility:

  • Labels map to Neo4j labels (e.g., :Person:User)
  • Properties map to Neo4j properties
  • ID must be unique across all nodes
  • Extensions stored with "_" prefix in Neo4j exports

Thread Safety: 

Node structs are NOT thread-safe. The storage engine handles concurrency.

func CopyNode 

func CopyNode(node *Node) *Node

CopyNode is the exported version of copyNode for use by other packages.

func FindNodeNeedingEmbedding 

func FindNodeNeedingEmbedding(engine Engine) *Node

FindNodeNeedingEmbedding finds a single node that needs embedding. This is more efficient than AllNodes() as it stops after finding one.

func (*Node) ExtractInternalProperties 

func (n *Node) ExtractInternalProperties()

ExtractInternalProperties extracts NornicDB-specific fields from properties.

func (*Node) GetDefaultEmbedding 

func (n *Node) GetDefaultEmbedding() []float32

GetDefaultEmbedding returns the default embedding for a node, implementing migration behavior from ChunkEmbeddings to NamedEmbeddings.

Migration behavior (temporary):

  • If NamedEmbeddings["default"] exists, return it
  • Otherwise, if ChunkEmbeddings has at least one vector, treat ChunkEmbeddings[0] as "default"
  • Otherwise, return nil

This allows backward compatibility with existing nodes that only have ChunkEmbeddings while transitioning to the new NamedEmbeddings model.

Example: 

node := &storage.Node{
	ID: storage.NodeID("doc-1"),
	NamedEmbeddings: map[string][]float32{
		"default": []float32{0.1, 0.2, 0.3},
	},
}
emb := node.GetDefaultEmbedding() // Returns []float32{0.1, 0.2, 0.3}

// Legacy node with ChunkEmbeddings
legacyNode := &storage.Node{
	ID: storage.NodeID("doc-2"),
	ChunkEmbeddings: [][]float32{{0.4, 0.5, 0.6}},
}
emb = legacyNode.GetDefaultEmbedding() // Returns []float32{0.4, 0.5, 0.6}

func (*Node) MarshalNeo4jJSON 

func (n *Node) MarshalNeo4jJSON() ([]byte, error)

MarshalNeo4jJSON serializes to Neo4j-compatible JSON.

type NodeConfig 

type NodeConfig struct {
	// Node identification
	NodeID    string    `json:"node_id"`
	CreatedAt time.Time `json:"created_at"`
	UpdatedAt time.Time `json:"updated_at"`

	// Pin/Deny lists
	PinList  []string `json:"pin_list"`  // Target node IDs that never decay
	DenyList []string `json:"deny_list"` // Target node IDs to never connect to

	// Edge caps
	MaxOutEdges   int `json:"max_out_edges"`   // Maximum outgoing edges (0 = unlimited)
	MaxInEdges    int `json:"max_in_edges"`    // Maximum incoming edges (0 = unlimited)
	MaxTotalEdges int `json:"max_total_edges"` // Maximum total edges (0 = unlimited)

	// Current edge counts (for cap enforcement)
	CurrentOutEdges   int `json:"current_out_edges"`
	CurrentInEdges    int `json:"current_in_edges"`
	CurrentTotalEdges int `json:"current_total_edges"`

	// Per-label configuration
	LabelConfigs map[string]LabelConfig `json:"label_configs"`

	// Trust level
	TrustLevel TrustLevel `json:"trust_level"`

	// Global overrides
	MinConfidence float64       `json:"min_confidence"` // Override default threshold (0 = use default)
	Cooldown      time.Duration `json:"cooldown"`       // Override default cooldown (0 = use default)
	Disabled      bool          `json:"disabled"`       // Completely disable edge creation to/from this node

	// Metadata
	Metadata map[string]interface{} `json:"metadata,omitempty"`
}

NodeConfig stores per-node edge materialization settings.

func NewNodeConfig 

func NewNodeConfig(nodeID string) *NodeConfig

NewNodeConfig creates a new node config with defaults.

func (*NodeConfig) AddToDeny 

func (c *NodeConfig) AddToDeny(targetID string)

AddToDeny adds a target to the deny list.

func (*NodeConfig) AddToPin 

func (c *NodeConfig) AddToPin(targetID string)

AddToPin adds a target to the pin list.

func (*NodeConfig) CanAddEdge 

func (c *NodeConfig) CanAddEdge(isOutgoing bool) bool

CanAddEdge returns true if another edge can be added (checks all caps).

func (*NodeConfig) CanAddInEdge 

func (c *NodeConfig) CanAddInEdge() bool

CanAddInEdge returns true if another incoming edge can be added.

func (*NodeConfig) CanAddOutEdge 

func (c *NodeConfig) CanAddOutEdge() bool

CanAddOutEdge returns true if another outgoing edge can be added.

func (*NodeConfig) DecrementEdgeCount 

func (c *NodeConfig) DecrementEdgeCount(isOutgoing bool)

DecrementEdgeCount decrements the edge count.

func (*NodeConfig) GetEffectiveConfidence 

func (c *NodeConfig) GetEffectiveConfidence(label string, baseThreshold float64) float64

GetEffectiveConfidence returns the effective minimum confidence threshold. Considers trust level adjustment and per-label overrides.

func (*NodeConfig) GetLabelConfig 

func (c *NodeConfig) GetLabelConfig(label string) (LabelConfig, bool)

GetLabelConfig returns the config for a specific label.

func (*NodeConfig) IncrementEdgeCount 

func (c *NodeConfig) IncrementEdgeCount(isOutgoing bool)

IncrementEdgeCount increments the edge count.

func (*NodeConfig) IsDenied 

func (c *NodeConfig) IsDenied(targetID string) bool

IsDenied returns true if the target is in the deny list.

func (*NodeConfig) IsPinned 

func (c *NodeConfig) IsPinned(targetID string) bool

IsPinned returns true if the target is in the pin list.

func (*NodeConfig) RemoveFromDeny 

func (c *NodeConfig) RemoveFromDeny(targetID string) bool

RemoveFromDeny removes a target from the deny list.

func (*NodeConfig) RemoveFromPin 

func (c *NodeConfig) RemoveFromPin(targetID string) bool

RemoveFromPin removes a target from the pin list.

func (*NodeConfig) SetLabelConfig 

func (c *NodeConfig) SetLabelConfig(label string, cfg LabelConfig)

SetLabelConfig sets the config for a specific label.

type NodeConfigStats 

type NodeConfigStats struct {
	TotalConfigs int64
	TotalChecks  int64
	TotalBlocked int64
	BlockRate    float64
}

NodeConfigStats provides observability into per-node config state.

type NodeConfigStore 

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

NodeConfigStore manages per-node configurations. Thread-safe for concurrent access.

func GlobalNodeConfigStore 

func GlobalNodeConfigStore() *NodeConfigStore

GlobalNodeConfigStore returns the global node config store singleton.

func NewNodeConfigStore 

func NewNodeConfigStore() *NodeConfigStore

NewNodeConfigStore creates a new per-node config store.

func NewNodeConfigStoreWithOptions 

func NewNodeConfigStoreWithOptions(opts ...NodeConfigStoreOption) *NodeConfigStore

NewNodeConfigStoreWithOptions creates a store with functional options.

func (*NodeConfigStore) AddToNodeDenyList 

func (s *NodeConfigStore) AddToNodeDenyList(nodeID, targetID string)

AddToNodeDenyList adds a target to a node's deny list.

func (*NodeConfigStore) AddToNodePinList 

func (s *NodeConfigStore) AddToNodePinList(nodeID, targetID string)

AddToNodePinList adds a target to a node's pin list.

func (*NodeConfigStore) Clear 

func (s *NodeConfigStore) Clear()

Clear removes all node configs.

func (*NodeConfigStore) Delete 

func (s *NodeConfigStore) Delete(nodeID string) bool

Delete removes a node config.

func (*NodeConfigStore) DisableNode 

func (s *NodeConfigStore) DisableNode(nodeID string)

DisableNode disables all edge creation to/from a node.

func (*NodeConfigStore) EnableNode 

func (s *NodeConfigStore) EnableNode(nodeID string)

EnableNode enables edge creation to/from a node.

func (*NodeConfigStore) Export 

func (s *NodeConfigStore) Export() []*NodeConfig

Export returns all configs for backup/export.

func (*NodeConfigStore) Get 

func (s *NodeConfigStore) Get(nodeID string) *NodeConfig

Get returns the config for a node, or nil if none exists.

func (*NodeConfigStore) GetAllNodeIDs 

func (s *NodeConfigStore) GetAllNodeIDs() []string

GetAllNodeIDs returns all configured node IDs.

func (*NodeConfigStore) GetDeniedTargets 

func (s *NodeConfigStore) GetDeniedTargets(nodeID string) []string

GetDeniedTargets returns all denied targets for a node.

func (*NodeConfigStore) GetEffectiveConfidence 

func (s *NodeConfigStore) GetEffectiveConfidence(sourceID, targetID, label string, baseThreshold float64) float64

GetEffectiveConfidence returns the effective confidence threshold for an edge.

func (*NodeConfigStore) GetOrCreate 

func (s *NodeConfigStore) GetOrCreate(nodeID string) *NodeConfig

GetOrCreate returns the config for a node, creating one if it doesn't exist.

func (*NodeConfigStore) GetPinnedTargets 

func (s *NodeConfigStore) GetPinnedTargets(nodeID string) []string

GetPinnedTargets returns all pinned targets for a node.

func (*NodeConfigStore) Import 

func (s *NodeConfigStore) Import(configs []*NodeConfig) int

Import loads configs from backup/import. Existing configs are NOT cleared - use Clear() first if needed.

func (*NodeConfigStore) IsEdgeAllowed 

func (s *NodeConfigStore) IsEdgeAllowed(sourceID, targetID, label string) bool

IsEdgeAllowed checks if an edge from source to target is allowed. Considers: feature flag, disabled state, deny list, and edge caps.

func (*NodeConfigStore) IsEdgeAllowedWithReason 

func (s *NodeConfigStore) IsEdgeAllowedWithReason(sourceID, targetID, label string) (bool, string)

IsEdgeAllowedWithReason checks if an edge is allowed and returns the reason.

func (*NodeConfigStore) IsPinned 

func (s *NodeConfigStore) IsPinned(sourceID, targetID string) bool

IsPinned checks if an edge should never decay.

func (*NodeConfigStore) RecordEdgeCreation 

func (s *NodeConfigStore) RecordEdgeCreation(sourceID, targetID string)

RecordEdgeCreation updates edge counts after an edge is created.

func (*NodeConfigStore) RecordEdgeDeletion 

func (s *NodeConfigStore) RecordEdgeDeletion(sourceID, targetID string)

RecordEdgeDeletion updates edge counts after an edge is deleted.

func (*NodeConfigStore) Set 

func (s *NodeConfigStore) Set(cfg NodeConfig)

Set stores a node config.

func (*NodeConfigStore) SetNodeEdgeCaps 

func (s *NodeConfigStore) SetNodeEdgeCaps(nodeID string, maxOut, maxIn, maxTotal int)

SetNodeEdgeCaps sets edge caps for a node.

func (*NodeConfigStore) SetNodeTrustLevel 

func (s *NodeConfigStore) SetNodeTrustLevel(nodeID string, level TrustLevel)

SetNodeTrustLevel sets the trust level for a node.

func (*NodeConfigStore) Size 

func (s *NodeConfigStore) Size() int

Size returns the number of configured nodes.

func (*NodeConfigStore) Stats 

func (s *NodeConfigStore) Stats() NodeConfigStats

Stats returns current store statistics.

type NodeConfigStoreOption 

type NodeConfigStoreOption func(*NodeConfigStore)

NodeConfigStoreOption configures a NodeConfigStore.

type NodeDeleteCallback 

type NodeDeleteCallback func(nodeID NodeID)

NodeDeleteCallback is called when a node is successfully deleted from storage.

type NodeEventCallback 

type NodeEventCallback func(node *Node)

NodeEventCallback is called when storage operations complete successfully. This allows external services (like search indexes) to stay synchronized with storage.

type NodeID 

type NodeID string

NodeID is a strongly-typed unique identifier for graph nodes.

Using a custom type provides:

  • Type safety (can't accidentally use EdgeID where NodeID is expected)
  • Clear API semantics
  • Future extensibility (could add methods)

Example: 

id := storage.NodeID("user-123")
node, err := engine.GetNode(id)

func EnsureNodeIDDatabasePrefixForEngine added in v1.1.0 

func EnsureNodeIDDatabasePrefixForEngine(engine Engine, id NodeID) NodeID

EnsureNodeIDDatabasePrefixForEngine adds the active engine namespace to an unprefixed node ID. Schema backfills use this so cache entries match the storage IDs later seen by transaction commit validation.

func FirstNodeIDByLabel 

func FirstNodeIDByLabel(engine Engine, label string) (NodeID, error)

FirstNodeIDByLabel returns the first node ID for a label without decoding nodes when possible. Returns ErrNotFound if no node matches.

func NodeIDsByLabel 

func NodeIDsByLabel(engine Engine, label string, limit int) ([]NodeID, error)

NodeIDsByLabel returns up to limit node IDs that have the label. If limit <= 0, all matches are returned (may be expensive).

type NodeVisitor 

type NodeVisitor func(node *Node) error

NodeVisitor is a function called for each node during streaming.

type Operation 

type Operation struct {
	Type      OperationType
	Timestamp time.Time

	// For node operations
	NodeID  NodeID
	Node    *Node // New state (for create/update) or nil
	OldNode *Node // Old state (for update/delete rollback)
	// For delete operations that cascade (e.g., DeleteNode deletes edges).
	EdgesDeleted   int64
	DeletedEdgeIDs []EdgeID

	// For edge operations
	EdgeID  EdgeID
	Edge    *Edge // New state (for create/update) or nil
	OldEdge *Edge // Old state (for update/delete rollback)

	// FreshID is set on OpCreateNode / OpCreateEdge when the caller asserted
	// the ID is newly minted and cannot collide with any prior tombstoned
	// MVCC head (the same contract that lets CreateNode skip its existence
	// read). When true, the commit loop writes the MVCC head without the
	// load-existing-floor round-trip. Safe default is false — the commit
	// loop falls back to the read-before-write path, preserving snapshot
	// semantics for recreated user-supplied IDs.
	FreshID bool
}

Operation represents a single operation within a transaction. Used by BadgerTransaction to track operations for constraint validation.

type OperationType 

type OperationType string

OperationType represents the type of operation in a transaction. 

const (
	OpCreateNode      OperationType = "create_node"
	OpUpdateNode      OperationType = "update_node"
	OpDeleteNode      OperationType = "delete_node"
	OpCreateEdge      OperationType = "create_edge"
	OpUpdateEdge      OperationType = "update_edge"
	OpDeleteEdge      OperationType = "delete_edge"
	OpUpdateEmbedding OperationType = "update_embedding" // Safe to skip on corruption - regenerable
)
const (
	OpBulkNodes       OperationType = "bulk_create_nodes"
	OpBulkEdges       OperationType = "bulk_create_edges"
	OpBulkDeleteNodes OperationType = "bulk_delete_nodes"
	OpBulkDeleteEdges OperationType = "bulk_delete_edges"
	OpCheckpoint      OperationType = "checkpoint" // Marks snapshot boundaries

	// Transaction boundary markers for ACID compliance
	OpTxBegin  OperationType = "tx_begin"  // Marks transaction start
	OpTxCommit OperationType = "tx_commit" // Marks successful transaction completion
	OpTxAbort  OperationType = "tx_abort"  // Marks explicit transaction rollback
)

Additional WAL operation types (extends OperationType from transaction.go)

type PrefixStatsEngine 

type PrefixStatsEngine interface {
	NodeCountByPrefix(prefix string) (int64, error)
	EdgeCountByPrefix(prefix string) (int64, error)
}

PrefixStatsEngine is an optional extension interface that provides fast per-prefix statistics without scanning and decoding all records.

The prefix refers to the *ID prefix* (e.g., a database namespace prefix like "nornic:") and is applied to stored NodeID/EdgeID values (not the internal key prefix bytes).

This is primarily used by NamespacedEngine so that NodeCount/EdgeCount can remain fast in multi-database deployments while still returning namespace-scoped results.

type PrefixStreamingEngine 

type PrefixStreamingEngine interface {
	StreamingEngine

	// StreamNodesByPrefix streams only nodes whose IDs start with the given prefix.
	StreamNodesByPrefix(ctx context.Context, prefix string, fn func(node *Node) error) error
}

PrefixStreamingEngine extends StreamingEngine with namespace/key-prefix-aware streaming for efficient bounded scans on engines where IDs embed tenant/db prefixes. This avoids full-store scans followed by callback filtering.

type PressureBand 

type PressureBand string

PressureBand represents the MVCC storage pressure level. 

const (
	PressureNormal   PressureBand = "normal"
	PressureHigh     PressureBand = "high"
	PressureCritical PressureBand = "critical"
)

type PropertyIndex 

type PropertyIndex struct {
	Name       string
	Label      string
	Properties []string
	// contains filtered or unexported fields
}

PropertyIndex represents a property index for faster lookups.

type PropertyType 

type PropertyType string

PropertyType represents the expected type of a property. 

const (
	PropertyTypeString   PropertyType = "STRING"
	PropertyTypeInteger  PropertyType = "INTEGER"
	PropertyTypeFloat    PropertyType = "FLOAT"
	PropertyTypeBoolean  PropertyType = "BOOLEAN"
	PropertyTypeDate     PropertyType = "DATE"
	PropertyTypeDateTime PropertyType = "DATETIME" // Legacy alias for zoned datetime
	// Neo4j temporal property type constraints.
	PropertyTypeZonedDateTime PropertyType = "ZONED DATETIME"
	PropertyTypeLocalDateTime PropertyType = "LOCAL DATETIME"
)

type PropertyTypeConstraint 

type PropertyTypeConstraint struct {
	Name         string               `json:"name"`
	EntityType   ConstraintEntityType `json:"entity_type,omitempty"` // defaults to NODE when empty
	Label        string               `json:"label"`                 // label for nodes, relationship type for relationships
	Property     string               `json:"property"`
	ExpectedType PropertyType         `json:"expected_type"`
}

PropertyTypeConstraint represents a type constraint on properties.

func (PropertyTypeConstraint) EffectiveEntityType 

func (c PropertyTypeConstraint) EffectiveEntityType() ConstraintEntityType

EffectiveEntityType returns the entity type, defaulting to NODE for backward compatibility.

type PropertyTypeConstraintOptions 

type PropertyTypeConstraintOptions struct {
	EntityType  ConstraintEntityType
	IfNotExists bool
}

AddPropertyTypeConstraint adds a property type constraint to the schema. This enforces a specific type for a property on a label (NULL values allowed). An optional entityType can be passed to specify RELATIONSHIP constraints. PropertyTypeConstraintOptions holds optional parameters for AddPropertyTypeConstraint.

type QueryAnalyzer 

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

QueryAnalyzer analyzes queries to extract routing information.

func NewQueryAnalyzer 

func NewQueryAnalyzer() *QueryAnalyzer

NewQueryAnalyzer creates a new query analyzer for composite routing.

func (*QueryAnalyzer) AnalyzeQuery 

func (a *QueryAnalyzer) AnalyzeQuery(cypher string) *QueryInfo

AnalyzeQuery extracts routing information from a Cypher query string. This is a simplified analyzer - a full implementation would parse the AST.

func (*QueryAnalyzer) RouteQuery 

func (a *QueryAnalyzer) RouteQuery(queryInfo *QueryInfo, allConstituents []string) []string

RouteQuery determines which constituents to query based on query info.

func (*QueryAnalyzer) SetLabelRouting 

func (a *QueryAnalyzer) SetLabelRouting(label string, constituents []string)

SetLabelRouting configures label-based routing.

func (*QueryAnalyzer) SetPropertyDefault 

func (p *QueryAnalyzer) SetPropertyDefault(propertyName string, constituent string)

SetPropertyDefault sets the default constituent for a property when value not found.

func (*QueryAnalyzer) SetPropertyRouting 

func (p *QueryAnalyzer) SetPropertyRouting(propertyName string, value interface{}, constituent string)

SetPropertyRouting configures property-based routing.

type QueryInfo 

type QueryInfo struct {
	Labels     []string
	Properties map[string]interface{}
	IsWrite    bool
	IsFullScan bool
}

QueryInfo contains information extracted from a query for routing.

type RangeIndex 

type RangeIndex struct {
	Name             string
	Label            string
	Property         string
	Properties       []string             // composite properties (for multi-property constraint indexes)
	EntityType       ConstraintEntityType // NODE or RELATIONSHIP
	OwningConstraint string               // name of the constraint that owns this index (empty if standalone)
	// contains filtered or unexported fields
}

RangeIndex represents an index for range queries on a single property. It maintains a sorted list of entries for efficient O(log n) range queries.

type Receipt 

type Receipt struct {
	TxID        string    `json:"tx_id"`
	WALSeqStart uint64    `json:"wal_seq_start"`
	WALSeqEnd   uint64    `json:"wal_seq_end"`
	Timestamp   time.Time `json:"timestamp"`
	Database    string    `json:"database,omitempty"`
	Hash        string    `json:"hash"`
}

Receipt represents a mutation receipt tied to WAL sequencing. WALSeqEnd should refer to the commit marker sequence for the transaction.

func NewReceipt 

func NewReceipt(txID string, walSeqStart, walSeqEnd uint64, database string, timestamp time.Time) (*Receipt, error)

NewReceipt creates a receipt and computes its hash.

func (*Receipt) UpdateHash 

func (r *Receipt) UpdateHash() error

UpdateHash recomputes the receipt hash from canonical fields.

type RelationshipConstraint 

type RelationshipConstraint struct {
	Name       string
	Type       ConstraintType
	RelType    string // Relationship type (e.g., "KNOWS", "FOLLOWS")
	Properties []string
}

RelationshipConstraint represents a constraint on relationship properties.

type RemoteCypherTx 

type RemoteCypherTx interface {
	QueryCypher(ctx context.Context, statement string, params map[string]interface{}) ([]string, [][]interface{}, error)
	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
}

RemoteCypherTx represents an explicit remote Cypher transaction handle.

type RemoteEngine 

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

RemoteEngine implements Engine by forwarding operations to a remote NornicDB instance via Bolt protocol (preferred) or HTTP tx API (fallback), auto-detected from the URI scheme.

func NewRemoteEngine 

func NewRemoteEngine(cfg RemoteEngineConfig) (*RemoteEngine, error)

NewRemoteEngine creates a remote engine. Transport is auto-detected from the URI scheme.

func (*RemoteEngine) AllEdges 

func (r *RemoteEngine) AllEdges() ([]*Edge, error)

func (*RemoteEngine) AllNodes 

func (r *RemoteEngine) AllNodes() ([]*Node, error)

func (*RemoteEngine) BatchGetNodes 

func (r *RemoteEngine) BatchGetNodes(ids []NodeID) (map[NodeID]*Node, error)

func (*RemoteEngine) BeginCypherTx 

func (r *RemoteEngine) BeginCypherTx(ctx context.Context) (RemoteCypherTx, error)

BeginCypherTx opens an explicit remote transaction handle.

func (*RemoteEngine) BulkCreateEdges 

func (r *RemoteEngine) BulkCreateEdges(edges []*Edge) error

func (*RemoteEngine) BulkCreateNodes 

func (r *RemoteEngine) BulkCreateNodes(nodes []*Node) error

func (*RemoteEngine) BulkDeleteEdges 

func (r *RemoteEngine) BulkDeleteEdges(ids []EdgeID) error

func (*RemoteEngine) BulkDeleteNodes 

func (r *RemoteEngine) BulkDeleteNodes(ids []NodeID) error

func (*RemoteEngine) Close 

func (r *RemoteEngine) Close() error

func (*RemoteEngine) CreateEdge 

func (r *RemoteEngine) CreateEdge(edge *Edge) error

func (*RemoteEngine) CreateNode 

func (r *RemoteEngine) CreateNode(node *Node) (NodeID, error)

func (*RemoteEngine) DeleteByPrefix 

func (r *RemoteEngine) DeleteByPrefix(prefix string) (nodesDeleted int64, edgesDeleted int64, err error)

func (*RemoteEngine) DeleteEdge 

func (r *RemoteEngine) DeleteEdge(id EdgeID) error

func (*RemoteEngine) DeleteNode 

func (r *RemoteEngine) DeleteNode(id NodeID) error

func (*RemoteEngine) EdgeCount 

func (r *RemoteEngine) EdgeCount() (int64, error)

func (*RemoteEngine) GetAllNodes 

func (r *RemoteEngine) GetAllNodes() []*Node

func (*RemoteEngine) GetEdge 

func (r *RemoteEngine) GetEdge(id EdgeID) (*Edge, error)

func (*RemoteEngine) GetEdgeBetween 

func (r *RemoteEngine) GetEdgeBetween(startID, endID NodeID, edgeType string) *Edge

func (*RemoteEngine) GetEdgesBetween 

func (r *RemoteEngine) GetEdgesBetween(startID, endID NodeID) ([]*Edge, error)

func (*RemoteEngine) GetEdgesByType 

func (r *RemoteEngine) GetEdgesByType(edgeType string) ([]*Edge, error)

func (*RemoteEngine) GetFirstNodeByLabel 

func (r *RemoteEngine) GetFirstNodeByLabel(label string) (*Node, error)

func (*RemoteEngine) GetInDegree 

func (r *RemoteEngine) GetInDegree(nodeID NodeID) int

func (*RemoteEngine) GetIncomingEdges 

func (r *RemoteEngine) GetIncomingEdges(nodeID NodeID) ([]*Edge, error)

func (*RemoteEngine) GetNode 

func (r *RemoteEngine) GetNode(id NodeID) (*Node, error)

func (*RemoteEngine) GetNodesByLabel 

func (r *RemoteEngine) GetNodesByLabel(label string) ([]*Node, error)

func (*RemoteEngine) GetOutDegree 

func (r *RemoteEngine) GetOutDegree(nodeID NodeID) int

func (*RemoteEngine) GetOutgoingEdges 

func (r *RemoteEngine) GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)

func (*RemoteEngine) GetSchema 

func (r *RemoteEngine) GetSchema() *SchemaManager

func (*RemoteEngine) NodeCount 

func (r *RemoteEngine) NodeCount() (int64, error)

func (*RemoteEngine) QueryCypher 

func (r *RemoteEngine) QueryCypher(ctx context.Context, statement string, params map[string]interface{}) ([]string, [][]interface{}, error)

QueryCypher executes an arbitrary Cypher query against the remote instance and returns column names and raw rows. This is used by the fabric layer to dispatch fragment queries to remote constituents without going through the node/edge Engine abstraction.

The Bolt transport returns columns from the result record keys. The HTTP transport returns columns from the tx API response.

func (*RemoteEngine) UpdateEdge 

func (r *RemoteEngine) UpdateEdge(edge *Edge) error

func (*RemoteEngine) UpdateNode 

func (r *RemoteEngine) UpdateNode(node *Node) error

type RemoteEngineConfig 

type RemoteEngineConfig struct {
	// URI is the remote NornicDB endpoint (bolt:// or http:// scheme).
	URI string

	// Database is the target database name on the remote instance.
	Database string

	// AuthToken is the caller's authorization header value (e.g. "Bearer <token>"),
	// used for OIDC credential forwarding.
	AuthToken string

	// User and Password are used for explicit basic auth (user_password auth mode).
	User     string
	Password string

	// HTTPClient is an optional custom HTTP client, used only when transport is HTTP.
	HTTPClient *http.Client
}

RemoteEngineConfig configures a remote storage engine.

The URI field determines the transport protocol:

  • bolt://, bolt+s://, bolt+ssc://, neo4j://, neo4j+s://, neo4j+ssc:// → Bolt transport (preferred)
  • http://, https:// → HTTP tx API transport (fallback)

type ReplayError 

type ReplayError struct {
	Sequence  uint64
	Operation OperationType
	Error     error
}

ReplayError captures details about a failed replay entry.

type ReplayResult 

type ReplayResult struct {
	Applied int           // Successfully applied entries
	Skipped int           // Expected skips (duplicates, checkpoints)
	Failed  int           // Unexpected failures
	Errors  []ReplayError // Detailed error information
}

ReplayResult tracks the outcome of WAL replay for observability.

func ReplayWALEntries 

func ReplayWALEntries(engine Engine, entries []WALEntry) ReplayResult

ReplayWALEntries replays multiple entries and tracks results. Expected errors (duplicates, checkpoints) are counted as skipped. Unexpected errors (corruption, constraint violations) are counted as failed.

func (ReplayResult) HasCriticalErrors 

func (r ReplayResult) HasCriticalErrors() bool

HasCriticalErrors returns true if there were unexpected failures.

func (ReplayResult) Summary 

func (r ReplayResult) Summary() string

Summary returns a human-readable summary of replay results.

type RetentionPolicy 

type RetentionPolicy struct {
	MaxVersionsPerKey int
	TTL               time.Duration
}

RetentionPolicy controls default MVCC historical retention for a storage engine. MaxVersionsPerKey applies to closed historical versions; the current head is always preserved. When MaxVersionsPerKey <= 0 the engine skips archival entirely: updates overwrite in place and deletes remove the primary key without creating a version record. TTL optionally protects versions newer than now-TTL from pruning.

func (RetentionPolicy) RetainsHistory added in v1.1.0 

func (p RetentionPolicy) RetainsHistory() bool

RetainsHistory reports whether this policy keeps any closed historical versions. Callers on the write hot path use this to short-circuit archival work when the head-only configuration is active.

type SchemaCompositeIndexDef 

type SchemaCompositeIndexDef struct {
	Name       string   `json:"name"`
	Label      string   `json:"label"`
	Properties []string `json:"properties"`
}

type SchemaDefinition 

type SchemaDefinition struct {
	Version int `json:"version"`

	Constraints         []Constraint         `json:"constraints,omitempty"`
	ConstraintContracts []ConstraintContract `json:"constraint_contracts,omitempty"`

	PropertyTypeConstraints []PropertyTypeConstraint `json:"property_type_constraints,omitempty"`

	PropertyIndexes  []SchemaPropertyIndexDef  `json:"property_indexes,omitempty"`
	CompositeIndexes []SchemaCompositeIndexDef `json:"composite_indexes,omitempty"`
	FulltextIndexes  []FulltextIndex           `json:"fulltext_indexes,omitempty"`
	VectorIndexes    []VectorIndex             `json:"vector_indexes,omitempty"`
	RangeIndexes     []SchemaRangeIndexDef     `json:"range_indexes,omitempty"`

	DecayProfileBundles  []knowledgepolicy.DecayProfileBundle  `json:"decay_profile_bundles,omitempty"`
	DecayProfileBindings []knowledgepolicy.DecayProfileBinding `json:"decay_profile_bindings,omitempty"`
	PromotionProfiles    []knowledgepolicy.PromotionProfileDef `json:"promotion_profiles,omitempty"`
	PromotionPolicies    []knowledgepolicy.PromotionPolicyDef  `json:"promotion_policies,omitempty"`
}

SchemaDefinition is the persisted representation of NornicDB schema rules.

It stores schema *definitions only* (constraints and index definitions), not derived/indexed data structures (like unique value maps), which are rebuilt from stored nodes/edges on startup.

This mirrors Neo4j’s model:

  • schema rules are durable metadata
  • index contents can be rebuilt if needed

type SchemaManager 

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

SchemaManager manages database schema including constraints and indexes.

func NewSchemaManager 

func NewSchemaManager() *SchemaManager

NewSchemaManager creates a new schema manager with empty constraint and index collections.

The schema manager provides thread-safe management of database schema including:

  • Unique constraints (enforce uniqueness on properties)
  • Node key constraints (composite unique keys)
  • Existence constraints (require properties to exist)
  • Property indexes (speed up lookups)
  • Vector indexes (semantic similarity search)
  • Full-text indexes (text search with scoring)

Returns:

  • *SchemaManager ready for use

Example 1 - Basic Usage: 

schema := storage.NewSchemaManager()

// Add unique constraint
constraint := &storage.UniqueConstraint{
	Name:     "unique_user_email",
	Label:    "User",
	Property: "email",
}
schema.AddUniqueConstraint(constraint)

// Validate before creating node
err := schema.ValidateUnique("User", "email", "alice@example.com", "")
if err != nil {
	log.Fatal("Email already exists!")
}

Example 2 - Multiple Constraints: 

schema := storage.NewSchemaManager()

// Email must be unique
schema.AddUniqueConstraint(&storage.UniqueConstraint{
	Name: "unique_email", Label: "User", Property: "email",
})

// Username must be unique
schema.AddUniqueConstraint(&storage.UniqueConstraint{
	Name: "unique_username", Label: "User", Property: "username",
})

// All users must have email property
schema.AddConstraint(storage.Constraint{
	Name: "user_must_have_email",
	Type: storage.ConstraintExists,
	Label: "User",
	Properties: []string{"email"},
})

Example 3 - With Indexes for Performance: 

schema := storage.NewSchemaManager()

// Index for fast lookups
schema.AddPropertyIndex(&storage.PropertyIndex{
	Name:       "idx_user_email",
	Label:      "User",
	Properties: []string{"email"},
})

// Vector index for semantic search
schema.AddVectorIndex(&storage.VectorIndex{
	Name:       "doc_embeddings",
	Label:      "Document",
	Property:   "embedding",
	Dimensions: 1024,
})

ELI12:

Think of a SchemaManager like a rule book for your database:

  • "Every person must have a unique name" (unique constraint)
  • "You can't create a person without an age" (existence constraint)
  • "Make a quick-lookup list for emails" (index)

Before you add data, the SchemaManager checks: "Does this follow the rules?" If yes, data goes in. If no, you get an error. It keeps your database clean!

Thread Safety: 

All methods are thread-safe for concurrent access.

func (*SchemaManager) AddCompositeIndex 

func (sm *SchemaManager) AddCompositeIndex(name, label string, properties []string) error

AddCompositeIndex creates a composite index on multiple properties. Composite indexes enable efficient queries that filter on multiple properties.

Example usage: 

sm.AddCompositeIndex("user_location_idx", "User", []string{"country", "city", "zipcode"})

This enables efficient queries like:

  • WHERE country = 'US' AND city = 'NYC' AND zipcode = '10001' (full match)
  • WHERE country = 'US' AND city = 'NYC' (prefix match)
  • WHERE country = 'US' (prefix match, uses first property only)

func (*SchemaManager) AddConstraint 

func (sm *SchemaManager) AddConstraint(c Constraint, ifNotExists ...bool) error

AddConstraint adds a constraint to the schema. Stores constraint in both the constraints map and uniqueConstraints (for backward compatibility).

Conflict rules (matching Neo4j behavior):

  • Same name, already exists with identical schema+type: error, unless ifNotExists (then no-op)
  • Same name, different schema or type: error
  • Different name, same schema + same type: error (duplicate schema), unless ifNotExists
  • Uniqueness vs relationship key on same schema: error (conflicting)

Pass ifNotExists=true when the DDL includes IF NOT EXISTS; duplicate-schema is then a no-op.

func (*SchemaManager) AddConstraintContractBundle 

func (sm *SchemaManager) AddConstraintContractBundle(contract ConstraintContract, compiledConstraints []Constraint, compiledTypes []PropertyTypeConstraint, ifNotExists bool) error

func (*SchemaManager) AddFulltextIndex 

func (sm *SchemaManager) AddFulltextIndex(name string, labels, properties []string) error

AddFulltextIndex adds a node-scoped full-text index.

func (*SchemaManager) AddFulltextRelationshipIndex added in v1.1.2 

func (sm *SchemaManager) AddFulltextRelationshipIndex(name string, relTypes, properties []string) error

AddFulltextRelationshipIndex adds a relationship-scoped full-text index. Mirrors AddFulltextIndex but populates RelationshipTypes instead of Labels. The two share the same `fulltextIndexes` map so every existing get/list/remove path works for both kinds; consumers that need to distinguish check which scope slice is non-empty.

func (*SchemaManager) AddPropertyIndex 

func (sm *SchemaManager) AddPropertyIndex(name, label string, properties []string) error

AddPropertyIndex adds a property index.

func (*SchemaManager) AddPropertyTypeConstraint 

func (sm *SchemaManager) AddPropertyTypeConstraint(name, label, property string, expectedType PropertyType, entityType ...ConstraintEntityType) error

AddPropertyTypeConstraint adds a property type constraint to the schema. The entityType parameter controls NODE vs RELATIONSHIP scoping.

func (*SchemaManager) AddPropertyTypeConstraintWithOptions 

func (sm *SchemaManager) AddPropertyTypeConstraintWithOptions(name, label, property string, expectedType PropertyType, opts PropertyTypeConstraintOptions) error

AddPropertyTypeConstraintWithOptions adds a property type constraint with full options.

func (*SchemaManager) AddRangeIndex 

func (sm *SchemaManager) AddRangeIndex(name, label, property string) error

AddRangeIndex adds a range index for a single property.

func (*SchemaManager) AddUniqueConstraint 

func (sm *SchemaManager) AddUniqueConstraint(name, label, property string, ifNotExists ...bool) error

AddUniqueConstraint adds a unique constraint. Stores in both uniqueConstraints (for value tracking) and constraints (for lookup by label). Pass ifNotExists=true for IF NOT EXISTS semantics (duplicate is no-op).

func (*SchemaManager) AddVectorIndex 

func (sm *SchemaManager) AddVectorIndex(name, label, property string, dimensions int, similarityFunc string) error

AddVectorIndex adds a vector index.

func (*SchemaManager) AlterDecayProfile added in v1.1.0 

func (sm *SchemaManager) AlterDecayProfile(name string, updates map[string]interface{}) error

AlterDecayProfile updates options on an existing decay profile bundle.

func (*SchemaManager) AlterPromotionPolicy added in v1.1.0 

func (sm *SchemaManager) AlterPromotionPolicy(name string, updates map[string]interface{}) error

AlterPromotionPolicy updates an existing promotion policy.

func (*SchemaManager) AlterPromotionProfile added in v1.1.0 

func (sm *SchemaManager) AlterPromotionProfile(name string, updates map[string]interface{}) error

AlterPromotionProfile updates options on an existing promotion profile.

func (*SchemaManager) CheckUniqueConstraint 

func (sm *SchemaManager) CheckUniqueConstraint(label, property string, value interface{}, excludeNode NodeID) error

CheckUniqueConstraint checks if a value violates a unique constraint. Returns error if constraint is violated.

func (*SchemaManager) CreateDecayProfileBinding added in v1.1.0 

func (sm *SchemaManager) CreateDecayProfileBinding(binding knowledgepolicy.DecayProfileBinding, ifNotExists ...bool) error

CreateDecayProfileBinding adds a decay profile binding to the schema.

func (*SchemaManager) CreateDecayProfileBundle added in v1.1.0 

func (sm *SchemaManager) CreateDecayProfileBundle(bundle knowledgepolicy.DecayProfileBundle, ifNotExists ...bool) error

CreateDecayProfileBundle adds a decay profile bundle to the schema.

func (*SchemaManager) CreatePromotionPolicy added in v1.1.0 

func (sm *SchemaManager) CreatePromotionPolicy(policy knowledgepolicy.PromotionPolicyDef, ifNotExists ...bool) error

CreatePromotionPolicy adds a promotion policy to the schema.

func (*SchemaManager) CreatePromotionProfile added in v1.1.0 

func (sm *SchemaManager) CreatePromotionProfile(profile knowledgepolicy.PromotionProfileDef, ifNotExists ...bool) error

CreatePromotionProfile adds a promotion profile to the schema.

func (*SchemaManager) DropConstraint 

func (sm *SchemaManager) DropConstraint(name string) error

DropConstraint removes a constraint (by name) from the schema. This supports both standard constraints and property type constraints.

func (*SchemaManager) DropDecayProfile added in v1.1.0 

func (sm *SchemaManager) DropDecayProfile(name string, ifExists ...bool) error

DropDecayProfile removes a decay profile bundle or binding by name.

func (*SchemaManager) DropIndex 

func (sm *SchemaManager) DropIndex(name string) error

DropIndex removes an index (by name) from the schema. It searches across all index types: property, composite, fulltext, vector, and range.

func (*SchemaManager) DropPromotionPolicy added in v1.1.0 

func (sm *SchemaManager) DropPromotionPolicy(name string, ifExists ...bool) error

DropPromotionPolicy removes a promotion policy by name.

func (*SchemaManager) DropPromotionProfile added in v1.1.0 

func (sm *SchemaManager) DropPromotionProfile(name string, ifExists ...bool) error

DropPromotionProfile removes a promotion profile by name.

func (*SchemaManager) ExportDefinition 

func (sm *SchemaManager) ExportDefinition() *SchemaDefinition

ExportDefinition returns a stable, persisted representation of the schema. The returned object is safe to serialize and does not include runtime caches.

func (*SchemaManager) GetAllConstraintContracts 

func (sm *SchemaManager) GetAllConstraintContracts() []ConstraintContract

func (*SchemaManager) GetAllConstraints 

func (sm *SchemaManager) GetAllConstraints() []Constraint

GetAllConstraints returns all constraints in the schema, regardless of label. This is used by db.constraints() procedure to list all constraints.

func (*SchemaManager) GetAllPropertyTypeConstraints 

func (sm *SchemaManager) GetAllPropertyTypeConstraints() []PropertyTypeConstraint

GetAllPropertyTypeConstraints returns all property type constraints.

func (*SchemaManager) GetBindingTable added in v1.1.0 

func (sm *SchemaManager) GetBindingTable() *knowledgepolicy.BindingTable

GetBindingTable returns the current compiled binding table, or nil if not built.

func (*SchemaManager) GetCompositeIndex 

func (sm *SchemaManager) GetCompositeIndex(name string) (*CompositeIndex, bool)

GetCompositeIndex returns a composite index by name.

func (*SchemaManager) GetCompositeIndexesForLabel 

func (sm *SchemaManager) GetCompositeIndexesForLabel(label string) []*CompositeIndex

GetCompositeIndexForLabel returns all composite indexes for a label.

func (*SchemaManager) GetConstraintContractsForTarget 

func (sm *SchemaManager) GetConstraintContractsForTarget(entityType ConstraintEntityType, labelOrType string) []ConstraintContract

func (*SchemaManager) GetConstraints 

func (sm *SchemaManager) GetConstraints() []UniqueConstraint

GetConstraints returns all unique constraints.

func (*SchemaManager) GetConstraintsForLabels 

func (sm *SchemaManager) GetConstraintsForLabels(labels []string) []Constraint

GetConstraintsForLabels returns all constraints for given labels. Returns constraints from the constraints map, preserving their original types.

func (*SchemaManager) GetFulltextIndex 

func (sm *SchemaManager) GetFulltextIndex(name string) (*FulltextIndex, bool)

GetFulltextIndex returns a fulltext index by name.

func (*SchemaManager) GetIndexStats 

func (sm *SchemaManager) GetIndexStats() []IndexStats

GetIndexStats returns statistics for all indexes.

func (*SchemaManager) GetIndexes 

func (sm *SchemaManager) GetIndexes() []interface{}

GetIndexes returns all indexes.

func (*SchemaManager) GetPropertyIndex 

func (sm *SchemaManager) GetPropertyIndex(label, property string) (*PropertyIndex, bool)

GetPropertyIndex returns a property index by label and property.

func (*SchemaManager) GetPropertyTypeConstraintsForLabels 

func (sm *SchemaManager) GetPropertyTypeConstraintsForLabels(labels []string) []PropertyTypeConstraint

GetPropertyTypeConstraintsForLabels returns type constraints for the given labels.

func (*SchemaManager) GetRangeIndex 

func (sm *SchemaManager) GetRangeIndex(name string) (*RangeIndex, bool)

GetRangeIndex returns a range index by name.

func (*SchemaManager) GetVectorIndex 

func (sm *SchemaManager) GetVectorIndex(name string) (*VectorIndex, bool)

GetVectorIndex returns a vector index by name.

func (*SchemaManager) HasAnyConstraintContract added in v1.1.0 

func (sm *SchemaManager) HasAnyConstraintContract() bool

HasAnyConstraintContract reports whether this schema has at least one constraint contract registered. Callers use it to short-circuit work in the per-commit validator (the expensive adjacency walk is pointless when no contract exists).

func (*SchemaManager) HasAnyPropertyIndexForLabel added in v1.1.0 

func (sm *SchemaManager) HasAnyPropertyIndexForLabel(label string) bool

HasAnyPropertyIndexForLabel reports whether ANY property index is declared against the given label. Used by storage-side index maintenance to short-circuit per-property lookups when no index touches the label at all.

func (*SchemaManager) HasPropertyIndex added in v1.1.0 

func (sm *SchemaManager) HasPropertyIndex(label, property string) bool

HasPropertyIndex reports whether a property index exists for the given label+property combination. Callers can use this to choose between index-backed per-row lookups and batch preloads.

func (*SchemaManager) LookupUniqueConstraintValue added in v1.0.43 

func (sm *SchemaManager) LookupUniqueConstraintValue(label, property string, value interface{}) (NodeID, bool, bool)

LookupUniqueConstraintValue returns the node currently registered for a single-property uniqueness constraint value. The second return value reports whether the value is present, and the third reports whether the unique constraint exists.

func (*SchemaManager) LookupUniqueConstraintValueForPlanning added in v1.1.0 

func (sm *SchemaManager) LookupUniqueConstraintValueForPlanning(label, property string, value interface{}) (nodeID NodeID, valueFound bool, constraintExists bool, cacheComplete bool)

LookupUniqueConstraintValueForPlanning returns the node currently registered for a single-property uniqueness constraint value, plus whether the values cache has been rebuilt from storage and can be trusted for misses. Planners may trust absence only when cacheComplete is true; otherwise they must retain a scan fallback because the cache may not have been rebuilt from storage yet.

func (*SchemaManager) PropertyIndexAllNonNil 

func (sm *SchemaManager) PropertyIndexAllNonNil(label, property string, descending bool) []NodeID

PropertyIndexAllNonNil returns all node IDs from the property index in key order, excluding nil keys.

func (*SchemaManager) PropertyIndexDelete 

func (sm *SchemaManager) PropertyIndexDelete(label, property string, nodeID NodeID, value interface{}) error

PropertyIndexDelete removes a node from a property index.

func (*SchemaManager) PropertyIndexInsert 

func (sm *SchemaManager) PropertyIndexInsert(label, property string, nodeID NodeID, value interface{}) error

PropertyIndexInsert adds a node to a property index.

func (*SchemaManager) PropertyIndexLookup 

func (sm *SchemaManager) PropertyIndexLookup(label, property string, value interface{}) []NodeID

PropertyIndexLookup looks up node IDs by property value using an index. Returns nil if no index exists for the label/property.

func (*SchemaManager) PropertyIndexTopK 

func (sm *SchemaManager) PropertyIndexTopK(label, property string, limit int, descending bool) []NodeID

PropertyIndexTopK returns up to limit node IDs from a property index ordered by indexed property value. Nil keys are skipped.

func (*SchemaManager) RangeIndexDelete 

func (sm *SchemaManager) RangeIndexDelete(name string, nodeID NodeID) error

RangeIndexDelete removes a value from a range index.

func (*SchemaManager) RangeIndexInsert 

func (sm *SchemaManager) RangeIndexInsert(name string, nodeID NodeID, value interface{}) error

RangeIndexInsert adds a value to a range index.

func (*SchemaManager) RangeQuery 

func (sm *SchemaManager) RangeQuery(name string, minVal, maxVal interface{}, includeMin, includeMax bool) ([]NodeID, error)

RangeQuery performs a range query on a range index. Returns node IDs where value is in range [minVal, maxVal]. Pass nil for minVal or maxVal to indicate unbounded.

func (*SchemaManager) RegisterUniqueValue 

func (sm *SchemaManager) RegisterUniqueValue(label, property string, value interface{}, nodeID NodeID)

RegisterUniqueValue registers a value for a unique constraint.

func (*SchemaManager) ReplaceFromDefinition 

func (sm *SchemaManager) ReplaceFromDefinition(def *SchemaDefinition) error

ReplaceFromDefinition replaces the in-memory schema contents with the given definition.

This does NOT persist anything, and it intentionally discards runtime caches (unique value maps, index maps, etc.). Those must be rebuilt from data.

func (*SchemaManager) SetBindingTable added in v1.1.0 

func (sm *SchemaManager) SetBindingTable(bt *knowledgepolicy.BindingTable)

SetBindingTable replaces the compiled binding table.

func (*SchemaManager) SetKnowledgePolicyChangedHook added in v1.1.0 

func (sm *SchemaManager) SetKnowledgePolicyChangedHook(hook func())

SetKnowledgePolicyChangedHook registers a callback that runs after a knowledge-policy mutation has been persisted and the schema lock released.

func (*SchemaManager) SetPersister 

func (sm *SchemaManager) SetPersister(persist func(def *SchemaDefinition) error)

SetPersister sets an optional persistence hook for schema changes. When set, schema mutations will attempt to persist the updated schema definition and will roll back the in-memory change if persistence fails.

func (*SchemaManager) ShowDecayProfiles added in v1.1.0 

func (sm *SchemaManager) ShowDecayProfiles() ([]knowledgepolicy.DecayProfileBundle, []knowledgepolicy.DecayProfileBinding)

ShowDecayProfiles returns all stored decay profile bundles and bindings.

func (*SchemaManager) ShowPromotionPolicies added in v1.1.0 

func (sm *SchemaManager) ShowPromotionPolicies() []knowledgepolicy.PromotionPolicyDef

ShowPromotionPolicies returns all stored promotion policies.

func (*SchemaManager) ShowPromotionProfiles added in v1.1.0 

func (sm *SchemaManager) ShowPromotionProfiles() []knowledgepolicy.PromotionProfileDef

ShowPromotionProfiles returns all stored promotion profiles.

func (*SchemaManager) UnregisterUniqueValue 

func (sm *SchemaManager) UnregisterUniqueValue(label, property string, value interface{})

UnregisterUniqueValue removes a value from a unique constraint.

type SchemaPropertyIndexDef 

type SchemaPropertyIndexDef struct {
	Name       string   `json:"name"`
	Label      string   `json:"label"`
	Properties []string `json:"properties"`
}

type SchemaRangeIndexDef 

type SchemaRangeIndexDef struct {
	Name             string               `json:"name"`
	Label            string               `json:"label"`
	Property         string               `json:"property"`
	Properties       []string             `json:"properties,omitempty"`
	EntityType       ConstraintEntityType `json:"entity_type,omitempty"`
	OwningConstraint string               `json:"owning_constraint,omitempty"`
}

type SearchSizeFn added in v1.1.0 

type SearchSizeFn func() int64

SearchSizeFn returns the cumulative search-index byte size across all per-database search services. Plan 04-05 wires this via search.SearchService IndexSizeBytes(); for Plan 04-04 it can be nil (returns 0).

type SerializerMigrationOptions 

type SerializerMigrationOptions struct {
	BatchSize int
	DryRun    bool
}

SerializerMigrationOptions controls the in-place gob → msgpack rewrite. There is no source/target choice: the engine only emits msgpack, and the only thing this tool does is upgrade legacy gob bodies in an existing data directory so they can be read by current code without going through the legacy decode arm forever.

type SerializerMigrationStats 

type SerializerMigrationStats struct {
	DataDir             string
	HasLegacyData       bool
	NodesConverted      int
	EdgesConverted      int
	EmbeddingsConverted int
	SkippedExisting     int
	TotalScanned        int
}

SerializerMigrationStats reports conversion results for the in-place gob → msgpack rewrite.

func MigrateBadgerToMsgpack added in v1.1.0 

func MigrateBadgerToMsgpack(dataDir string, opts SerializerMigrationOptions) (SerializerMigrationStats, error)

MigrateBadgerToMsgpack rewrites any legacy gob-encoded node, edge, or embedding bodies in the given data directory as msgpack. The database must be offline (no running server). New writes have always been msgpack since this engine version, so on a fresh database this is a no-op.

func MigrateBadgerToMsgpackWithDB added in v1.1.0 

func MigrateBadgerToMsgpackWithDB(db *badger.DB, dataDir string, opts SerializerMigrationOptions) (SerializerMigrationStats, error)

MigrateBadgerToMsgpackWithDB is MigrateBadgerToMsgpack against an already-open *badger.DB. Used by tests and offline tooling.

type SignalType 

type SignalType string

SignalType describes how an edge was detected/created. 

const (
	SignalSimilarity   SignalType = "similarity" // Semantic similarity
	SignalCoAccess     SignalType = "coaccess"   // Co-access pattern
	SignalTopology     SignalType = "topology"   // Topological link prediction
	SignalLLMInference SignalType = "llm-infer"  // LLM-based inference
	SignalManual       SignalType = "manual"     // User-created
	SignalTransitive   SignalType = "transitive" // Transitive inference
	SignalTemporal     SignalType = "temporal"   // Temporal proximity
)

type Snapshot 

type Snapshot struct {
	Sequence  uint64    `json:"sequence"`
	Timestamp time.Time `json:"timestamp"`
	Nodes     []*Node   `json:"nodes"`
	Edges     []*Edge   `json:"edges"`
	Version   string    `json:"version"`
}

Snapshot represents a point-in-time snapshot of the database.

func LoadSnapshot 

func LoadSnapshot(path string) (*Snapshot, error)

LoadSnapshot reads a snapshot from disk.

type SnapshotReaderInfo 

type SnapshotReaderInfo struct {
	ReaderID        string
	SnapshotVersion MVCCVersion
	StartTime       time.Time
	Namespace       string
}

SnapshotReaderInfo describes an active MVCC snapshot reader.

type SnapshotReaderRegistry 

type SnapshotReaderRegistry interface {
	Register(info SnapshotReaderInfo) (string, func())
	ActiveCount() int64
	Snapshot() []SnapshotReaderInfo
}

SnapshotReaderRegistry exposes active snapshot-reader tracking.

type StorageEventNotifier 

type StorageEventNotifier interface {
	// Node events
	OnNodeCreated(callback NodeEventCallback)
	OnNodeUpdated(callback NodeEventCallback)
	OnNodeDeleted(callback NodeDeleteCallback)

	// Edge events
	OnEdgeCreated(callback EdgeEventCallback)
	OnEdgeUpdated(callback EdgeEventCallback)
	OnEdgeDeleted(callback EdgeDeleteCallback)
}

StorageEventNotifier is an optional interface that storage engines can implement to notify listeners of storage changes. This enables automatic synchronization between storage and external services (search indexes, embeddings, caches, etc.).

Events are fired AFTER the storage operation succeeds, ensuring consistency.

Example: 

if notifier, ok := engine.(storage.StorageEventNotifier); ok {
	notifier.OnNodeCreated(func(node *storage.Node) {
		searchService.IndexNode(node)
	})
	notifier.OnNodeDeleted(func(nodeID storage.NodeID) {
		searchService.RemoveNode(nodeID)
	})
	notifier.OnEdgeCreated(func(edge *storage.Edge) {
		graphAnalyzer.UpdateMetrics(edge)
	})
}

type StreamingEngine 

type StreamingEngine interface {
	Engine

	// StreamNodes iterates over all nodes without loading all into memory.
	// The callback is called for each node. Return an error to stop iteration.
	// Returns nil on successful completion, context.Canceled on cancellation.
	StreamNodes(ctx context.Context, fn func(node *Node) error) error

	// StreamEdges iterates over all edges without loading all into memory.
	StreamEdges(ctx context.Context, fn func(edge *Edge) error) error

	// StreamNodeChunks iterates over nodes in chunks for batch processing.
	// More efficient than StreamNodes when processing in batches.
	StreamNodeChunks(ctx context.Context, chunkSize int, fn func(nodes []*Node) error) error
}

StreamingEngine extends Engine with streaming iteration support. This is optional - engines that don't support streaming will use the default AllNodes/AllEdges with chunked processing.

type StructuredLogger added in v1.1.0 

type StructuredLogger interface {
	Logger() *slog.Logger
}

StructuredLogger is implemented by storage engines that expose their structured *slog.Logger so wrappers (NamespacedEngine, WALEngine) can derive a child logger without constructor plumbing.

type SuppressionStateChange added in v1.1.0 

type SuppressionStateChange struct {
	EntityID string
	Tokens   []string
	IsEdge   bool
}

type TemporalCurrentNodeEngine 

type TemporalCurrentNodeEngine interface {
	IsCurrentTemporalNode(node *Node, asOf time.Time) (bool, error)
}

TemporalCurrentNodeEngine is an optional extension interface for deciding whether a temporal node represents the current/live version for indexing and query routing.

type TemporalLookupEngine 

type TemporalLookupEngine interface {
	GetTemporalNodeAsOf(label, keyProp string, keyValue interface{}, validFromProp, validToProp string, asOf time.Time) (*Node, error)
}

TemporalLookupEngine is an optional extension interface for efficient temporal lookups on a namespaced engine view.

type TemporalMaintenanceEngine 

type TemporalMaintenanceEngine interface {
	RebuildTemporalIndexes(ctx context.Context) error
	PruneTemporalHistory(ctx context.Context, opts TemporalPruneOptions) (int64, error)
}

TemporalMaintenanceEngine is an optional extension interface for rebuilding and pruning temporal index state after upgrades, restores, or operator maintenance.

type TemporalPruneOptions 

type TemporalPruneOptions struct {
	// MaxVersionsPerKey keeps at most this many closed historical versions per temporal key.
	// Zero means unlimited.
	MaxVersionsPerKey int

	// MinRetentionAge protects versions newer than now-MinRetentionAge from pruning.
	// Zero means no age-based protection.
	MinRetentionAge time.Duration
}

TemporalPruneOptions controls pruning of older temporal versions.

type TracedEngine added in v1.1.0 

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

TracedEngine wraps an Engine and emits nornicdb.storage.<op> spans for each operation. The context used for span parenting is set via SetContext by the calling layer (cypher executor) which has the active span context.

func NewTracedEngine added in v1.1.0 

func NewTracedEngine(inner Engine) *TracedEngine

NewTracedEngine wraps inner with tracing instrumentation.

func (*TracedEngine) AllEdges added in v1.1.0 

func (t *TracedEngine) AllEdges() ([]*Edge, error)

func (*TracedEngine) AllNodes added in v1.1.0 

func (t *TracedEngine) AllNodes() ([]*Node, error)

func (*TracedEngine) BatchGetNodes added in v1.1.0 

func (t *TracedEngine) BatchGetNodes(ids []NodeID) (map[NodeID]*Node, error)

func (*TracedEngine) BulkCreateEdges added in v1.1.0 

func (t *TracedEngine) BulkCreateEdges(edges []*Edge) error

func (*TracedEngine) BulkCreateNodes added in v1.1.0 

func (t *TracedEngine) BulkCreateNodes(nodes []*Node) error

func (*TracedEngine) CreateEdge added in v1.1.0 

func (t *TracedEngine) CreateEdge(edge *Edge) error

func (*TracedEngine) CreateNode added in v1.1.0 

func (t *TracedEngine) CreateNode(node *Node) (NodeID, error)

func (*TracedEngine) DeleteEdge added in v1.1.0 

func (t *TracedEngine) DeleteEdge(id EdgeID) error

func (*TracedEngine) DeleteNode added in v1.1.0 

func (t *TracedEngine) DeleteNode(id NodeID) error

func (*TracedEngine) GetEdge added in v1.1.0 

func (t *TracedEngine) GetEdge(id EdgeID) (*Edge, error)

func (*TracedEngine) GetEdgesByType added in v1.1.0 

func (t *TracedEngine) GetEdgesByType(edgeType string) ([]*Edge, error)

func (*TracedEngine) GetIncomingEdges added in v1.1.0 

func (t *TracedEngine) GetIncomingEdges(nodeID NodeID) ([]*Edge, error)

func (*TracedEngine) GetNode added in v1.1.0 

func (t *TracedEngine) GetNode(id NodeID) (*Node, error)

func (*TracedEngine) GetNodesByLabel added in v1.1.0 

func (t *TracedEngine) GetNodesByLabel(label string) ([]*Node, error)

func (*TracedEngine) GetOutgoingEdges added in v1.1.0 

func (t *TracedEngine) GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)

func (*TracedEngine) SetContext added in v1.1.0 

func (t *TracedEngine) SetContext(ctx context.Context)

SetContext updates the context used for span parenting. Called by the cypher executor at the start of Execute so storage spans nest under the cypher span.

func (*TracedEngine) Unwrap added in v1.1.0 

func (t *TracedEngine) Unwrap() Engine

Unwrap returns the underlying Engine (for type assertions by callers that need access to the concrete engine, e.g. AsyncEngine).

func (*TracedEngine) UpdateEdge added in v1.1.0 

func (t *TracedEngine) UpdateEdge(edge *Edge) error

func (*TracedEngine) UpdateNode added in v1.1.0 

func (t *TracedEngine) UpdateNode(node *Node) error

type Transaction 

type Transaction = BadgerTransaction

Transaction is the public closure-facing transaction type used by DB.Update and DB.View.

type TransactionRecoveryResult 

type TransactionRecoveryResult struct {
	SnapshotSeq            uint64
	Transactions           map[string]*TransactionState
	CommittedTransactions  int
	RolledBackTransactions int
	AbortedTransactions    int
	FailedTransactions     []string
	UndoErrors             []string
	NonTxApplied           int
	NonTxErrors            []string
}

TransactionRecoveryResult contains detailed statistics from transaction-aware recovery.

func (*TransactionRecoveryResult) HasErrors 

func (r *TransactionRecoveryResult) HasErrors() bool

HasErrors returns true if there were any errors during recovery.

func (*TransactionRecoveryResult) Summary 

func (r *TransactionRecoveryResult) Summary() string

Summary returns a human-readable summary of the recovery.

type TransactionState 

type TransactionState struct {
	TxID    string
	Entries []WALEntry // All entries in this transaction (in order)
	Started bool       // True if we saw TxBegin
	Done    bool       // True if we saw TxCommit or TxAbort
	Aborted bool       // True if explicitly aborted
}

TransactionState tracks the state of an in-progress transaction during recovery.

type TransactionStatus 

type TransactionStatus string

TransactionStatus represents the current state of a transaction. 

const (
	TxStatusActive     TransactionStatus = "active"
	TxStatusCommitted  TransactionStatus = "committed"
	TxStatusRolledBack TransactionStatus = "rolled_back"
)

type TrustLevel 

type TrustLevel int

TrustLevel defines how trustworthy a node is for edge materialization. 

const (
	// TrustLevelDefault is the standard trust level
	TrustLevelDefault TrustLevel = 0
	// TrustLevelLow requires higher confidence for edges
	TrustLevelLow TrustLevel = -1
	// TrustLevelHigh allows lower confidence edges
	TrustLevelHigh TrustLevel = 1
	// TrustLevelPinned edges never decay
	TrustLevelPinned TrustLevel = 2
)

func (TrustLevel) ConfidenceAdjustment 

func (t TrustLevel) ConfidenceAdjustment() float64

ConfidenceAdjustment returns the confidence threshold adjustment for this trust level. Positive values increase required confidence, negative values decrease it.

func (TrustLevel) String 

func (t TrustLevel) String() string

String returns the string representation of TrustLevel.

type UniqueConstraint 

type UniqueConstraint struct {
	Name     string
	Label    string
	Property string
	// contains filtered or unexported fields
}

UniqueConstraint represents a unique constraint on a label and property.

type VectorIndex 

type VectorIndex struct {
	Name           string
	Label          string
	Property       string
	Dimensions     int
	SimilarityFunc string // "cosine", "euclidean", "dot"
}

VectorIndex represents a vector similarity index.

type WAL 

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

WAL provides write-ahead logging for durability. Thread-safe for concurrent writes.

func NewWAL 

func NewWAL(dir string, cfg *WALConfig) (*WAL, error)

NewWAL creates a new write-ahead log.

func (*WAL) Append 

func (w *WAL) Append(op OperationType, data interface{}) error

Append writes a new entry to the WAL using atomic write format.

The atomic format ensures partial writes are detectable: 

[4 bytes: magic "WALE"]
[1 byte: format version]
[4 bytes: payload length]
[N bytes: JSON-encoded entry]
[4 bytes: CRC32 of payload]

If crash occurs mid-write:

  • Missing magic: entry doesn't exist
  • Missing/truncated payload: length mismatch detected
  • Missing checksum: CRC verification fails

func (*WAL) AppendReturningSeq 

func (w *WAL) AppendReturningSeq(op OperationType, data interface{}) (uint64, error)

AppendReturningSeq writes a new entry to the WAL and returns its sequence number.

func (*WAL) AppendTxAbort 

func (w *WAL) AppendTxAbort(database, txID, reason string) (uint64, error)

AppendTxAbort writes a transaction-abort marker to the WAL.

func (*WAL) AppendTxBegin 

func (w *WAL) AppendTxBegin(database, txID string, metadata map[string]string) (uint64, error)

AppendTxBegin writes a transaction-begin marker to the WAL.

func (*WAL) AppendTxCommit 

func (w *WAL) AppendTxCommit(database, txID string, opCount int) (uint64, error)

AppendTxCommit writes a transaction-commit marker to the WAL.

func (*WAL) AppendWithDatabase 

func (w *WAL) AppendWithDatabase(op OperationType, data interface{}, database string) error

AppendWithDatabase writes a new entry to the WAL with database/namespace information.

func (*WAL) AppendWithDatabaseReturningSeq 

func (w *WAL) AppendWithDatabaseReturningSeq(op OperationType, data interface{}, database string) (uint64, error)

AppendWithDatabaseReturningSeq writes a new entry to the WAL with database/namespace information and returns the assigned sequence number.

func (*WAL) ApplyRetention 

func (w *WAL) ApplyRetention(snapshotSeq uint64) error

ApplyRetention deletes sealed segments that are safe to drop after a snapshot. Segments are only deleted if their last sequence is <= snapshotSeq.

func (*WAL) Checkpoint 

func (w *WAL) Checkpoint() error

Checkpoint creates a checkpoint marker for snapshot boundaries.

func (*WAL) Close 

func (w *WAL) Close() error

Close closes the WAL, flushing all pending writes.

func (*WAL) Config 

func (w *WAL) Config() *WALConfig

Config returns the WAL configuration (read-only access).

func (*WAL) CreateSnapshot 

func (w *WAL) CreateSnapshot(engine Engine) (*Snapshot, error)

CreateSnapshot creates a point-in-time snapshot from the engine.

func (*WAL) IsDegraded 

func (w *WAL) IsDegraded() bool

IsDegraded reports whether this WAL has detected corruption and performed best-effort recovery actions. When true, callers should consider the durability state degraded and surface it in health endpoints.

func (*WAL) LastCorruptionDiagnostics 

func (w *WAL) LastCorruptionDiagnostics() *CorruptionDiagnostics

LastCorruptionDiagnostics returns the last corruption diagnostics captured by this WAL. Returns nil if no corruption has been observed.

func (*WAL) NewBatch 

func (w *WAL) NewBatch() *BatchWriter

NewBatch creates a new batch writer.

func (*WAL) NewBatchWithTxID 

func (w *WAL) NewBatchWithTxID(txID string) *BatchWriter

NewBatchWithTxID creates a new batch writer that stamps each entry with a TxID. The TxID will be embedded in WAL payloads (e.g., WALNodeData.TxID).

func (*WAL) Sequence 

func (w *WAL) Sequence() uint64

Sequence returns the current sequence number.

func (*WAL) Stats 

func (w *WAL) Stats() WALStats

Stats returns current WAL statistics.

func (*WAL) Sync 

func (w *WAL) Sync() error

Sync flushes all buffered writes to disk.

Lock discipline: we acquire w.mu ONLY to drain the bufio userspace buffer to the kernel (a short, bounded memcpy+write), then release it BEFORE the kernel fsync. The fsync is serialized instead through syncMu so concurrent Sync() callers don't issue duplicate fsyncs while still letting Append() progress against w.mu in parallel. This eliminated the dominant mutex-contention source seen during bulk seed: batchSyncLoop's 100ms ticks used to block every Append for the duration of each fsync.

func (*WAL) TruncateAfterSnapshot 

func (w *WAL) TruncateAfterSnapshot(snapshotSeq uint64) error

TruncateAfterSnapshot truncates the WAL after a successful snapshot. This removes all entries up to and including the snapshot sequence number, preventing unbounded WAL growth. Call this after SaveSnapshot succeeds.

The process is crash-safe:

  1. Close current WAL file
  2. Read entries after snapshot sequence
  3. Write new WAL with only post-snapshot entries
  4. Atomically rename new WAL over old
  5. Reopen WAL for appends

If the system crashes during truncation:

  • Old WAL remains intact (rename is atomic)
  • Recovery will replay full WAL (safe, just slower)
  • Retry truncation on next snapshot

Example: 

snapshot, _ := wal.CreateSnapshot(engine)
SaveSnapshot(snapshot, "data/snapshot.json")
wal.TruncateAfterSnapshot(snapshot.Sequence) // Reclaim disk space

type WALBulkDeleteEdgesData 

type WALBulkDeleteEdgesData struct {
	IDs      []string `json:"ids"`
	OldEdges []*Edge  `json:"old_edges,omitempty"` // Complete edges being deleted (undo)
	TxID     string   `json:"tx_id,omitempty"`     // Transaction ID for grouping
}

WALBulkDeleteEdgesData holds bulk edge deletion data with undo support.

type WALBulkDeleteNodesData 

type WALBulkDeleteNodesData struct {
	IDs      []string `json:"ids"`
	OldNodes []*Node  `json:"old_nodes,omitempty"` // Complete nodes being deleted (undo)
	TxID     string   `json:"tx_id,omitempty"`     // Transaction ID for grouping
}

WALBulkDeleteNodesData holds bulk node deletion data with undo support.

type WALBulkEdgesData 

type WALBulkEdgesData struct {
	Edges []*Edge `json:"edges"`
	TxID  string  `json:"tx_id,omitempty"` // Transaction ID for grouping
}

WALBulkEdgesData holds bulk edge creation data.

type WALBulkNodesData 

type WALBulkNodesData struct {
	Nodes []*Node `json:"nodes"`
	TxID  string  `json:"tx_id,omitempty"` // Transaction ID for grouping
}

WALBulkNodesData holds bulk node creation data.

type WALConfig 

type WALConfig struct {
	// Directory for WAL files
	Dir string

	// SyncMode controls when writes are synced to disk
	// "immediate": fsync after each write (safest, slowest)
	// "batch": fsync periodically (faster, some risk)
	// "none": no fsync (fastest, data loss on crash)
	SyncMode string

	// BatchSyncInterval for "batch" sync mode
	BatchSyncInterval time.Duration

	// MaxFileSize triggers rotation when exceeded
	MaxFileSize int64

	// MaxEntries triggers rotation when exceeded
	MaxEntries int64

	// SnapshotInterval for automatic snapshots
	SnapshotInterval time.Duration

	// RetentionMaxSegments keeps at most N sealed segments (0 = unlimited).
	RetentionMaxSegments int

	// RetentionMaxAge keeps segments newer than this duration (0 = unlimited).
	RetentionMaxAge time.Duration

	// SnapshotRetentionMaxCount is the maximum number of snapshot files to keep in the
	// snapshot directory (0 = keep all). Oldest snapshots are deleted after each new one.
	// Recommended 3–5 so disk space stays bounded while keeping recovery options.
	SnapshotRetentionMaxCount int

	// SnapshotRetentionMaxAge is the maximum age of snapshot files to keep (0 = unlimited).
	// Snapshots older than this are deleted even if under MaxCount.
	SnapshotRetentionMaxAge time.Duration

	// Logger receives WAL diagnostics events (optional, legacy structured
	// channel implemented by pkg/storage's WALLogger interface). If nil,
	// a default slog-backed logger is installed at ctor entry — the
	// previous stdlib-printer-backed default has been removed per LOG-01.
	Logger WALLogger

	// SlogLogger is the structured *slog.Logger used by D-07 WAL recovery
	// emissions (subsystem=wal, subsystem=wal_recovery). Optional; nil
	// falls back to a discard handler at NewWAL entry per D-01a.
	SlogLogger *slog.Logger

	// OnCorruption is called when WAL corruption diagnostics are produced (optional).
	// This allows the server layer to surface "WAL degraded" health state without
	// parsing logs. The callback MUST be fast and non-blocking.
	OnCorruption func(diag *CorruptionDiagnostics, cause error)
}

WALConfig configures WAL behavior.

func DefaultWALConfig 

func DefaultWALConfig() *WALConfig

DefaultWALConfig returns sensible defaults.

type WALDeleteData 

type WALDeleteData struct {
	ID      string `json:"id"`
	OldNode *Node  `json:"old_node,omitempty"` // Complete node being deleted (undo)
	OldEdge *Edge  `json:"old_edge,omitempty"` // Complete edge being deleted (undo)
	TxID    string `json:"tx_id,omitempty"`    // Transaction ID for grouping
}

WALDeleteData holds delete operation data with undo support. For proper undo, we store the complete entity being deleted.

type WALEdgeData 

type WALEdgeData struct {
	Edge    *Edge  `json:"edge"`               // New state (redo)
	OldEdge *Edge  `json:"old_edge,omitempty"` // Previous state (undo) - for updates
	TxID    string `json:"tx_id,omitempty"`    // Transaction ID for grouping
}

WALEdgeData holds edge data for WAL entries with optional undo support. For update/delete operations, OldEdge contains the "before image" for rollback.

type WALEngine 

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

WALEngine wraps a storage engine with write-ahead logging.

All mutating operations are appended to the WAL before they are applied to the wrapped engine. This provides crash recovery via snapshot + replay while keeping the underlying Engine implementations simple and fast.

Design notes:

  • Database routing: WALEngine supports multi-database usage by recording the database/namespace in each WAL entry and by normalizing IDs when needed.
  • Embedding updates: embedding-only updates are logged using OpUpdateEmbedding which is safe to skip during recovery because embeddings are regenerable.
  • Auto-compaction: optional periodic snapshots + WAL truncation to prevent unbounded WAL growth.

func NewWALEngine 

func NewWALEngine(engine Engine, wal *WAL) *WALEngine

NewWALEngine creates a WAL-backed storage engine.

func (*WALEngine) AddToPendingEmbeddings 

func (w *WALEngine) AddToPendingEmbeddings(nodeID NodeID)

AddToPendingEmbeddings delegates to underlying engine if it supports it.

func (*WALEngine) AllEdges 

func (w *WALEngine) AllEdges() ([]*Edge, error)

AllEdges delegates to underlying engine.

func (*WALEngine) AllNodes 

func (w *WALEngine) AllNodes() ([]*Node, error)

AllNodes delegates to underlying engine.

func (*WALEngine) BatchGetNodes 

func (w *WALEngine) BatchGetNodes(ids []NodeID) (map[NodeID]*Node, error)

BatchGetNodes delegates to underlying engine.

func (*WALEngine) BulkCreateEdges 

func (w *WALEngine) BulkCreateEdges(edges []*Edge) error

BulkCreateEdges logs then executes bulk edge creation.

func (*WALEngine) BulkCreateNodes 

func (w *WALEngine) BulkCreateNodes(nodes []*Node) error

BulkCreateNodes logs then executes bulk node creation.

func (*WALEngine) BulkDeleteEdges 

func (w *WALEngine) BulkDeleteEdges(ids []EdgeID) error

BulkDeleteEdges logs then executes bulk edge deletion.

func (*WALEngine) BulkDeleteNodes 

func (w *WALEngine) BulkDeleteNodes(ids []NodeID) error

BulkDeleteNodes logs then executes bulk node deletion.

func (*WALEngine) Close 

func (w *WALEngine) Close() error

Close closes both the WAL and underlying engine.

func (*WALEngine) CreateEdge 

func (w *WALEngine) CreateEdge(edge *Edge) error

CreateEdge logs then executes edge creation.

func (*WALEngine) CreateNode 

func (w *WALEngine) CreateNode(node *Node) (NodeID, error)

CreateNode logs then executes node creation.

func (*WALEngine) DeleteByPrefix 

func (w *WALEngine) DeleteByPrefix(prefix string) (nodesDeleted int64, edgesDeleted int64, err error)

DeleteByPrefix delegates to the underlying engine.

func (*WALEngine) DeleteEdge 

func (w *WALEngine) DeleteEdge(id EdgeID) error

DeleteEdge logs then executes edge deletion.

func (*WALEngine) DeleteNode 

func (w *WALEngine) DeleteNode(id NodeID) error

DeleteNode logs then executes node deletion.

func (*WALEngine) DisableAutoCompaction 

func (w *WALEngine) DisableAutoCompaction()

DisableAutoCompaction stops automatic snapshot creation and WAL truncation. Waits for the auto-compaction goroutine to finish before returning to prevent race conditions when the engine is closed.

func (*WALEngine) EdgeCount 

func (w *WALEngine) EdgeCount() (int64, error)

EdgeCount delegates to underlying engine.

func (*WALEngine) EdgeCountByPrefix 

func (w *WALEngine) EdgeCountByPrefix(prefix string) (int64, error)

func (*WALEngine) EnableAutoCompaction 

func (w *WALEngine) EnableAutoCompaction(snapshotDir string) error

EnableAutoCompaction starts automatic snapshot creation and WAL truncation. Snapshots are created at the configured SnapshotInterval, and the WAL is truncated after each successful snapshot to prevent unbounded growth.

Snapshots are saved to snapshotDir/snapshot-<timestamp>.json

This solves the "WAL grows forever" problem by automatically removing old entries that are already captured in snapshots.

Example: 

walEngine.EnableAutoCompaction("data/snapshots")
// WAL will now be automatically truncated every SnapshotInterval

func (*WALEngine) FindNodeNeedingEmbedding 

func (w *WALEngine) FindNodeNeedingEmbedding() *Node

FindNodeNeedingEmbedding delegates to underlying engine if it supports it.

func (*WALEngine) ForEachNodeIDByLabel 

func (w *WALEngine) ForEachNodeIDByLabel(label string, visit func(NodeID) bool) error

ForEachNodeIDByLabel delegates label-to-nodeID iteration to the underlying engine when available. This keeps LIMIT + label paths fast without forcing full node materialization.

func (*WALEngine) GetAdjacentEdges added in v1.1.2 

func (w *WALEngine) GetAdjacentEdges(nodeID NodeID) ([]*Edge, []*Edge, error)

GetAdjacentEdges delegates to the inner engine when it implements the optional AdjacentEdgesEngine capability. WAL writes are never reflected in reads (it only logs), so a forwarded call sees the same data as the pair of single-direction methods.

func (*WALEngine) GetAllNodes 

func (w *WALEngine) GetAllNodes() []*Node

GetAllNodes delegates to underlying engine.

func (*WALEngine) GetEdge 

func (w *WALEngine) GetEdge(id EdgeID) (*Edge, error)

GetEdge delegates to underlying engine.

func (*WALEngine) GetEdgeBetween 

func (w *WALEngine) GetEdgeBetween(startID, endID NodeID, edgeType string) *Edge

GetEdgeBetween delegates to underlying engine.

func (*WALEngine) GetEdgeCurrentHead 

func (w *WALEngine) GetEdgeCurrentHead(id EdgeID) (MVCCHead, error)

GetEdgeCurrentHead delegates edge head lookup when supported.

func (*WALEngine) GetEdgeLatestEffective 

func (w *WALEngine) GetEdgeLatestEffective(id EdgeID) (*Edge, error)

GetEdgeLatestEffective delegates MVCC latest-effective edge reads to the wrapped engine when supported.

func (*WALEngine) GetEdgeLatestVisible 

func (w *WALEngine) GetEdgeLatestVisible(id EdgeID) (*Edge, error)

GetEdgeLatestVisible delegates latest-visible edge reads when supported.

func (*WALEngine) GetEdgeVisibleAt 

func (w *WALEngine) GetEdgeVisibleAt(id EdgeID, version MVCCVersion) (*Edge, error)

GetEdgeVisibleAt delegates snapshot-visible edge reads when supported.

func (*WALEngine) GetEdgesBetween 

func (w *WALEngine) GetEdgesBetween(startID, endID NodeID) ([]*Edge, error)

GetEdgesBetween delegates to underlying engine.

func (*WALEngine) GetEdgesBetweenVisibleAt 

func (w *WALEngine) GetEdgesBetweenVisibleAt(startID, endID NodeID, version MVCCVersion) ([]*Edge, error)

GetEdgesBetweenVisibleAt delegates snapshot-visible topology queries when supported.

func (*WALEngine) GetEdgesByType 

func (w *WALEngine) GetEdgesByType(edgeType string) ([]*Edge, error)

GetEdgesByType delegates to underlying engine.

func (*WALEngine) GetEdgesByTypeVisibleAt 

func (w *WALEngine) GetEdgesByTypeVisibleAt(edgeType string, version MVCCVersion) ([]*Edge, error)

GetEdgesByTypeVisibleAt delegates snapshot-visible edge-type queries when supported.

func (*WALEngine) GetEngine 

func (w *WALEngine) GetEngine() Engine

GetEngine returns the underlying engine.

func (*WALEngine) GetFirstNodeByLabel 

func (w *WALEngine) GetFirstNodeByLabel(label string) (*Node, error)

GetFirstNodeByLabel delegates to underlying engine.

func (*WALEngine) GetInDegree 

func (w *WALEngine) GetInDegree(nodeID NodeID) int

GetInDegree delegates to underlying engine.

func (*WALEngine) GetIncomingEdges 

func (w *WALEngine) GetIncomingEdges(nodeID NodeID) ([]*Edge, error)

GetIncomingEdges delegates to underlying engine.

func (*WALEngine) GetInnerEngine 

func (w *WALEngine) GetInnerEngine() Engine

GetInnerEngine returns the wrapped storage engine.

func (*WALEngine) GetNode 

func (w *WALEngine) GetNode(id NodeID) (*Node, error)

GetNode delegates to underlying engine.

func (*WALEngine) GetNodeCurrentHead 

func (w *WALEngine) GetNodeCurrentHead(id NodeID) (MVCCHead, error)

GetNodeCurrentHead delegates node head lookup when supported.

func (*WALEngine) GetNodeLatestEffective 

func (w *WALEngine) GetNodeLatestEffective(id NodeID) (*Node, error)

GetNodeLatestEffective delegates MVCC latest-effective reads to the wrapped engine when supported.

func (*WALEngine) GetNodeLatestVisible 

func (w *WALEngine) GetNodeLatestVisible(id NodeID) (*Node, error)

GetNodeLatestVisible delegates latest-visible node reads when supported.

func (*WALEngine) GetNodeVisibleAt 

func (w *WALEngine) GetNodeVisibleAt(id NodeID, version MVCCVersion) (*Node, error)

GetNodeVisibleAt delegates snapshot-visible node reads when supported.

func (*WALEngine) GetNodesByLabel 

func (w *WALEngine) GetNodesByLabel(label string) ([]*Node, error)

GetNodesByLabel delegates to underlying engine.

func (*WALEngine) GetNodesByLabelVisibleAt 

func (w *WALEngine) GetNodesByLabelVisibleAt(label string, version MVCCVersion) ([]*Node, error)

GetNodesByLabelVisibleAt delegates snapshot-visible label queries when supported.

func (*WALEngine) GetOutDegree 

func (w *WALEngine) GetOutDegree(nodeID NodeID) int

GetOutDegree delegates to underlying engine.

func (*WALEngine) GetOutgoingEdges 

func (w *WALEngine) GetOutgoingEdges(nodeID NodeID) ([]*Edge, error)

GetOutgoingEdges delegates to underlying engine.

func (*WALEngine) GetSchema 

func (w *WALEngine) GetSchema() *SchemaManager

GetSchema delegates to underlying engine.

func (*WALEngine) GetSchemaForNamespace 

func (w *WALEngine) GetSchemaForNamespace(namespace string) *SchemaManager

GetSchemaForNamespace implements NamespaceSchemaProvider when the underlying engine supports it.

func (*WALEngine) GetSnapshotStats 

func (w *WALEngine) GetSnapshotStats() (totalSnapshots int64, lastSnapshotTime time.Time)

GetSnapshotStats returns statistics about automatic snapshots.

func (*WALEngine) GetWAL 

func (w *WALEngine) GetWAL() *WAL

GetWAL returns the underlying WAL for direct access.

func (*WALEngine) IsCurrentTemporalNode 

func (w *WALEngine) IsCurrentTemporalNode(node *Node, asOf time.Time) (bool, error)

IsCurrentTemporalNode delegates current-version checks to the wrapped engine when supported.

func (*WALEngine) IterateNodes 

func (w *WALEngine) IterateNodes(fn func(*Node) bool) error

IterateNodes delegates to underlying engine if it supports streaming iteration.

func (*WALEngine) LastWriteTime 

func (w *WALEngine) LastWriteTime() time.Time

LastWriteTime returns the last WAL entry timestamp (best-effort).

func (*WALEngine) LifecycleStatus 

func (w *WALEngine) LifecycleStatus() map[string]interface{}

LifecycleStatus delegates lifecycle status when supported.

func (*WALEngine) ListNamespaces 

func (w *WALEngine) ListNamespaces() []string

ListNamespaces returns known namespaces from the wrapped engine, if supported.

func (*WALEngine) MarkNodeEmbedded 

func (w *WALEngine) MarkNodeEmbedded(nodeID NodeID)

MarkNodeEmbedded delegates to underlying engine if it supports it.

func (*WALEngine) NodeCount 

func (w *WALEngine) NodeCount() (int64, error)

NodeCount delegates to underlying engine.

func (*WALEngine) NodeCountByPrefix 

func (w *WALEngine) NodeCountByPrefix(prefix string) (int64, error)

func (*WALEngine) PauseLifecycle 

func (w *WALEngine) PauseLifecycle()

PauseLifecycle delegates lifecycle pause when supported.

func (*WALEngine) PendingEmbeddingsCount 

func (w *WALEngine) PendingEmbeddingsCount() int

PendingEmbeddingsCount delegates to underlying engine if it supports it.

func (*WALEngine) PruneMVCCVersions 

func (w *WALEngine) PruneMVCCVersions(ctx context.Context, opts MVCCPruneOptions) (int64, error)

PruneMVCCVersions delegates MVCC pruning to the wrapped engine when supported.

func (*WALEngine) PruneTemporalHistory 

func (w *WALEngine) PruneTemporalHistory(ctx context.Context, opts TemporalPruneOptions) (int64, error)

PruneTemporalHistory delegates temporal pruning to the wrapped engine when supported.

func (*WALEngine) RebuildMVCCHeads 

func (w *WALEngine) RebuildMVCCHeads(ctx context.Context) error

RebuildMVCCHeads delegates MVCC head rebuild to the wrapped engine when supported.

func (*WALEngine) RebuildTemporalIndexes 

func (w *WALEngine) RebuildTemporalIndexes(ctx context.Context) error

RebuildTemporalIndexes delegates temporal index rebuild to the wrapped engine when supported.

func (*WALEngine) RecordMaterializedAccess added in v1.1.0 

func (w *WALEngine) RecordMaterializedAccess(entityID string)

RecordMaterializedAccess delegates result-materialization access recording to the underlying engine, if supported.

func (*WALEngine) RefreshPendingEmbeddingsIndex 

func (w *WALEngine) RefreshPendingEmbeddingsIndex() int

RefreshPendingEmbeddingsIndex delegates to underlying engine if it supports it.

func (*WALEngine) RegisterSnapshotReader 

func (w *WALEngine) RegisterSnapshotReader(info SnapshotReaderInfo) func()

RegisterSnapshotReader delegates snapshot-reader registration when supported.

func (*WALEngine) ResumeLifecycle 

func (w *WALEngine) ResumeLifecycle()

ResumeLifecycle delegates lifecycle resume when supported.

func (*WALEngine) SetLifecycleSchedule 

func (w *WALEngine) SetLifecycleSchedule(interval time.Duration) error

SetLifecycleSchedule delegates lifecycle cadence updates when supported.

func (*WALEngine) StreamEdges 

func (w *WALEngine) StreamEdges(ctx context.Context, fn func(edge *Edge) error) error

StreamEdges implements StreamingEngine.StreamEdges by delegating to the underlying engine.

func (*WALEngine) StreamNodeChunks 

func (w *WALEngine) StreamNodeChunks(ctx context.Context, chunkSize int, fn func(nodes []*Node) error) error

StreamNodeChunks implements StreamingEngine.StreamNodeChunks by delegating to the underlying engine.

func (*WALEngine) StreamNodes 

func (w *WALEngine) StreamNodes(ctx context.Context, fn func(node *Node) error) error

StreamNodes implements StreamingEngine.StreamNodes by delegating to the underlying engine.

func (*WALEngine) StreamNodesByPrefix 

func (w *WALEngine) StreamNodesByPrefix(ctx context.Context, prefix string, fn func(node *Node) error) error

StreamNodesByPrefix implements PrefixStreamingEngine by delegating prefix-scoped iteration to the wrapped engine when available. This preserves namespace-aware early termination behavior for MATCH ... LIMIT hot paths.

func (*WALEngine) TopLifecycleDebtKeys 

func (w *WALEngine) TopLifecycleDebtKeys(limit int) []MVCCLifecycleDebtKey

TopLifecycleDebtKeys delegates lifecycle debt inspection when supported.

func (*WALEngine) TriggerPruneNow 

func (w *WALEngine) TriggerPruneNow(ctx context.Context) error

TriggerPruneNow delegates lifecycle prune-now when supported.

func (*WALEngine) UpdateEdge 

func (w *WALEngine) UpdateEdge(edge *Edge) error

UpdateEdge logs then executes edge update.

func (*WALEngine) UpdateNode 

func (w *WALEngine) UpdateNode(node *Node) error

UpdateNode logs then executes node update.

func (*WALEngine) UpdateNodeEmbedding 

func (w *WALEngine) UpdateNodeEmbedding(node *Node) error

UpdateNodeEmbedding logs then executes embedding-only node update. Uses OpUpdateEmbedding which is safe to skip during WAL recovery since embeddings can be regenerated automatically.

type WALEntry 

type WALEntry struct {
	Sequence  uint64        `json:"seq"`                // Monotonically increasing sequence number
	Timestamp time.Time     `json:"ts"`                 // When the operation occurred
	Operation OperationType `json:"op"`                 // Operation type (create_node, update_node, etc.)
	Data      []byte        `json:"data"`               // JSON-serialized operation data
	Checksum  uint32        `json:"checksum"`           // CRC32 checksum for integrity
	Database  string        `json:"database,omitempty"` // Database/namespace name (for multi-database support)
}

WALEntry represents a single write-ahead log entry. Each mutating operation is recorded as an entry before execution.

func FindWALEntriesByTxID 

func FindWALEntriesByTxID(walDir, txID string, maxEntries int) ([]WALEntry, error)

FindWALEntriesByTxID scans entries and returns those with a matching tx_id. Use maxEntries <= 0 to return all matches.

func ReadWALEntries 

func ReadWALEntries(walPath string) ([]WALEntry, error)

ReadWALEntries reads all entries from a WAL file. Supports both legacy JSON format and new atomic format with automatic detection. Returns error on corruption of critical entries (nodes, edges). Embedding updates are safe to skip as they can be regenerated.

Uses a discard *slog.Logger for recovery diagnostics. Callers that want structured visibility into partial-write detection / skipped embedding entries should use ReadWALEntriesWithLogger.

func ReadWALEntriesAfter 

func ReadWALEntriesAfter(walPath string, afterSeq uint64) ([]WALEntry, error)

ReadWALEntriesAfter reads entries after a given sequence number.

func ReadWALEntriesAfterFromDir 

func ReadWALEntriesAfterFromDir(walDir string, afterSeq uint64) ([]WALEntry, error)

ReadWALEntriesAfterFromDir reads WAL entries after a given sequence across all segments.

func ReadWALEntriesFromDir 

func ReadWALEntriesFromDir(walDir string) ([]WALEntry, error)

ReadWALEntriesFromDir reads WAL entries across all segments and the active WAL file.

func ReadWALEntriesRangeFromDir 

func ReadWALEntriesRangeFromDir(walDir string, fromSeq, toSeq uint64) ([]WALEntry, error)

ReadWALEntriesRangeFromDir reads entries in [fromSeq, toSeq] (inclusive).

func ReadWALEntriesWithLogger added in v1.1.0 

func ReadWALEntriesWithLogger(walPath string, logger *slog.Logger) ([]WALEntry, error)

ReadWALEntriesWithLogger is the slog-aware variant of ReadWALEntries. D-07: callers (notably WAL recovery in pkg/nornicdb/storage_recovery.go) thread a subsystem=wal_recovery logger so partial-write and corrupted- embedding diagnostics land in the structured log stream.

type WALIntegrityReport 

type WALIntegrityReport struct {
	Healthy           bool                    `json:"healthy"`
	TotalEntries      int                     `json:"total_entries"`
	ValidEntries      int                     `json:"valid_entries"`
	CorruptedEntries  int                     `json:"corrupted_entries"`
	SkippedEmbeddings int                     `json:"skipped_embeddings"`
	FirstSeq          uint64                  `json:"first_seq"`
	LastSeq           uint64                  `json:"last_seq"`
	FileSize          int64                   `json:"file_size"`
	Format            string                  `json:"format"` // "atomic" or "legacy"
	Errors            []string                `json:"errors,omitempty"`
	CorruptionDetails []CorruptionDiagnostics `json:"corruption_details,omitempty"`
}

WALIntegrityReport provides detailed integrity check results

func CheckWALIntegrity 

func CheckWALIntegrity(walPath string) (*WALIntegrityReport, error)

CheckWALIntegrity performs a comprehensive integrity check on a WAL file. This can be used for health checks, startup validation, or manual diagnostics. It does NOT modify the WAL - read-only operation.

type WALLogger 

type WALLogger interface {
	Log(level string, msg string, fields map[string]any)
}

WALLogger receives structured diagnostics emitted by WAL recovery / corruption handlers.

This is intentionally minimal to avoid coupling storage to a specific logging library. Implementations should treat fields as a stable machine-readable contract.

Phase 2 LOG-01 note: the legacy default implementation routed records through stdlib log printers. That implementation is gone — defaultWALLogger now wraps a *slog.Logger so all storage-package emissions flow through the production 4-layer slog handler stack (recovering → mandatory → redactor → JSON).

type WALManifest 

type WALManifest struct {
	Version  int          `json:"version"`
	Segments []WALSegment `json:"segments"`
}

WALManifest indexes all sealed WAL segments.

type WALNodeData 

type WALNodeData struct {
	Node    *Node  `json:"node"`               // New state (redo)
	OldNode *Node  `json:"old_node,omitempty"` // Previous state (undo) - for updates
	TxID    string `json:"tx_id,omitempty"`    // Transaction ID for grouping
}

WALNodeData holds node data for WAL entries with optional undo support. For update/delete operations, OldNode contains the "before image" for rollback.

type WALSegment 

type WALSegment struct {
	FirstSeq  uint64    `json:"first_seq"`
	LastSeq   uint64    `json:"last_seq"`
	SizeBytes int64     `json:"size_bytes"`
	CreatedAt time.Time `json:"created_at"`
	Path      string    `json:"path"`
}

WALSegment describes a sealed WAL segment stored on disk.

type WALStats 

type WALStats struct {
	Sequence      uint64
	EntryCount    int64
	BytesWritten  int64
	TotalWrites   int64
	TotalSyncs    int64
	LastSyncTime  time.Time
	LastEntryTime time.Time
	Closed        bool
}

WALStats provides observability into WAL state.

type WALTxData 

type WALTxData struct {
	TxID     string            `json:"tx_id"`              // Transaction identifier
	Metadata map[string]string `json:"metadata,omitempty"` // Optional transaction metadata
	Reason   string            `json:"reason,omitempty"`   // For abort: why was it aborted
	OpCount  int               `json:"op_count,omitempty"` // For commit: number of operations
}

WALTxData holds transaction boundary data.

Source Files

View all Source files

Directories

Path Synopsis

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.