Update 10 - Analyse

internal/graph/sqlite_backend.go

@@ -17,7 +17,7 @@ import (
 	_ "modernc.org/sqlite"
 )
 
-const schemaVersion = 2
+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(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
@@ -128,6 +128,7 @@ func Open(dir string) (*Store, error) {
 		db.Close()
 		return nil, fmt.Errorf("load sqlite graph state: %w", err)
 	}
+	s.initAnalysisWriter()
 	return s, nil
 }
 
@@ -280,6 +281,69 @@ func (s *Store) initSchema(ctx context.Context) error {
     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 {
@@ -299,7 +363,7 @@ func (s *Store) initSchema(ctx context.Context) error {
 		return fmt.Errorf("unsupported graph database schema %d (expected at most %d)", current, schemaVersion)
 	}
 	if current < schemaVersion {
-		if current != 1 {
+		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 {
@@ -692,6 +756,7 @@ func (s *Store) Close() error {
 	if s.db == nil {
 		return nil
 	}
+	s.closeAnalysisWriter()
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	_, persistErr := s.PersistVersionContext(ctx)