#include "sqlite/backend_impl.h"

#include "ddl.h"

#include "embedder.h"

#include "config.h"

#include "lang.h"

#include "util/fs.h"

#include "util/time.h"

#include "util/sqlite.h"

#include "double_metaphone.h"

#include "fts_tokenize.h"

#include "sqlite/text_index.h"

#include "vector_codec.h"

#include "Logger.h"

#include <format>

#include "nlohmann_json.hpp"

#include <sqlite3.h>

#include <Eigen/Dense>

#include <algorithm>

#include <cctype>

#include <chrono>

#include <cstdlib>

#include <cstring>

#include <ctime>

#include <unordered_map>

#include <unordered_set>

#include <iomanip>

#include <iterator>

#include <filesystem>

#include <numeric>

#include <optional>

#include <regex>

#include <set>

#include <mutex>

#include <iostream>

#include <sstream>

#include <stdexcept>

namespace ragger::sqlite {

using json = nlohmann::json;

namespace fs = std::filesystem;

namespace {

template <class Impl>

bool migration_pending(Impl& self, const std::string& current_version);

template <class Impl>

void maybe_backup_before_migration(Impl& self,

const std::string& current_version);

template <class Impl>

void reopen_db(Impl& self);

template <class Impl>

void require_foreign_keys(Impl& self);

template <class Impl>

bool pragma_bool(Impl& self, const char* name);

template <class Impl>

void rebuild_table_column_order(Impl& self,

const std::string& table,

const std::string& new_ddl,

const std::string& cols);

} // anonymous namespace

Backend::Impl::Impl(Embedder& emb, const std::string& path)

: embedder(&emb)

{

const auto& cfg = config();

db_path = path.empty() ? cfg.resolved_db_path() : expand_path(path);

vtype_ = vector_codec::parse(cfg.embedding_vector_type)

.value_or(vector_codec::VectorType::F16);

// Create parent dirs

fs::create_directories(fs::path(db_path).parent_path());

int rc = sqlite3_open(db_path.c_str(), &db);

if (rc != SQLITE_OK) {

std::string err = sqlite3_errmsg(db);

sqlite3_close(db);

db = nullptr;

throw std::runtime_error(std::format(lang::ERR_SQLITE_OPEN, err));

}

exec("PRAGMA journal_mode=WAL");

exec("PRAGMA foreign_keys = ON");

require_foreign_keys(*this);

// Wait up to 10s on a locked DB instead of failing instantly —

// WAL allows concurrent readers, but two writers (daemon +

// import CLI) still serialize; without this any collision is an

// immediate "database is locked" error.

sqlite3_busy_timeout(db, 10000);

text_index_.emplace(db);

create_schema();

// Load the current embedding version from the settings table.

// If absent (fresh DB or pre-version DB), default to 0 and stamp it.

{

Stmt s(db, "SELECT value FROM settings WHERE key = 'embedding_version'");

if (s.step()) {

embedding_version_ = static_cast<uint8_t>(

std::stoi(s.column_text(0)) & 0xff);

} else {

embedding_version_ = 1;

Stmt ins(db,

"INSERT OR IGNORE INTO settings (key, value) "

"VALUES ('embedding_version', '1')");

ins.exec();

}

}

}

/// DB-only constructor — no embedder.

/// readonly=true: opens SQLITE_OPEN_READONLY, skips schema creation (for export).

/// readonly=false: opens read-write and creates users/settings tables.

Backend::Impl::Impl(const std::string& path, bool readonly)

: embedder(nullptr), readonly_(readonly)

{

db_path = expand_path(path);

if (!readonly)

fs::create_directories(fs::path(db_path).parent_path());

int rc;

if (readonly) {

rc = sqlite3_open_v2(db_path.c_str(), &db, SQLITE_OPEN_READONLY, nullptr);

} else {

rc = sqlite3_open(db_path.c_str(), &db);

}

if (rc != SQLITE_OK) {

std::string err = sqlite3_errmsg(db);

sqlite3_close(db);

db = nullptr;

throw std::runtime_error(std::format(lang::ERR_SQLITE_OPEN, err));

}

if (!readonly) {

exec("PRAGMA journal_mode=WAL");

exec("PRAGMA foreign_keys = ON");

require_foreign_keys(*this);

}

sqlite3_busy_timeout(db, 10000);

text_index_.emplace(db);

// Only ensure users + settings tables exist (skip memory tables/FTS).

// Skip entirely for readonly connections (export path — no side-effects).

if (!readonly) {

create_user_schema();

}

}

Backend::Impl::~Impl() { close(); }

// ---- helpers -------------------------------------------------------

void Backend::Impl::exec(const char* sql) {

char* errmsg = nullptr;

int rc = sqlite3_exec(db, sql, nullptr, nullptr, &errmsg);

if (rc != SQLITE_OK) {

std::string err = errmsg ? errmsg : "unknown error";

sqlite3_free(errmsg);

throw std::runtime_error(std::format(lang::ERR_SQL, err));

}

}

void Backend::Impl::exec(const std::string& sql) { exec(sql.c_str()); }

/// True if `table` has a column named `col`. Used to guard one-time

/// ADD COLUMN migrations (SQLite has no ADD COLUMN IF NOT EXISTS).

/// The table name is inlined (PRAGMA table_info doesn't take a bound

/// parameter reliably); callers pass internal constants, never user input.

bool Backend::Impl::column_exists(const std::string& table, const std::string& col) {

Stmt s(db, "PRAGMA table_info(" + table + ")");

while (s.step()) {

if (s.column_text(1) == col) return true; // col 1 = column name

}

return false;

}

bool Backend::Impl::table_exists(const std::string& table) {

Stmt s(db, "SELECT 1 FROM sqlite_master WHERE type='table' AND name = ?");

s.bind(1, table);

return s.step();

}

void Backend::Impl::create_schema() {

// Capture "was this DB already populated?" BEFORE any CREATE TABLE

// IF NOT EXISTS runs below -- this is the only reliable way to tell

// a genuinely fresh install (no memory tables at all yet) from an

// existing pre-0.12 DB that simply never had its db_version row

// stamped. Only a fresh install should be auto-stamped '0.12' at

// the bottom of this function; an existing un-migrated DB must

// fail the startup version-gate check so the user runs

// scripts/migrate_to_db0.12.sh instead of silently being treated

// as current.

bool db_preexisted = table_exists("turns");

// Capture the on-disk schema version BEFORE any CREATE/ALTER below

// mutates the file, then snapshot the DB if an in-binary migration is

// pending -- so the backup reflects the pristine pre-migration state.

// db_version() is only meaningful for an existing DB (settings table

// already present); a fresh install has no version yet and needs no

// backup. One snapshot covers a whole migration chain (0.12 -> 0.15 ->

// ...): it is taken here, once, before the first structural change.

const std::string pre_migration_version =

db_preexisted ? db_version() : std::string();

maybe_backup_before_migration(*this, pre_migration_version);

// Users, settings, and session tables — credentials (for read-only

// document access) plus web/chat session persistence. These are a

// separate concern from the v2 memory tables; one declarative

// definition, shared with the DB-only constructor.

create_user_schema();

// ---- v2 fading-memory schema (issue #33): turns / summaries /

// decisions / documents / models, with FTS5 (issue #49).

// Pre-v2 data is exported out-of-band, not migrated in place.

// models — lookup table; turns + summaries reference it. created_at

// was added in a later schema revision; the column sits at the end so

// it maps cleanly to the canonical (id, keys, data, timestamps) order.

exec(R"(

)");

// sessions — lookup table; turns + summaries reference it. Normalizes

// the long conversation GUID (from the agent turn hook) to a compact

// integer id, mirroring `models`. Grouping key only (PHASE4): guid +

// title + derived started_on/ended_on span; no project_id.

{

exec(ddl::if_not_exists(ddl::sessions("sessions")));

}

// turns (L1) — raw verbatim exchanges. embedding nullable for the

// deferred-embedding path (partial row written, backfilled later).

// created_at has NO DEFAULT (matches scripts/schema_db0.12.sql

// exactly) -- the app always supplies it explicitly on INSERT.

// unigram_count/bigram_count are TF denominators for the custom terms

// index (v0.16); placed before embedding_version to keep text-index

// columns grouped together (cosmetic — SQLite appends physically).

exec(R"(

)");

// summaries is GONE in 0.17 (PHASE8): its rows moved to episodes

// (§2) and knowledge kind='note' (§4b) by the 0.16 -> 0.17 leg; a

// fresh DB never creates it. The legacy ALTERs below stay guarded on

// table_exists() for pre-0.16 DBs that still have it at open time.

// turn_summaries (L2) -- NEW at db_version 0.12. One row per

// summarized turn, split out of `summaries`. turn_id is a real FK,

// ON DELETE SET NULL: pruning the raw turn leaves the summary behind

// with turn_id NULL rather than stranding it via a fragile

// (session_id, created_at) match. 0.17 adds episode_id (recorded

// membership) — the ddl:: definition is the single source of truth

// for fresh DBs and the 0.16->0.17 rebuild alike.

exec(ddl::if_not_exists(ddl::turn_summaries("turn_summaries")));

// In-place migration for pre-sessions databases: add the session_id

// FK column to turns/summaries if an existing DB predates it. SQLite

// has no ADD COLUMN IF NOT EXISTS, so guard on pragma_table_info.

// (Fresh DBs already have the column from the CREATE above.)

if (!column_exists("turns", "session_id"))

exec("ALTER TABLE turns ADD COLUMN session_id "

"INTEGER REFERENCES sessions(session_id) ON DELETE SET NULL");

if (table_exists("summaries") && !column_exists("summaries", "session_id"))

exec("ALTER TABLE summaries ADD COLUMN session_id "

"INTEGER REFERENCES sessions(session_id) ON DELETE SET NULL");

exec("CREATE INDEX IF NOT EXISTS idx_turns_session ON turns(session_id)");

if (table_exists("summaries"))

exec("CREATE INDEX IF NOT EXISTS idx_summaries_session ON summaries(session_id)");

// document_sources (L5 metadata) — one row per curated document.

// Holds title/path/year/tags/imported_at once, instead of repeating

// them on every chunk. document_source_id gives a document a stable

// identity. imported_at is a unix epoch (user's local import time).

// No FTS: tiny table; its title rides the chunk embed/phon signals.

exec(R"(

)");

// documents (L5) — user-curated RAG chunks; the only sharable table.

// Per-document metadata lives in document_sources (document_source_id

// FK). Each chunk keeps only its distinctive tags; chunk_index is a

// positional handle; modified_on (epoch) = source.imported_at until the

// chunk text is edited. `text + '\n' + title` is embedded/phonized (see

// store_document), so title participates in vector + phon search.

exec(R"(

)");

// In-place migration for pre-0.15 databases: add embedding_version

// to each embedded table if an existing DB predates it. Fresh DBs

// already have the column from the CREATE above. New rows get NULL

// (mirrors a NULL embedding) until backfilled/re-embedded.

if (!column_exists("turns", "embedding_version"))

exec("ALTER TABLE turns ADD COLUMN embedding_version INTEGER");

if (table_exists("summaries") && !column_exists("summaries", "embedding_version"))

exec("ALTER TABLE summaries ADD COLUMN embedding_version INTEGER");

if (!column_exists("turn_summaries", "embedding_version"))

exec("ALTER TABLE turn_summaries ADD COLUMN embedding_version INTEGER");

if (table_exists("decisions") && !column_exists("decisions", "embedding_version"))

exec("ALTER TABLE decisions ADD COLUMN embedding_version INTEGER");

if (!column_exists("documents", "embedding_version"))

exec("ALTER TABLE documents ADD COLUMN embedding_version INTEGER");

// In-place migration for pre-0.16 databases: add unigram_count and

// bigram_count to each text table (TF denominators for the custom

// terms index). Fresh DBs already have them from the CREATE above;

// existing rows get 0 until reindexed by the v0.16 migration leg

// (Step 7 of docs/plans/migration-plan-fts-index.md). Idempotent: once

// present, these ALTERs never fire again.

if (!column_exists("turns", "unigram_count"))

exec("ALTER TABLE turns ADD COLUMN unigram_count INTEGER NOT NULL DEFAULT 0");

if (!column_exists("turns", "bigram_count"))

exec("ALTER TABLE turns ADD COLUMN bigram_count INTEGER NOT NULL DEFAULT 0");

if (table_exists("summaries") && !column_exists("summaries", "unigram_count"))

exec("ALTER TABLE summaries ADD COLUMN unigram_count INTEGER NOT NULL DEFAULT 0");

if (table_exists("summaries") && !column_exists("summaries", "bigram_count"))

exec("ALTER TABLE summaries ADD COLUMN bigram_count INTEGER NOT NULL DEFAULT 0");

if (!column_exists("turn_summaries", "unigram_count"))

exec("ALTER TABLE turn_summaries ADD COLUMN unigram_count INTEGER NOT NULL DEFAULT 0");

if (!column_exists("turn_summaries", "bigram_count"))

exec("ALTER TABLE turn_summaries ADD COLUMN bigram_count INTEGER NOT NULL DEFAULT 0");

if (table_exists("decisions") && !column_exists("decisions", "unigram_count"))

exec("ALTER TABLE decisions ADD COLUMN unigram_count INTEGER NOT NULL DEFAULT 0");

if (table_exists("decisions") && !column_exists("decisions", "bigram_count"))

exec("ALTER TABLE decisions ADD COLUMN bigram_count INTEGER NOT NULL DEFAULT 0");

if (!column_exists("documents", "unigram_count"))

exec("ALTER TABLE documents ADD COLUMN unigram_count INTEGER NOT NULL DEFAULT 0");

if (!column_exists("documents", "bigram_count"))

exec("ALTER TABLE documents ADD COLUMN bigram_count INTEGER NOT NULL DEFAULT 0");

// In-place migration to the normalized documents schema (0.15):

// extract per-document metadata (path/title/year/imported_at) into

// document_sources and slim `documents` to chunks + a document_source_id

// FK. Gated on the OLD shape (documents.path exists), NOT the db_version

// string -- Reid's live DB is already stamped 0.15 with the OLD shape,

// so a version check would wrongly skip it. Idempotent: once path is

// gone, this never runs again.

if (column_exists("documents", "path"))

migrate_documents_normalize();

// Custom terms index (v0.16) — replaces both FTS5 layers with a single

// flat term table + per-content-type junction tables. Created BEFORE

// migrations run, so migrate_0_15_to_0_16() can populate them via

// reindex_table(). See ownCloud/Ragger/custom-search-schema.md for

// the design rationale and docs/plans/migration-plan-fts-index.md for

// the step-by-step build plan.

create_terms_schema();

// Run any pending in-binary version migration for an existing DB

// (e.g. the 0.12 -> 0.15 embedding version-byte split). Advances

// db_version toward kExpectedDbVersion so the hard gate below passes.

// No-op on a fresh install or an already-current DB; an unknown/too-old

// version is left untouched and rejected by the gate.

if (db_preexisted)

run_pending_migrations(pre_migration_version);

// Schema-version hard gate. In-binary migration for known upgrade

// origins (0.12, 0.15) ran just above; anything still not matching

// kExpectedDbVersion here is an unknown/too-old version and is rejected.

// First, stamp db_version for a genuinely fresh install ONLY --

// db_preexisted (captured before any CREATE TABLE ran, at the top

// of this function) distinguishes "no memory tables existed yet"

// from an existing pre-0.12 DB that simply lacks the version row.

// An existing un-migrated DB must NOT get auto-stamped here; it

// needs to fail the check below so the user runs

// scripts/migrate_to_db0.12.sh instead of silently being treated

// as current.

if (!db_preexisted)

exec(std::format("INSERT OR IGNORE INTO settings (key, value) VALUES ('db_version', '{}')", kExpectedDbVersion));

{

std::string actual = db_version();

if (actual != kExpectedDbVersion) {

// Throw, never exit: a library must not kill its host

// process (B §1.11). Callers (main, server, mcp) already

// report std::exception and exit non-zero.

throw std::runtime_error(std::format(

"Ragger: database schema version mismatch (found '{}', need '{}').\n"

"This binary requires a database migrated to schema {}.\n"

"Restore the pre-migration backup next to the database, "

"or run the build that created this database.",

actual.empty() ? "<none>" : actual,

kExpectedDbVersion, kExpectedDbVersion));

}

}

// Post-migration reclaim: if a migration actually ran this session

// (same gating condition maybe_backup_before_migration used), the

// chain just dropped a lot of data (old FTS5 tables/triggers, phon

// columns, etc). Reclaim that space with a plain in-place VACUUM

// now that the DB is confirmed at kExpectedDbVersion. Skipped on a

// fresh install or a DB that was already current (no migration).

if (migration_pending(*this, pre_migration_version)) {

Diskerror::Logger::info(

"Post-migration VACUUM starting (reclaiming space from dropped "

"schema) -- this may take a while on a large DB.");

exec("VACUUM");

Diskerror::Logger::info("Post-migration VACUUM complete.");

}

// Indexes — (re)created here, after migration, so a table rebuild

// (which drops the table and with it every index) can't leave them

// missing. All IF NOT EXISTS, so fresh / migrated / already-current

// DBs all converge to the same set.

exec("CREATE INDEX IF NOT EXISTS idx_turns_created_at ON turns(created_at)");

exec("CREATE INDEX IF NOT EXISTS idx_turns_session ON turns(session_id)");

// A document is identified by its path (0.17): the same file cannot

// be imported twice. NULL/empty paths from old DBs are given a unique

// value by migrate_0_16_to_0_17() before this index is built.

exec("CREATE UNIQUE INDEX IF NOT EXISTS idx_document_sources_path ON document_sources(path)");

exec("CREATE INDEX IF NOT EXISTS idx_sessions_name ON sessions(name)");

// knowledge (PHASE7) — ONE table for decision | insight | note,

// replacing decisions + insights + their evidence tables (dropped by

// the 0.16 -> 0.17 leg; a fresh DB never creates the old ones).

// Created BEFORE create_terms_schema() so knowledge_terms can

// reference it, and AFTER migrations so it never takes part in a

// rebuild dance.

exec(ddl::if_not_exists(ddl::knowledge()));

exec(ddl::if_not_exists(ddl::knowledge_indexes()));

exec(ddl::if_not_exists(ddl::knowledge_turn_evidence()));

exec(ddl::if_not_exists(ddl::knowledge_document_evidence()));

exec(ddl::if_not_exists(ddl::knowledge_document_evidence_indexes()));

// episodes (PHASE5) — narrative segments. Created AFTER migrations

// (like the evidence tables) so it never takes part in a rebuild

// dance; the 0.16->0.17 leg creates and populates it itself (§2).

exec(ddl::if_not_exists(ddl::episodes()));

exec(ddl::if_not_exists(ddl::episodes_indexes()));

// turn_summaries indexes + the orphan-protection trigger. An orphaned

// turn summary (its turn culled) is the only surviving record of that

// exchange; code never deletes one, and this makes the schema refuse

// too. Empty-text marker rows stay deletable so

// reset_abandoned_turn_summaries() keeps working. Recreated here after

// any rebuild (dropping a table drops its indexes and triggers).

exec(ddl::if_not_exists(ddl::turn_summaries_indexes_and_trigger()));

exec("CREATE INDEX IF NOT EXISTS idx_document_sources_imported_at ON document_sources(imported_at)");

exec("CREATE INDEX IF NOT EXISTS idx_documents_document_source_id ON documents(document_source_id)");

// FTS5 — external-content virtual tables + sync triggers replace

// the old hand-rolled bm25_* sidecars (issue #49).

// (v0.16: FTS5 schema removed; custom terms index created below via

// create_terms_schema().)

// Human-readable views (datetime()-rendered timestamps,

// has_embedding booleans) mirroring scripts/schema_db0.12.sql

// exactly. Always (re)created so fresh/migrated DBs converge.

create_views();

}

std::string Backend::Impl::db_version() {

Stmt s(db, "SELECT value FROM settings WHERE key = 'db_version'");

if (s.step()) return s.column_text(0);

return ""; // absent

}

void Backend::Impl::set_db_version(const std::string& v) {

Stmt s(db,

"INSERT INTO settings (key, value) VALUES ('db_version', ?) "

"ON CONFLICT(key) DO UPDATE SET value = excluded.value");

s.bind(1, v);

s.exec();

}

// ---- in-binary version migrations ------------------------------------

// Historically, DB schema upgrades were external shell scripts

// (scripts/migrate_to_db*.sh) and the binary merely hard-gated on the

// version. To make GitHub-cloned upgrades painless, the known upgrade

// legs now run in-process on open, chained until the DB reaches

// kExpectedDbVersion. Each leg mirrors migrate_documents_normalize()'s

// shape: one BEGIN/try/COMMIT/ROLLBACK transaction, Logger progress,

// and post-migration verification that aborts (rolls back) on any

// inconsistency rather than silently shipping a corrupt index.

/// True iff a pre-migration backup/reopen (and, later, a post-migration

/// VACUUM) should happen for `current_version`. Mirrors the guard

/// conditions maybe_backup_before_migration() has always used: shared

/// by that function and by the post-migration VACUUM decision so both

/// stay in lockstep.

namespace {

template <class Impl>

bool migration_pending(Impl& self, const std::string& current_version) {

}

} // anonymous namespace

/// Archive the raw DB file(s) as "<db>_BACKUP_<timestamp>.tar.gz" (or

/// .zip / .db, see fallback chain below) before the first structural

/// change of a migration run. Only fires when an in-binary migration is

/// actually pending (see migration_pending()). One snapshot covers a

/// whole chain (0.12 -> 0.15 -> ...).

///

/// Preferred path: WAL-checkpoint in TRUNCATE mode WITHOUT closing the

/// connection, then archive the now-fully-checkpointed main .db file

/// while the connection stays open. A TRUNCATE checkpoint that reports

/// zero "busy" frames means every WAL frame was folded back into the

/// main file and the -wal file was truncated to empty -- at that point

/// the main .db file alone is a complete, consistent point-in-time

/// snapshot, and it's safe to read/archive it while `db` stays open,

/// PROVIDED no other writer is concurrently active (true here: this is

/// single-threaded startup code, before any other work on this

/// connection begins). This avoids ever closing/reopening `db`, so

/// text_index_'s non-owning pointer never goes stale.

///

/// Falls back to a full close -> archive -> reopen (re-pointing

/// text_index_ at the new handle) only if the checkpoint could NOT

/// fully complete without closing (checkpoint reports busy frames,

/// meaning some other connection/transaction is holding the WAL open --

/// in that case the main .db file alone would not be a safe snapshot).

///

/// Unlike the old VACUUM INTO approach (which snapshotted the live WAL

/// database via a separate materialized copy query), this archives the

/// raw file(s) directly using the existing timestamped naming

/// convention (tar.gz -> zip -> plain copy fallback chain).

namespace {

template <class Impl>

void maybe_backup_before_migration(Impl& self, const std::string& current_version) {

}

} // anonymous namespace

/// Reopen `db` (after maybe_backup_before_migration's checkpoint+close)

/// with the exact same open flags/pragmas used in the constructor, and

/// re-point the non-owning text_index_ at the new handle -- sqlite3_open

/// is not guaranteed to reuse the same pointer value.

namespace {

template <class Impl>

void reopen_db(Impl& self) {

}

} // anonymous namespace

/// Chain the in-binary migration legs until the DB reaches

/// kExpectedDbVersion (or a leg we don't know about is hit -> leave it for

/// the hard gate to reject). `from_version` is the on-disk version captured

/// before create_schema() mutated anything. Each leg advances db_version;

/// the loop re-reads it so 0.12 flows 0.12 -> 0.15 -> (0.16 once its leg

/// lands) under the single backup taken above.

void Backend::Impl::run_pending_migrations(const std::string& from_version) {

std::string v = from_version;

// Guard against an unbounded loop if a leg ever fails to advance the

// version. Legs are few; a handful of iterations is plenty.

for (int guard = 0; guard < 8; ++guard) {

if (v.empty() || v == kExpectedDbVersion) return;

if (v == "0.12") {

migrate_0_12_to_0_15();

v = db_version();

continue;

}

if (v == "0.15") {

migrate_0_15_to_0_16();

v = db_version();

continue;

}

if (v == "0.16") {

migrate_0_16_to_0_17();

v = db_version();

continue;

}

// No known leg from this version -> stop; the hard gate reports it.

return;

}

}

/// 0.12 -> 0.15 migration (in-binary port of scripts/migrate_to_db0.15.sh).

///

/// By the time this runs, create_schema() has ALREADY performed the

/// structural half of the 0.15 upgrade on this existing DB:

/// - added embedding_version INTEGER to all five embedded tables

/// (ALTER ADD COLUMN guards, above), and

/// - normalized documents into document_sources via

/// migrate_documents_normalize(), which also CLEARS documents'

/// embedding/embedding_version/phon (they get re-embedded with the

/// title appended).

///

/// The remaining data step is the embedding version-byte SPLIT for the

/// four tables migrate_documents_normalize did NOT wipe: turns,

/// turn_summaries, summaries, decisions. Pre-0.15 blobs stored a 1-byte

/// version tag at byte 0 (EmbeddingCodec offset=1); 0.15 blobs are payload

/// only (offset=0) with the tag living in the embedding_version column. So

/// for every row with a non-NULL embedding: embedding_version = blob[0],

/// embedding = blob[1:]. NULL embeddings stay NULL in both columns.

///

/// Whole thing runs in one transaction. On any verification failure the

/// transaction is rolled back (original rows untouched) and the throw

/// propagates -- the pre-migration backup made above is the safety net.

void Backend::Impl::migrate_0_12_to_0_15() {

Diskerror::Logger::info(

"Migrating DB 0.12 -> 0.15 (embedding version-byte split)...");

// (table, primary-key column) for the four tables whose embeddings

// survive into 0.15 and therefore need the byte split. documents is

// intentionally excluded: migrate_documents_normalize() already

// cleared its embeddings.

struct T { const char* table; const char* pk; };

const T tables[] = {

{"turns", "turn_id"},

{"turn_summaries", "turn_summary_id"},

{"summaries", "summary_id"},

{"decisions", "decision_id"},

};

Stmt(db, "BEGIN").exec();

try {

for (const auto& t : tables) {

// Read each row's blob, split off byte 0, write both columns

// back. Collect first, then write, so we never mutate a table

// mid-scan.

struct Row { int64_t id; int version; std::vector<uint8_t> payload; };

std::vector<Row> rows;

{

Stmt s(db, std::format(

"SELECT {}, embedding FROM {} WHERE embedding IS NOT NULL",

t.pk, t.table));

while (s.step_checked()) {

const auto* blob =

static_cast<const uint8_t*>(s.column_blob(1));

int n = s.column_bytes(1);

if (blob == nullptr || n < 1)

throw std::runtime_error(std::format(

"{}: non-NULL embedding with <1 byte (id={})",

t.table, s.column_int64(0)));

Row r;

r.id = s.column_int64(0);

r.version = blob[0];

r.payload.assign(blob + 1, blob + n); // bytes [1, n)

rows.push_back(std::move(r));

}

}

for (const auto& r : rows) {

Stmt u(db, std::format(

"UPDATE {} SET embedding_version = ?, embedding = ? "

"WHERE {} = ?", t.table, t.pk));

u.bind(1, r.version);

u.bind_blob(2, r.payload.data(),

static_cast<int>(r.payload.size()));

u.bind(3, r.id);

if (!u.exec())

throw std::runtime_error(std::format(

"{}: embedding split UPDATE failed (id={}): {}",

t.table, r.id, sqlite3_errmsg(db)));

}

Diskerror::Logger::info(std::format(

" {}: split {} embedding(s)", t.table, rows.size()));

}

// ---- verification (abort/rollback on any mismatch) ------------

// For each migrated table: embedding NULL-ness and

// embedding_version NULL-ness must move together. (The blob is now

// the pure payload; we already stripped exactly one byte in C++,

// so a per-row "1 byte shorter" DB check isn't reconstructable

// post-split -- the correctness guarantee is the byte-exact

// std::vector assign above plus this NULL-parity check.)

for (const auto& t : tables) {

Stmt s(db, std::format(

"SELECT COUNT(*) FROM {} "

"WHERE (embedding IS NULL) != (embedding_version IS NULL)",

t.table));

s.step_checked();

int64_t bad = s.column_int64(0);

if (bad != 0)

throw std::runtime_error(std::format(

"{}: {} row(s) with embedding/embedding_version NULL "

"mismatch after split", t.table, bad));

}

// Structural integrity of the whole file.

{

Stmt s(db, "PRAGMA integrity_check");

s.step_checked();

std::string res = s.column_text(0);

if (res != "ok")

throw std::runtime_error(

"PRAGMA integrity_check failed after 0.12->0.15: " + res);

}

set_db_version("0.15");

Stmt(db, "COMMIT").exec();

Diskerror::Logger::info("DB migration 0.12 -> 0.15 complete.");

} catch (...) {

Stmt(db, "ROLLBACK").exec();

throw;

}

}

/// One-time in-place migration to the normalized documents schema (0.15).

/// Extracts per-document metadata (path/title/year/imported_at) from the

/// old flat `documents` rows into `document_sources`, converts imported_at

/// text -> unix epoch (stored text is a LOCAL wall-clock timestamp), points

/// each chunk at its source via document_source_id, redistributes tags

/// (source = intersection of a (path,title) group's chunk tags; chunk keeps

/// its own set minus that intersection), clears embedding/embedding_version/

/// phon on every doc row (so housekeeping re-embeds with title appended),

/// then drops the four extracted columns. Whole thing runs in one

/// transaction. Caller gates on column_exists("documents","path").

void Backend::Impl::migrate_documents_normalize() {

Diskerror::Logger::info("Migrating documents to normalized schema (document_sources)...");

Stmt(db, "BEGIN").exec();

try {

// document_sources may not exist yet on an old DB (the CREATE TABLE

// IF NOT EXISTS above makes it, but be defensive/idempotent).

exec(R"(

)");

// The old documents_fts / triggers reference new.title/old.title.

// They MUST go before we UPDATE documents or DROP COLUMN title,

// else the AFTER UPDATE trigger fires against a missing column.

exec("DROP TRIGGER IF EXISTS documents_ai");

exec("DROP TRIGGER IF EXISTS documents_ad");

exec("DROP TRIGGER IF EXISTS documents_au");

exec("DROP TABLE IF EXISTS documents_fts");

// Same for the phon sidecar FTS + its triggers. Critically, the

// external-content documents_phon_fts must be dropped too: after we

// rebuild `documents` below, its shadow tables would still hold the

// OLD rows. The row COUNT happens to match (same 8627), so

// create_phon_fts_schema()'s docsize-vs-base desync probe would NOT

// fire a rebuild -- leaving stale rowid->content pointers that make

// every later phon UPDATE fail "database disk image is malformed".

// Dropping it here forces a fresh, empty index (docsize 0 != base)

// so the probe rebuilds and resyncs it.

exec("DROP TRIGGER IF EXISTS documents_pai");

exec("DROP TRIGGER IF EXISTS documents_pad");

exec("DROP TRIGGER IF EXISTS documents_pau");

exec("DROP TABLE IF EXISTS documents_phon_fts");

// The old documents_view SELECTs path/title/year/imported_at, so

// SQLite won't let us DROP those columns while it exists. Drop it

// here; create_views() rebuilds the new (joined) view afterward.

exec("DROP VIEW IF EXISTS documents_view");

// The old idx_documents_imported_at indexes a column we're about to

// drop -- SQLite refuses DROP COLUMN while an index references it.

exec("DROP INDEX IF EXISTS idx_documents_imported_at");

// Helper: split a comma-separated tag string into a token set

// (trimmed, empties dropped).

auto split_tags = [](const std::string& s) {

std::set<std::string> out;

std::string tok;

std::stringstream ss(s);

while (std::getline(ss, tok, ',')) {

size_t a = tok.find_first_not_of(" \t");

size_t b = tok.find_last_not_of(" \t");

if (a != std::string::npos)

out.insert(tok.substr(a, b - a + 1));

}

return out;

};

auto join_tags = [](const std::set<std::string>& s) {

std::string out;

for (const auto& t : s) { if (!out.empty()) out += ','; out += t; }

return out;

};

// Group existing chunks by (path, title). Collect each group's

// year, earliest imported_at (as epoch), the intersection of its

// chunks' tag sets, and the lowest document_id it appears in (so

// sources can be inserted in document_id order -- tidy + stable).

struct Group {

};

// key = path + '\x1f' + title (unit separator can't appear in text)

std::unordered_map<std::string, Group> groups;

{

// strftime('%s', ts, 'utc') interprets the stored text as LOCAL

// wall-clock and returns the corresponding unix epoch. (SQLite's

// bare strftime treats input as UTC; the 'utc' modifier flips a

// localtime reading back to UTC epoch.) Handles both the live

// 'YYYY-MM-DD HH:MM:SS' form and 'YYYYMMDD' stragglers (the

// latter parses as midnight local).

Stmt s(db,

"SELECT document_id, path, title, year, tags, "

" CAST(strftime('%s', imported_at, 'utc') AS INTEGER) AS ep "

"FROM documents ORDER BY document_id");

while (s.step()) {

long long doc_id = s.column_int64(0);

std::string path = s.is_null(1) ? "" : s.column_text(1);

std::string title = s.is_null(2) ? "" : s.column_text(2);

std::string key = path + "\x1f" + title;

auto it = groups.find(key);

bool is_new = (it == groups.end());

Group& g = groups[key];

if (is_new) { g.first_doc_id = doc_id; g.path = path; g.title = title; }

if (!s.is_null(3)) g.year = s.column_int64(3);

if (!s.is_null(5)) {

long long ep = s.column_int64(5);

if (!g.have_epoch || ep < g.imported_epoch) {

g.imported_epoch = ep; g.have_epoch = true;

}

}

std::set<std::string> ts = split_tags(s.is_null(4) ? "" : s.column_text(4));

if (g.first_tag) { g.tag_intersection = ts; g.first_tag = false; }

else {

std::set<std::string> inter;

std::set_intersection(g.tag_intersection.begin(), g.tag_intersection.end(),

ts.begin(), ts.end(),

std::inserter(inter, inter.begin()));

g.tag_intersection = std::move(inter);

}

}

}

// Insert sources in document_id order (tidy IDs), remember each id.

std::vector<std::pair<std::string, Group*>> ordered;

ordered.reserve(groups.size());

for (auto& [key, g] : groups) ordered.emplace_back(key, &g);

std::sort(ordered.begin(), ordered.end(),

[](const auto& a, const auto& b) {

return a.second->first_doc_id < b.second->first_doc_id;