jbergner
440423c5b6
All checks were successful
release-tag / release-image (push) Successful in 2m43s
895 lines
31 KiB
Go
895 lines
31 KiB
Go
package graph
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/binary"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/local/glpi-neural-brain/internal/model"
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
const schemaVersion = 3
|
|
|
|
const nodeUpsertSQL = `INSERT INTO nodes(id,kind,label,summary,status,origin,external_id,uri,categories_json,keywords_json,metadata_json,weight,x,y,z,updated_at_ns)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(id) DO UPDATE SET kind=excluded.kind,label=excluded.label,summary=excluded.summary,status=excluded.status,origin=excluded.origin,external_id=excluded.external_id,uri=excluded.uri,categories_json=excluded.categories_json,keywords_json=excluded.keywords_json,metadata_json=excluded.metadata_json,weight=excluded.weight,x=excluded.x,y=excluded.y,z=excluded.z,updated_at_ns=excluded.updated_at_ns`
|
|
|
|
const edgeUpsertSQL = `INSERT INTO edges(id,source,target,type,origin,status,confidence,weight,explanation,evidence_json,metadata_json,created_at_ns,updated_at_ns)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(id) DO UPDATE SET source=excluded.source,target=excluded.target,type=excluded.type,origin=excluded.origin,status=excluded.status,confidence=excluded.confidence,weight=excluded.weight,explanation=excluded.explanation,evidence_json=excluded.evidence_json,metadata_json=excluded.metadata_json,created_at_ns=excluded.created_at_ns,updated_at_ns=excluded.updated_at_ns`
|
|
|
|
const vectorUpsertSQL = `INSERT INTO vectors(node_id,dimensions,data,updated_at_ns) VALUES(?,?,?,?)
|
|
ON CONFLICT(node_id) DO UPDATE SET dimensions=excluded.dimensions,data=excluded.data,updated_at_ns=excluded.updated_at_ns`
|
|
|
|
// StorageStatus describes the persistent SQLite graph store. The live graph is
|
|
// still held in memory; only changed rows are written during a flush.
|
|
type StorageStatus struct {
|
|
Backend string `json:"backend"`
|
|
Path string `json:"path"`
|
|
SchemaVersion int `json:"schema_version"`
|
|
JournalMode string `json:"journal_mode"`
|
|
DatabaseBytes int64 `json:"database_bytes"`
|
|
WALBytes int64 `json:"wal_bytes"`
|
|
NodeRows int `json:"node_rows"`
|
|
EdgeRows int `json:"edge_rows"`
|
|
VectorRows int `json:"vector_rows"`
|
|
VectorBytes int64 `json:"vector_bytes"`
|
|
PersistedVersion uint64 `json:"persisted_version"`
|
|
EmbeddingModel string `json:"embedding_model,omitempty"`
|
|
EmbeddingDigest string `json:"embedding_digest,omitempty"`
|
|
LegacyJSONPresent bool `json:"legacy_json_present"`
|
|
PendingNodes int `json:"pending_nodes"`
|
|
PendingEdges int `json:"pending_edges"`
|
|
PendingVectors int `json:"pending_vectors"`
|
|
PendingDeletions int `json:"pending_deletions"`
|
|
PendingMetadata bool `json:"pending_metadata"`
|
|
}
|
|
|
|
func Open(dir string) (*Store, error) {
|
|
absoluteDir, err := filepath.Abs(dir)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("resolve graph data directory: %w", err)
|
|
}
|
|
if err := os.MkdirAll(absoluteDir, 0o750); err != nil {
|
|
return nil, fmt.Errorf("create graph data directory %q: %w", absoluteDir, err)
|
|
}
|
|
if err := ensureWritableDirectory(absoluteDir); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
path := filepath.Join(absoluteDir, "graph.db")
|
|
if info, statErr := os.Stat(path); statErr == nil && info.IsDir() {
|
|
return nil, fmt.Errorf("graph database path %q is a directory; remove or rename it", path)
|
|
} else if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
|
|
return nil, fmt.Errorf("inspect graph database %q: %w", path, statErr)
|
|
}
|
|
|
|
// Use the plain filename here instead of applying PRAGMAs through the DSN.
|
|
// This makes startup failures attributable to one concrete step and avoids
|
|
// asking the driver to reserve a large mmap/cache while the connection is
|
|
// being created. The application keeps the full graph in Go memory already,
|
|
// so an additional 64 MiB SQLite page cache and 256 MiB mmap are wasteful on
|
|
// small systems.
|
|
db, err := sql.Open("sqlite", path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create sqlite handle for %q: %w", path, err)
|
|
}
|
|
// The Store serializes persistence and reads the live graph from memory.
|
|
// One SQLite connection is sufficient and keeps all connection-local
|
|
// PRAGMAs deterministic.
|
|
db.SetMaxOpenConns(1)
|
|
db.SetMaxIdleConns(1)
|
|
db.SetConnMaxLifetime(0)
|
|
db.SetConnMaxIdleTime(0)
|
|
|
|
openCtx, openCancel := context.WithTimeout(context.Background(), 15*time.Second)
|
|
if err := db.PingContext(openCtx); err != nil {
|
|
openCancel()
|
|
db.Close()
|
|
return nil, fmt.Errorf("open sqlite database %q: %w", path, err)
|
|
}
|
|
journalMode, err := configureSQLite(openCtx, db)
|
|
openCancel()
|
|
if err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("configure sqlite database %q: %w", path, err)
|
|
}
|
|
|
|
s := &Store{
|
|
nodes: map[string]model.Node{},
|
|
edges: map[string]model.Edge{},
|
|
vectors: map[string][]float32{},
|
|
db: db,
|
|
dbPath: path,
|
|
journalMode: journalMode,
|
|
dirtyNodes: map[string]uint64{},
|
|
dirtyEdges: map[string]uint64{},
|
|
dirtyVectors: map[string]uint64{},
|
|
deletedNodes: map[string]uint64{},
|
|
deletedEdges: map[string]uint64{},
|
|
deletedVectors: map[string]uint64{},
|
|
}
|
|
loadCtx, loadCancel := context.WithTimeout(context.Background(), 10*time.Minute)
|
|
defer loadCancel()
|
|
if err := s.initSchema(loadCtx); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("initialize sqlite graph schema: %w", err)
|
|
}
|
|
analysisDB, err := openAnalysisSQLite(path)
|
|
if err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("open dedicated sqlite analysis writer: %w", err)
|
|
}
|
|
s.analysisDB = analysisDB
|
|
if err := s.load(loadCtx); err != nil {
|
|
analysisDB.Close()
|
|
db.Close()
|
|
return nil, fmt.Errorf("load sqlite graph state: %w", err)
|
|
}
|
|
s.initAnalysisWriter()
|
|
return s, nil
|
|
}
|
|
|
|
// openAnalysisSQLite gives the append-only analysis journal its own SQLite
|
|
// connection. The primary graph store intentionally uses MaxOpenConns(1), and
|
|
// sharing that single Go connection caused telemetry to time out behind long
|
|
// graph persistence operations even while WAL itself was healthy.
|
|
func openAnalysisSQLite(path string) (*sql.DB, error) {
|
|
db, err := sql.Open("sqlite", path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db.SetMaxOpenConns(1)
|
|
db.SetMaxIdleConns(1)
|
|
db.SetConnMaxLifetime(0)
|
|
db.SetConnMaxIdleTime(0)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
|
defer cancel()
|
|
if err := db.PingContext(ctx); err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
for _, statement := range []string{
|
|
`PRAGMA busy_timeout=60000`,
|
|
`PRAGMA foreign_keys=ON`,
|
|
`PRAGMA synchronous=NORMAL`,
|
|
`PRAGMA temp_store=FILE`,
|
|
`PRAGMA cache_size=-2048`,
|
|
`PRAGMA mmap_size=0`,
|
|
} {
|
|
if _, err := db.ExecContext(ctx, statement); err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
}
|
|
var mode string
|
|
if err := db.QueryRowContext(ctx, `PRAGMA journal_mode`).Scan(&mode); err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
if strings.ToLower(strings.TrimSpace(mode)) != "wal" {
|
|
db.Close()
|
|
return nil, fmt.Errorf("analysis connection expected WAL mode, got %q", mode)
|
|
}
|
|
return db, nil
|
|
}
|
|
|
|
func ensureWritableDirectory(dir string) error {
|
|
info, err := os.Stat(dir)
|
|
if err != nil {
|
|
return fmt.Errorf("inspect graph data directory %q: %w", dir, err)
|
|
}
|
|
if !info.IsDir() {
|
|
return fmt.Errorf("graph data path %q is not a directory", dir)
|
|
}
|
|
probe, err := os.CreateTemp(dir, ".brain-write-test-*")
|
|
if err != nil {
|
|
return fmt.Errorf("graph data directory %q is not writable by the brain process: %w", dir, err)
|
|
}
|
|
name := probe.Name()
|
|
closeErr := probe.Close()
|
|
removeErr := os.Remove(name)
|
|
if closeErr != nil {
|
|
return fmt.Errorf("close graph data write test %q: %w", name, closeErr)
|
|
}
|
|
if removeErr != nil {
|
|
return fmt.Errorf("remove graph data write test %q: %w", name, removeErr)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func configureSQLite(ctx context.Context, db *sql.DB) (string, error) {
|
|
// Conservative defaults: the live graph and vectors already reside in Go
|
|
// memory. SQLite therefore only needs a small page cache and no mmap window.
|
|
// This keeps startup predictable inside memory-limited containers.
|
|
statements := []struct {
|
|
name string
|
|
sql string
|
|
}{
|
|
{name: "busy_timeout", sql: `PRAGMA busy_timeout=5000`},
|
|
{name: "foreign_keys", sql: `PRAGMA foreign_keys=ON`},
|
|
{name: "synchronous", sql: `PRAGMA synchronous=NORMAL`},
|
|
{name: "temp_store", sql: `PRAGMA temp_store=FILE`},
|
|
{name: "cache_size", sql: `PRAGMA cache_size=-8192`},
|
|
{name: "mmap_size", sql: `PRAGMA mmap_size=0`},
|
|
{name: "wal_autocheckpoint", sql: `PRAGMA wal_autocheckpoint=0`},
|
|
{name: "journal_size_limit", sql: `PRAGMA journal_size_limit=67108864`},
|
|
}
|
|
for _, statement := range statements {
|
|
if _, err := db.ExecContext(ctx, statement.sql); err != nil {
|
|
return "", fmt.Errorf("apply PRAGMA %s: %w", statement.name, err)
|
|
}
|
|
}
|
|
|
|
var journalMode string
|
|
if err := db.QueryRowContext(ctx, `PRAGMA journal_mode=WAL`).Scan(&journalMode); err != nil {
|
|
return "", fmt.Errorf("enable WAL journal mode: %w", err)
|
|
}
|
|
journalMode = strings.ToLower(strings.TrimSpace(journalMode))
|
|
if journalMode != "wal" {
|
|
return "", fmt.Errorf("enable WAL journal mode: SQLite selected %q", journalMode)
|
|
}
|
|
return journalMode, nil
|
|
}
|
|
|
|
func (s *Store) initSchema(ctx context.Context) error {
|
|
statements := []string{
|
|
`CREATE TABLE IF NOT EXISTS graph_meta (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
) WITHOUT ROWID`,
|
|
`CREATE TABLE IF NOT EXISTS nodes (
|
|
id TEXT PRIMARY KEY,
|
|
kind TEXT NOT NULL,
|
|
label TEXT NOT NULL,
|
|
summary TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL DEFAULT '',
|
|
origin TEXT NOT NULL,
|
|
external_id TEXT NOT NULL DEFAULT '',
|
|
uri TEXT NOT NULL DEFAULT '',
|
|
categories_json TEXT NOT NULL DEFAULT '[]',
|
|
keywords_json TEXT NOT NULL DEFAULT '[]',
|
|
metadata_json TEXT NOT NULL DEFAULT '{}',
|
|
weight REAL NOT NULL DEFAULT 1,
|
|
x REAL NOT NULL DEFAULT 0,
|
|
y REAL NOT NULL DEFAULT 0,
|
|
z REAL NOT NULL DEFAULT 0,
|
|
updated_at_ns INTEGER NOT NULL
|
|
) WITHOUT ROWID`,
|
|
`CREATE INDEX IF NOT EXISTS idx_nodes_origin ON nodes(origin)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_nodes_kind_status ON nodes(kind, status)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_nodes_external_id ON nodes(external_id)`,
|
|
`CREATE TABLE IF NOT EXISTS edges (
|
|
id TEXT PRIMARY KEY,
|
|
source TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
|
|
target TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
|
|
type TEXT NOT NULL,
|
|
origin TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT '',
|
|
confidence REAL NOT NULL DEFAULT 0,
|
|
weight REAL NOT NULL DEFAULT 1,
|
|
explanation TEXT NOT NULL DEFAULT '',
|
|
evidence_json TEXT NOT NULL DEFAULT '[]',
|
|
metadata_json TEXT NOT NULL DEFAULT '{}',
|
|
created_at_ns INTEGER NOT NULL,
|
|
updated_at_ns INTEGER NOT NULL
|
|
) WITHOUT ROWID`,
|
|
`CREATE INDEX IF NOT EXISTS idx_edges_source ON edges(source)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_edges_target ON edges(target)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_edges_origin_status ON edges(origin, status)`,
|
|
`CREATE TABLE IF NOT EXISTS vectors (
|
|
node_id TEXT PRIMARY KEY REFERENCES nodes(id) ON DELETE CASCADE,
|
|
dimensions INTEGER NOT NULL,
|
|
data BLOB NOT NULL,
|
|
updated_at_ns INTEGER NOT NULL
|
|
) WITHOUT ROWID`,
|
|
`CREATE TABLE IF NOT EXISTS research_tasks (
|
|
id TEXT PRIMARY KEY,
|
|
dedupe_key TEXT NOT NULL DEFAULT '',
|
|
topic TEXT NOT NULL,
|
|
reason TEXT NOT NULL DEFAULT '',
|
|
requested_by TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL,
|
|
priority REAL NOT NULL DEFAULT 0,
|
|
seed_node_ids_json TEXT NOT NULL DEFAULT '[]',
|
|
questions_json TEXT NOT NULL DEFAULT '[]',
|
|
queries_de_json TEXT NOT NULL DEFAULT '[]',
|
|
queries_en_json TEXT NOT NULL DEFAULT '[]',
|
|
attempts INTEGER NOT NULL DEFAULT 0,
|
|
max_attempts INTEGER NOT NULL DEFAULT 3,
|
|
evidence_count INTEGER NOT NULL DEFAULT 0,
|
|
article_created INTEGER NOT NULL DEFAULT 0,
|
|
article_title TEXT NOT NULL DEFAULT '',
|
|
article_path TEXT NOT NULL DEFAULT '',
|
|
outcome TEXT NOT NULL DEFAULT '',
|
|
last_error TEXT NOT NULL DEFAULT '',
|
|
metadata_json TEXT NOT NULL DEFAULT '{}',
|
|
created_at_ns INTEGER NOT NULL,
|
|
updated_at_ns INTEGER NOT NULL,
|
|
available_at_ns INTEGER NOT NULL,
|
|
lease_until_ns INTEGER NOT NULL DEFAULT 0,
|
|
started_at_ns INTEGER NOT NULL DEFAULT 0,
|
|
completed_at_ns INTEGER NOT NULL DEFAULT 0
|
|
) WITHOUT ROWID`,
|
|
`CREATE INDEX IF NOT EXISTS idx_research_tasks_status_priority ON research_tasks(status, priority DESC, available_at_ns, created_at_ns)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_research_tasks_dedupe_updated ON research_tasks(dedupe_key, updated_at_ns DESC)`,
|
|
`CREATE TABLE IF NOT EXISTS research_task_attempts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
task_id TEXT NOT NULL REFERENCES research_tasks(id) ON DELETE CASCADE,
|
|
attempt INTEGER NOT NULL,
|
|
status TEXT NOT NULL,
|
|
message TEXT NOT NULL DEFAULT '',
|
|
metadata_json TEXT NOT NULL DEFAULT '{}',
|
|
created_at_ns INTEGER NOT NULL
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_research_task_attempts_task ON research_task_attempts(task_id, attempt, created_at_ns)`,
|
|
`CREATE TABLE IF NOT EXISTS analysis_events (
|
|
id TEXT PRIMARY KEY,
|
|
type TEXT NOT NULL,
|
|
source TEXT NOT NULL DEFAULT '',
|
|
phase TEXT NOT NULL DEFAULT '',
|
|
query TEXT NOT NULL DEFAULT '',
|
|
message TEXT NOT NULL DEFAULT '',
|
|
node_ids_json TEXT NOT NULL DEFAULT '[]',
|
|
edge_ids_json TEXT NOT NULL DEFAULT '[]',
|
|
strength REAL NOT NULL DEFAULT 0,
|
|
metadata_json TEXT NOT NULL DEFAULT '{}',
|
|
timestamp_ns INTEGER NOT NULL,
|
|
process_id TEXT NOT NULL DEFAULT ''
|
|
) WITHOUT ROWID`,
|
|
`CREATE INDEX IF NOT EXISTS idx_analysis_events_timestamp ON analysis_events(timestamp_ns DESC)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_analysis_events_type_timestamp ON analysis_events(type, timestamp_ns DESC)`,
|
|
`CREATE TABLE IF NOT EXISTS analysis_points (
|
|
event_id TEXT PRIMARY KEY REFERENCES analysis_events(id) ON DELETE CASCADE,
|
|
timestamp_ns INTEGER NOT NULL,
|
|
process_id TEXT NOT NULL DEFAULT '',
|
|
graph_version INTEGER NOT NULL DEFAULT 0,
|
|
node_count INTEGER NOT NULL DEFAULT 0,
|
|
edge_count INTEGER NOT NULL DEFAULT 0,
|
|
vector_count INTEGER NOT NULL DEFAULT 0,
|
|
node_created INTEGER NOT NULL DEFAULT 0,
|
|
node_updated INTEGER NOT NULL DEFAULT 0,
|
|
node_deleted INTEGER NOT NULL DEFAULT 0,
|
|
edge_created INTEGER NOT NULL DEFAULT 0,
|
|
edge_updated INTEGER NOT NULL DEFAULT 0,
|
|
edge_deleted INTEGER NOT NULL DEFAULT 0,
|
|
vector_created INTEGER NOT NULL DEFAULT 0,
|
|
vector_updated INTEGER NOT NULL DEFAULT 0,
|
|
vector_deleted INTEGER NOT NULL DEFAULT 0,
|
|
delta_node_created INTEGER NOT NULL DEFAULT 0,
|
|
delta_node_updated INTEGER NOT NULL DEFAULT 0,
|
|
delta_node_deleted INTEGER NOT NULL DEFAULT 0,
|
|
delta_edge_created INTEGER NOT NULL DEFAULT 0,
|
|
delta_edge_updated INTEGER NOT NULL DEFAULT 0,
|
|
delta_edge_deleted INTEGER NOT NULL DEFAULT 0,
|
|
delta_vector_created INTEGER NOT NULL DEFAULT 0,
|
|
delta_vector_updated INTEGER NOT NULL DEFAULT 0,
|
|
delta_vector_deleted INTEGER NOT NULL DEFAULT 0,
|
|
change_count INTEGER NOT NULL DEFAULT 0,
|
|
changes_truncated INTEGER NOT NULL DEFAULT 0
|
|
) WITHOUT ROWID`,
|
|
`CREATE INDEX IF NOT EXISTS idx_analysis_points_timestamp ON analysis_points(timestamp_ns DESC)`,
|
|
`CREATE TABLE IF NOT EXISTS analysis_changes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id TEXT NOT NULL REFERENCES analysis_events(id) ON DELETE CASCADE,
|
|
timestamp_ns INTEGER NOT NULL,
|
|
process_id TEXT NOT NULL DEFAULT '',
|
|
graph_version INTEGER NOT NULL DEFAULT 0,
|
|
entity_kind TEXT NOT NULL,
|
|
action TEXT NOT NULL,
|
|
entity_id TEXT NOT NULL,
|
|
label TEXT NOT NULL DEFAULT '',
|
|
relation_type TEXT NOT NULL DEFAULT '',
|
|
origin TEXT NOT NULL DEFAULT '',
|
|
details_json TEXT NOT NULL DEFAULT '{}'
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_analysis_changes_timestamp ON analysis_changes(timestamp_ns DESC)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_analysis_changes_event ON analysis_changes(event_id, id)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_analysis_changes_entity ON analysis_changes(entity_kind, action, timestamp_ns DESC)`,
|
|
}
|
|
for _, statement := range statements {
|
|
if _, err := s.db.ExecContext(ctx, statement); err != nil {
|
|
return fmt.Errorf("initialize graph sqlite schema: %w", err)
|
|
}
|
|
}
|
|
var current int
|
|
err := s.db.QueryRowContext(ctx, `SELECT CAST(value AS INTEGER) FROM graph_meta WHERE key='schema_version'`).Scan(¤t)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
_, err = s.db.ExecContext(ctx, `INSERT INTO graph_meta(key,value) VALUES('schema_version', ?)`, schemaVersion)
|
|
return err
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if current > schemaVersion {
|
|
return fmt.Errorf("unsupported graph database schema %d (expected at most %d)", current, schemaVersion)
|
|
}
|
|
if current < schemaVersion {
|
|
if current < 1 || current > 2 {
|
|
return fmt.Errorf("unsupported graph database schema migration %d -> %d", current, schemaVersion)
|
|
}
|
|
if _, err := s.db.ExecContext(ctx, `UPDATE graph_meta SET value=? WHERE key='schema_version'`, schemaVersion); err != nil {
|
|
return fmt.Errorf("record graph schema migration: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) load(ctx context.Context) error {
|
|
if err := s.loadNodes(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := s.loadEdges(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := s.loadVectors(ctx); err != nil {
|
|
return err
|
|
}
|
|
var version uint64
|
|
if err := s.db.QueryRowContext(ctx, `SELECT CAST(value AS INTEGER) FROM graph_meta WHERE key='graph_version'`).Scan(&version); err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return err
|
|
}
|
|
var embeddingModel string
|
|
if err := s.db.QueryRowContext(ctx, `SELECT value FROM graph_meta WHERE key='embedding_model'`).Scan(&embeddingModel); err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return err
|
|
}
|
|
var embeddingDigest string
|
|
if err := s.db.QueryRowContext(ctx, `SELECT value FROM graph_meta WHERE key='embedding_digest'`).Scan(&embeddingDigest); err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return err
|
|
}
|
|
s.version = version
|
|
s.persistedVersion = version
|
|
s.embeddingModel = embeddingModel
|
|
s.embeddingDigest = embeddingDigest
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) loadNodes(ctx context.Context) error {
|
|
rows, err := s.db.QueryContext(ctx, `SELECT id,kind,label,summary,status,origin,external_id,uri,categories_json,keywords_json,metadata_json,weight,x,y,z,updated_at_ns FROM nodes`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var n model.Node
|
|
var categories, keywords, metadata string
|
|
var updated int64
|
|
if err := rows.Scan(&n.ID, &n.Kind, &n.Label, &n.Summary, &n.Status, &n.Origin, &n.ExternalID, &n.URI, &categories, &keywords, &metadata, &n.Weight, &n.X, &n.Y, &n.Z, &updated); err != nil {
|
|
return err
|
|
}
|
|
if err := decodeJSON(categories, &n.Categories); err != nil {
|
|
return fmt.Errorf("decode node %s categories: %w", n.ID, err)
|
|
}
|
|
if err := decodeJSON(keywords, &n.Keywords); err != nil {
|
|
return fmt.Errorf("decode node %s keywords: %w", n.ID, err)
|
|
}
|
|
if err := decodeJSON(metadata, &n.Metadata); err != nil {
|
|
return fmt.Errorf("decode node %s metadata: %w", n.ID, err)
|
|
}
|
|
if n.Metadata == nil {
|
|
n.Metadata = map[string]any{}
|
|
}
|
|
n.UpdatedAt = time.Unix(0, updated).UTC()
|
|
s.nodes[n.ID] = n
|
|
}
|
|
return rows.Err()
|
|
}
|
|
|
|
func (s *Store) loadEdges(ctx context.Context) error {
|
|
rows, err := s.db.QueryContext(ctx, `SELECT id,source,target,type,origin,status,confidence,weight,explanation,evidence_json,metadata_json,created_at_ns,updated_at_ns FROM edges`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var e model.Edge
|
|
var evidence, metadata string
|
|
var created, updated int64
|
|
if err := rows.Scan(&e.ID, &e.Source, &e.Target, &e.Type, &e.Origin, &e.Status, &e.Confidence, &e.Weight, &e.Explanation, &evidence, &metadata, &created, &updated); err != nil {
|
|
return err
|
|
}
|
|
if err := decodeJSON(evidence, &e.Evidence); err != nil {
|
|
return fmt.Errorf("decode edge %s evidence: %w", e.ID, err)
|
|
}
|
|
if err := decodeJSON(metadata, &e.Metadata); err != nil {
|
|
return fmt.Errorf("decode edge %s metadata: %w", e.ID, err)
|
|
}
|
|
if e.Metadata == nil {
|
|
e.Metadata = map[string]any{}
|
|
}
|
|
e.CreatedAt = time.Unix(0, created).UTC()
|
|
e.UpdatedAt = time.Unix(0, updated).UTC()
|
|
s.edges[e.ID] = e
|
|
}
|
|
return rows.Err()
|
|
}
|
|
|
|
func (s *Store) loadVectors(ctx context.Context) error {
|
|
rows, err := s.db.QueryContext(ctx, `SELECT node_id,dimensions,data FROM vectors`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var id string
|
|
var dimensions int
|
|
var data []byte
|
|
if err := rows.Scan(&id, &dimensions, &data); err != nil {
|
|
return err
|
|
}
|
|
v, err := decodeVector(data, dimensions)
|
|
if err != nil {
|
|
return fmt.Errorf("decode vector %s: %w", id, err)
|
|
}
|
|
s.vectors[id] = v
|
|
}
|
|
return rows.Err()
|
|
}
|
|
|
|
func decodeJSON(raw string, target any) error {
|
|
if strings.TrimSpace(raw) == "" {
|
|
return nil
|
|
}
|
|
dec := json.NewDecoder(strings.NewReader(raw))
|
|
dec.UseNumber()
|
|
return dec.Decode(target)
|
|
}
|
|
|
|
func encodeJSON(value any, empty string) (string, error) {
|
|
if value == nil {
|
|
return empty, nil
|
|
}
|
|
b, err := json.Marshal(value)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return string(b), nil
|
|
}
|
|
|
|
func encodeVector(v []float32) []byte {
|
|
out := make([]byte, len(v)*4)
|
|
for i, value := range v {
|
|
binary.LittleEndian.PutUint32(out[i*4:], math.Float32bits(value))
|
|
}
|
|
return out
|
|
}
|
|
|
|
func decodeVector(data []byte, dimensions int) ([]float32, error) {
|
|
if dimensions < 0 || len(data) != dimensions*4 {
|
|
return nil, fmt.Errorf("invalid float32 vector blob: dimensions=%d bytes=%d", dimensions, len(data))
|
|
}
|
|
out := make([]float32, dimensions)
|
|
for i := range out {
|
|
out[i] = math.Float32frombits(binary.LittleEndian.Uint32(data[i*4:]))
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// PersistVersion writes only records changed since the previous successful
|
|
// flush. Vectors are encoded as little-endian float32 BLOBs.
|
|
func (s *Store) PersistVersion() (uint64, error) {
|
|
return s.PersistVersionContext(context.Background())
|
|
}
|
|
|
|
func (s *Store) PersistVersionContext(ctx context.Context) (uint64, error) {
|
|
type nodeChange struct {
|
|
gen uint64
|
|
node model.Node
|
|
}
|
|
type edgeChange struct {
|
|
gen uint64
|
|
edge model.Edge
|
|
}
|
|
type vectorChange struct {
|
|
gen uint64
|
|
vector []float32
|
|
}
|
|
|
|
s.mu.RLock()
|
|
version := s.version
|
|
persistedVersion := s.persistedVersion
|
|
embeddingModel := s.embeddingModel
|
|
embeddingDigest := s.embeddingDigest
|
|
embeddingMetaGeneration := s.embeddingMetaGeneration
|
|
nodes := make(map[string]nodeChange, len(s.dirtyNodes))
|
|
for id, gen := range s.dirtyNodes {
|
|
if n, ok := s.nodes[id]; ok {
|
|
nodes[id] = nodeChange{gen: gen, node: n}
|
|
}
|
|
}
|
|
edges := make(map[string]edgeChange, len(s.dirtyEdges))
|
|
for id, gen := range s.dirtyEdges {
|
|
if e, ok := s.edges[id]; ok {
|
|
edges[id] = edgeChange{gen: gen, edge: e}
|
|
}
|
|
}
|
|
vectors := make(map[string]vectorChange, len(s.dirtyVectors))
|
|
for id, gen := range s.dirtyVectors {
|
|
if v, ok := s.vectors[id]; ok {
|
|
vectors[id] = vectorChange{gen: gen, vector: append([]float32(nil), v...)}
|
|
}
|
|
}
|
|
deletedNodes := cloneGenerations(s.deletedNodes)
|
|
deletedEdges := cloneGenerations(s.deletedEdges)
|
|
deletedVectors := cloneGenerations(s.deletedVectors)
|
|
s.mu.RUnlock()
|
|
|
|
if len(nodes)+len(edges)+len(vectors)+len(deletedNodes)+len(deletedEdges)+len(deletedVectors) == 0 && embeddingMetaGeneration == 0 && version == persistedVersion {
|
|
return version, nil
|
|
}
|
|
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
for id := range deletedEdges {
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM edges WHERE id=?`, id); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
for id := range deletedNodes {
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM nodes WHERE id=?`, id); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
for id := range deletedVectors {
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM vectors WHERE node_id=?`, id); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
|
|
if len(nodes) > 0 {
|
|
statement, err := tx.PrepareContext(ctx, nodeUpsertSQL)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
for _, change := range nodes {
|
|
if err := upsertNode(ctx, statement, change.node); err != nil {
|
|
_ = statement.Close()
|
|
return 0, err
|
|
}
|
|
}
|
|
if err := statement.Close(); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
if len(edges) > 0 {
|
|
statement, err := tx.PrepareContext(ctx, edgeUpsertSQL)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
for _, change := range edges {
|
|
if err := upsertEdge(ctx, statement, change.edge); err != nil {
|
|
_ = statement.Close()
|
|
return 0, err
|
|
}
|
|
}
|
|
if err := statement.Close(); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
if len(vectors) > 0 {
|
|
statement, err := tx.PrepareContext(ctx, vectorUpsertSQL)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
now := time.Now().UTC().UnixNano()
|
|
for id, change := range vectors {
|
|
if _, err := statement.ExecContext(ctx, id, len(change.vector), encodeVector(change.vector), now); err != nil {
|
|
_ = statement.Close()
|
|
return 0, err
|
|
}
|
|
}
|
|
if err := statement.Close(); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
if embeddingMetaGeneration > 0 {
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO graph_meta(key,value) VALUES('embedding_model',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, embeddingModel); err != nil {
|
|
return 0, err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO graph_meta(key,value) VALUES('embedding_digest',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, embeddingDigest); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO graph_meta(key,value) VALUES('graph_version',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, version); err != nil {
|
|
return 0, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
s.mu.Lock()
|
|
for id, change := range nodes {
|
|
if s.dirtyNodes[id] == change.gen {
|
|
delete(s.dirtyNodes, id)
|
|
}
|
|
}
|
|
for id, change := range edges {
|
|
if s.dirtyEdges[id] == change.gen {
|
|
delete(s.dirtyEdges, id)
|
|
}
|
|
}
|
|
for id, change := range vectors {
|
|
if s.dirtyVectors[id] == change.gen {
|
|
delete(s.dirtyVectors, id)
|
|
}
|
|
}
|
|
clearDeletedIfSame(s.deletedNodes, deletedNodes)
|
|
clearDeletedIfSame(s.deletedEdges, deletedEdges)
|
|
clearDeletedIfSame(s.deletedVectors, deletedVectors)
|
|
if s.embeddingMetaGeneration == embeddingMetaGeneration {
|
|
s.embeddingMetaGeneration = 0
|
|
}
|
|
if version > s.persistedVersion {
|
|
s.persistedVersion = version
|
|
}
|
|
s.mu.Unlock()
|
|
|
|
// Keep the WAL bounded. PASSIVE never blocks readers and leaves busy pages
|
|
// for the next scheduled flush.
|
|
_, _ = s.db.ExecContext(ctx, `PRAGMA wal_checkpoint(PASSIVE)`)
|
|
return version, nil
|
|
}
|
|
|
|
func upsertNode(ctx context.Context, statement *sql.Stmt, n model.Node) error {
|
|
categories, err := encodeJSON(n.Categories, "[]")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
keywords, err := encodeJSON(n.Keywords, "[]")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
metadata, err := encodeJSON(n.Metadata, "{}")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = statement.ExecContext(ctx, n.ID, n.Kind, n.Label, n.Summary, n.Status, n.Origin, n.ExternalID, n.URI, categories, keywords, metadata, n.Weight, n.X, n.Y, n.Z, n.UpdatedAt.UnixNano())
|
|
return err
|
|
}
|
|
|
|
func upsertEdge(ctx context.Context, statement *sql.Stmt, e model.Edge) error {
|
|
evidence, err := encodeJSON(e.Evidence, "[]")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
metadata, err := encodeJSON(e.Metadata, "{}")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = statement.ExecContext(ctx, e.ID, e.Source, e.Target, e.Type, e.Origin, e.Status, e.Confidence, e.Weight, e.Explanation, evidence, metadata, e.CreatedAt.UnixNano(), e.UpdatedAt.UnixNano())
|
|
return err
|
|
}
|
|
|
|
func cloneGenerations(in map[string]uint64) map[string]uint64 {
|
|
out := make(map[string]uint64, len(in))
|
|
for id, gen := range in {
|
|
out[id] = gen
|
|
}
|
|
return out
|
|
}
|
|
|
|
func clearDeletedIfSame(current, snapshot map[string]uint64) {
|
|
for id, gen := range snapshot {
|
|
if current[id] == gen {
|
|
delete(current, id)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Store) Persist() error {
|
|
_, err := s.PersistVersion()
|
|
return err
|
|
}
|
|
|
|
func walCheckpointNeeded(dirty bool, walBytes, minBytes int64) bool {
|
|
return !dirty && minBytes > 0 && walBytes >= minBytes
|
|
}
|
|
|
|
func (s *Store) CheckpointIfWALLarge(ctx context.Context, minBytes int64) (bool, error) {
|
|
if s == nil || s.db == nil || minBytes <= 0 {
|
|
return false, nil
|
|
}
|
|
// Never force a truncating checkpoint while graph mutations are still
|
|
// waiting to be persisted. A concurrent writer may still make SQLite return
|
|
// BUSY; that is treated as a harmless retry-later condition below.
|
|
status := s.StorageStatus()
|
|
if !walCheckpointNeeded(s.Dirty(), status.WALBytes, minBytes) {
|
|
return false, nil
|
|
}
|
|
var busy, logFrames, checkpointedFrames int
|
|
if err := s.db.QueryRowContext(ctx, `PRAGMA wal_checkpoint(TRUNCATE)`).Scan(&busy, &logFrames, &checkpointedFrames); err != nil {
|
|
return false, err
|
|
}
|
|
if busy != 0 {
|
|
return false, nil
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func (s *Store) Checkpoint(ctx context.Context, truncate bool) error {
|
|
mode := "PASSIVE"
|
|
if truncate {
|
|
mode = "TRUNCATE"
|
|
}
|
|
_, err := s.db.ExecContext(ctx, `PRAGMA wal_checkpoint(`+mode+`)`)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) Close() error {
|
|
if s.db == nil {
|
|
return nil
|
|
}
|
|
s.closeAnalysisWriter()
|
|
var analysisCloseErr error
|
|
if s.analysisDB != nil {
|
|
analysisCloseErr = s.analysisDB.Close()
|
|
s.analysisDB = nil
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
_, persistErr := s.PersistVersionContext(ctx)
|
|
checkpointErr := s.Checkpoint(ctx, true)
|
|
closeErr := s.db.Close()
|
|
return errors.Join(persistErr, checkpointErr, analysisCloseErr, closeErr)
|
|
}
|
|
|
|
func (s *Store) StorageStatus() StorageStatus {
|
|
s.mu.RLock()
|
|
var vectorBytes int64
|
|
for _, vector := range s.vectors {
|
|
vectorBytes += int64(len(vector) * 4)
|
|
}
|
|
status := StorageStatus{
|
|
Backend: "sqlite", Path: s.dbPath, SchemaVersion: schemaVersion, JournalMode: s.journalMode,
|
|
NodeRows: len(s.nodes), EdgeRows: len(s.edges), VectorRows: len(s.vectors), VectorBytes: vectorBytes,
|
|
PersistedVersion: s.persistedVersion,
|
|
EmbeddingModel: s.embeddingModel,
|
|
EmbeddingDigest: s.embeddingDigest,
|
|
PendingNodes: len(s.dirtyNodes), PendingEdges: len(s.dirtyEdges), PendingVectors: len(s.dirtyVectors),
|
|
PendingDeletions: len(s.deletedNodes) + len(s.deletedEdges) + len(s.deletedVectors),
|
|
PendingMetadata: s.embeddingMetaGeneration > 0,
|
|
}
|
|
s.mu.RUnlock()
|
|
if info, err := os.Stat(s.dbPath); err == nil {
|
|
status.DatabaseBytes = info.Size()
|
|
}
|
|
if info, err := os.Stat(s.dbPath + "-wal"); err == nil {
|
|
status.WALBytes = info.Size()
|
|
}
|
|
_, err := os.Stat(filepath.Join(filepath.Dir(s.dbPath), "graph-state.json"))
|
|
status.LegacyJSONPresent = err == nil
|
|
return status
|
|
}
|
|
|
|
// Export creates a compact, transactionally consistent SQLite copy. It can be
|
|
// moved to another installation with the same knowledge-base identifiers.
|
|
func (s *Store) Export(ctx context.Context, destination string) error {
|
|
if _, err := s.PersistVersionContext(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := os.MkdirAll(filepath.Dir(destination), 0o750); err != nil {
|
|
return err
|
|
}
|
|
if err := os.Remove(destination); err != nil && !errors.Is(err, os.ErrNotExist) {
|
|
return err
|
|
}
|
|
quoted := strings.ReplaceAll(destination, "'", "''")
|
|
if _, err := s.db.ExecContext(ctx, `VACUUM INTO '`+quoted+`'`); err != nil {
|
|
return fmt.Errorf("export graph database: %w", err)
|
|
}
|
|
return os.Chmod(destination, 0o640)
|
|
}
|