kami
03fa52dfd4
facts gain a Subject/EntityID/ResolutionState triple and an async enrichment worker that resolves free-text subjects to canonical Nexus entity_ids, mirroring Praxis's enrichment-worker pattern. Ambiguous or unreachable Nexus never guesses an entity_id — the fact stays pending or terminal-ambiguous instead. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ghELqYhZNLub2TXGMazqA
104 lines
4.3 KiB
Go
104 lines
4.3 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
)
|
|
|
|
// migrations are ordered, forward-only schema steps applied after schema.sql.
|
|
// Index i (1-based) is the user_version the step at migrations[i-1] brings the
|
|
// DB TO; there is no step 0 — schema.sql is the idempotent baseline (version 0).
|
|
// An empty slice is a clean no-op that leaves user_version at 0.
|
|
//
|
|
// To add migration #1 (e.g. the sqlcipher rekey), append its SQL:
|
|
//
|
|
// var migrations = []string{
|
|
// `ALTER TABLE ...;`, // #1
|
|
// }
|
|
var migrations = []string{
|
|
`ALTER TABLE tools ADD COLUMN scope TEXT NOT NULL DEFAULT 'homelab';`, // #1
|
|
`ALTER TABLE reminders ADD COLUMN cron TEXT;
|
|
ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2
|
|
`CREATE TABLE memory_vectors (
|
|
id TEXT PRIMARY KEY,
|
|
vec BLOB NOT NULL,
|
|
meta TEXT NOT NULL DEFAULT '{}',
|
|
created_ts INTEGER NOT NULL
|
|
);`, // #3 — long-term vector memory (persistent backend for internal/memory)
|
|
`CREATE TABLE IF NOT EXISTS events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
fact_id INTEGER NOT NULL REFERENCES facts(id),
|
|
action TEXT NOT NULL,
|
|
object TEXT NOT NULL,
|
|
ts INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_events_action_object ON events (action, object, ts DESC);
|
|
CREATE TABLE IF NOT EXISTS proposed_routines (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
action TEXT NOT NULL,
|
|
object TEXT NOT NULL,
|
|
interval_days REAL NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'proposed' CHECK (status IN ('proposed','accepted','dismissed')),
|
|
created_ts INTEGER NOT NULL,
|
|
reminder_id INTEGER REFERENCES reminders(id),
|
|
UNIQUE(action, object)
|
|
);`, // #4 — event extraction + pattern inference
|
|
|
|
`CREATE TABLE IF NOT EXISTS ack_sends (
|
|
rule TEXT NOT NULL,
|
|
sent_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_ack_sends_rule ON ack_sends (rule, sent_at DESC);`, // #5 — sev4 telegram repeat-til-ack tracking
|
|
|
|
`CREATE TABLE IF NOT EXISTS delivery_attempts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
kind TEXT NOT NULL CHECK (kind IN ('nudge','reminder')),
|
|
rule TEXT NOT NULL DEFAULT '',
|
|
reminder_id INTEGER NOT NULL DEFAULT 0,
|
|
channel TEXT NOT NULL,
|
|
body_hash TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','sent','failed','unknown')),
|
|
created_ts INTEGER NOT NULL,
|
|
completed_ts INTEGER
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_delivery_attempts_status ON delivery_attempts (status);`, // #6 — durable delivery outbox
|
|
|
|
`ALTER TABLE facts ADD COLUMN subject TEXT NOT NULL DEFAULT '';
|
|
ALTER TABLE facts ADD COLUMN entity_id TEXT;
|
|
ALTER TABLE facts ADD COLUMN resolution_state TEXT NOT NULL DEFAULT 'none'
|
|
CHECK (resolution_state IN ('none','pending','resolved','ambiguous','not_found'));
|
|
CREATE INDEX IF NOT EXISTS idx_facts_entity_id ON facts (entity_id) WHERE entity_id IS NOT NULL;
|
|
CREATE INDEX IF NOT EXISTS idx_facts_resolution_pending ON facts (resolution_state) WHERE resolution_state = 'pending';`, // #7 — entity-aware memory (Vikunja #279): facts about a subject get resolved to a Nexus entity_id async
|
|
}
|
|
|
|
// migrate applies every migration with a number greater than the DB's current
|
|
// user_version, each in its own transaction that also bumps user_version. Fails
|
|
// closed: the first erroring step aborts and leaves prior steps committed.
|
|
func migrate(ctx context.Context, db *sql.DB) error {
|
|
var current int
|
|
if err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(¤t); err != nil {
|
|
return fmt.Errorf("read user_version: %w", err)
|
|
}
|
|
for i := current; i < len(migrations); i++ {
|
|
version := i + 1 // 1-based: migrations[i] brings DB to `version`
|
|
tx, err := db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("migration %d begin: %w", version, err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, migrations[i]); err != nil {
|
|
_ = tx.Rollback()
|
|
return fmt.Errorf("migration %d: %w", version, err)
|
|
}
|
|
// PRAGMA user_version can't be parameterized; version is our own int.
|
|
if _, err := tx.ExecContext(ctx, fmt.Sprintf("PRAGMA user_version = %d", version)); err != nil {
|
|
_ = tx.Rollback()
|
|
return fmt.Errorf("migration %d bump: %w", version, err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("migration %d commit: %w", version, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|