MeshTerm/meshterm/persistence/db.py

# SPDX-License-Identifier: Apache-2.0
"""SQLite database bootstrap and schema.

The schema is intentionally generic: a central ``runs`` table records every tool
execution (with arguments and a JSON summary), and measurement tables reference it so
new tools can persist data without schema churn.
"""

from __future__ import annotations

import sqlite3
from pathlib import Path

SCHEMA_VERSION = 15

_SCHEMA = """
CREATE TABLE IF NOT EXISTS schema_meta (
    key   TEXT PRIMARY KEY,
    value TEXT NOT NULL
);

-- Small key/value store for UI state that should survive across sessions (e.g. the last
-- map viewport), independent of any tool run. New keys need no schema change.
CREATE TABLE IF NOT EXISTS app_state (
    key   TEXT PRIMARY KEY,
    value TEXT NOT NULL
);

-- One row per tool execution; the spine that measurements hang off of.
CREATE TABLE IF NOT EXISTS runs (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    tool        TEXT    NOT NULL,
    profile     TEXT,
    args_json   TEXT    NOT NULL DEFAULT '{}',
    status      TEXT    NOT NULL DEFAULT 'running',  -- running | ok | error
    summary_json TEXT,
    started_at  TEXT    NOT NULL,
    finished_at TEXT
);

-- Aggregated outcome of a single path trace, linked to its run.
CREATE TABLE IF NOT EXISTS traces (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id        INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
    target        TEXT    NOT NULL,
    success       INTEGER NOT NULL,
    hop_count     INTEGER NOT NULL DEFAULT 0,
    min_snr       REAL,
    round_trip_ms REAL,
    tx_power      INTEGER,
    created_at    TEXT    NOT NULL
);

-- Per-hop SNR for a trace.
CREATE TABLE IF NOT EXISTS trace_hops (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    trace_id  INTEGER NOT NULL REFERENCES traces(id) ON DELETE CASCADE,
    hop_index INTEGER NOT NULL,
    node      TEXT,
    snr       REAL    NOT NULL
);

-- One robust sample per (tx_power) tried during TX-power optimization.
CREATE TABLE IF NOT EXISTS tx_samples (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id        INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
    target        TEXT    NOT NULL,
    tx_power      INTEGER NOT NULL,
    median_min_snr REAL,
    success_rate  REAL,
    samples       INTEGER NOT NULL,
    created_at    TEXT    NOT NULL
);

-- A candidate path evaluated during path optimization.
CREATE TABLE IF NOT EXISTS path_candidates (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id        INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
    target        TEXT    NOT NULL,
    path_json     TEXT    NOT NULL,
    bottleneck_snr REAL,
    success_rate  REAL,
    median_rtt_ms REAL,
    created_at    TEXT    NOT NULL
);

-- One entry of a remote repeater's neighbour table, fetched over the mesh (v8). Each row
-- is a link the repeater reported hearing directly, with SNR measured at the repeater;
-- refetching appends a new snapshot and readers take the latest row per pair.
CREATE TABLE IF NOT EXISTS neighbour_reports (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id     INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
    repeater   TEXT    NOT NULL,  -- canonical id of the repeater that was asked
    neighbour  TEXT    NOT NULL,  -- reported neighbour's key-prefix hex, as replied
    snr        REAL,              -- dB, measured at the repeater
    heard_at   TEXT,              -- when the repeater last heard the neighbour
    fetched_at TEXT    NOT NULL
);

-- One record-holding walk in the trophy case (v11), scored from a trace that came
-- home (see services/records). Records live per
-- (category, width_bytes) because the per-hop hash width bounds both the walk's maximum
-- length (the path field is 64 bytes) and its collision odds — a 1-byte record is not
-- comparable to a 4-byte one. Standalone rows (no run_id): the walks' traces are already
-- persisted under their own runs, and a record must outlive history pruning. app_version
-- stamps the discoverer so future schema/scoring migrations can tell eras apart.
CREATE TABLE IF NOT EXISTS discovered_paths (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    category      TEXT    NOT NULL,
    width_bytes   INTEGER NOT NULL,
    spec          TEXT    NOT NULL,   -- the transmitted hex spec, comma-separated
    route_json    TEXT    NOT NULL,   -- canonical walked node ids, aligned with spec
    score         REAL    NOT NULL,
    stats_json    TEXT    NOT NULL,
    app_version   TEXT    NOT NULL,
    discovered_at TEXT    NOT NULL
);

-- One packet overheard while passively monitoring the mesh (advert/telemetry/...).
CREATE TABLE IF NOT EXISTS observations (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id      INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
    node        TEXT,             -- 12-hex canonical id (grouped/joined on)
    public_key  TEXT,             -- the node's full public key, when the advert carried one
    name        TEXT,
    kind        TEXT    NOT NULL DEFAULT 'advert',
    node_type   INTEGER,          -- advert type (repeater/chat/room/...), when carried
    snr         REAL,
    rssi        REAL,
    lat         REAL,
    lon         REAL,
    path        TEXT,             -- 'packet' rows: comma-separated relay-hop hex hashes
                                  --   (never a TRACE's: see trace_snrs)
    observed_at TEXT    NOT NULL,
    chan_hash   TEXT,             -- overheard channel frames: channel-hash fingerprint (hex)
    cipher_mac  TEXT,             -- …its 2-byte MAC (hex)
    crypted     TEXT,             -- …its ciphertext (hex), so a stored channel text still decrypts
    payload_typename TEXT,        -- 'packet' rows: the frame's payload class (GRP_TXT/TRACE/…)
    dest        TEXT,             -- …the recipient's key hash, for an addressed class (hex)
    src         TEXT,             -- …the sender's key hash, or its whole key (anon request)
    tag         TEXT,             -- …the frame's own token: an ack's checksum, a trace's tag
    trace_snrs  TEXT,             -- TRACE rows: comma-separated per-hop SNR readings (dB)
    route       TEXT              -- 'packet' rows: FLOOD/DIRECT/… — what `path` MEANS here
);

-- One chat message, sent or received, on a channel or with a contact. Unlike the other
-- measurement tables these are not tied to a single tool run (they arrive unsolicited via
-- the event hub), so run_id is nullable and detaches rather than cascades.
CREATE TABLE IF NOT EXISTS messages (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id      INTEGER REFERENCES runs(id) ON DELETE SET NULL,
    outbound    INTEGER NOT NULL DEFAULT 0,
    is_channel  INTEGER NOT NULL DEFAULT 0,
    channel_id  TEXT,             -- a channel's slot-independent identity (its history key)
    channel_idx INTEGER,          -- the slot it went out on / arrived on (reference only)
    peer        TEXT,
    peer_name   TEXT,
    text        TEXT    NOT NULL,
    snr         REAL,
    acked       INTEGER,
    created_at  TEXT    NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_traces_run ON traces(run_id);
CREATE INDEX IF NOT EXISTS idx_trace_hops_trace ON trace_hops(trace_id);
CREATE INDEX IF NOT EXISTS idx_tx_samples_run ON tx_samples(run_id);
CREATE INDEX IF NOT EXISTS idx_neighbour_reports_pair ON neighbour_reports(repeater, neighbour);
CREATE INDEX IF NOT EXISTS idx_discovered_paths_cat ON discovered_paths(category, width_bytes);
CREATE INDEX IF NOT EXISTS idx_observations_run ON observations(run_id);
CREATE INDEX IF NOT EXISTS idx_observations_node ON observations(node);
-- The reception history is queried by time window from every direction — the monitor's
-- session seed, the dashboard's recent slice, the Time Machine's hour range, every
-- ``since=`` filter — and observed_at stores UTC ISO-8601, which orders lexically, so a
-- plain index turns each of those from a full-table scan into a range walk.
CREATE INDEX IF NOT EXISTS idx_observations_observed_at ON observations(observed_at);
CREATE INDEX IF NOT EXISTS idx_messages_peer ON messages(peer);
"""


def _tune(conn: sqlite3.Connection) -> None:
    """Apply the durability/throughput pragmas, on the safe side of every trade-off.

    MeshTerm writes constantly and in tiny pieces — every packet the monitor hears becomes
    an ``observations`` row — and its most constrained host (the PicoCalc) keeps its
    database on an SD card, where SQLite's default of a full ``fsync`` per transaction
    costs milliseconds *each*. Three pragmas move that cost without moving the risk:

    * ``journal_mode=WAL`` — writers append to a log instead of rewriting the rollback
      journal, which turns each insert's several synchronous seeks into one append, and
      lets a reader (a screen redrawing) run against a writer (the monitor recording)
      instead of blocking on it. This one is *persistent*: it lives in the database file,
      so it survives into every later connection once set.
    * ``synchronous=NORMAL`` — the actual latency win, but **only requested once WAL is
      confirmed engaged**. Under WAL, NORMAL risks losing the last few transactions to a
      power cut and nothing worse; under the rollback journal it can leave the file
      *corrupt*. A PicoCalc running on a battery pack loses power for real, so on any
      filesystem that refused WAL this stays at the default ``FULL`` — slow and intact
      beats fast and unreadable.
    * ``temp_store=MEMORY`` — sorts and temporary b-trees stay in RAM rather than landing
      on the card. This workload's temporaries are small (screen-sized query results), so
      the memory is bounded and the card sees strictly less traffic.

    WAL can legitimately be unavailable — a read-only mount, or a filesystem without the
    shared-memory primitive it needs — and SQLite reports that by *returning* the mode it
    actually left the database in rather than raising, so the result is inspected instead
    of trusted. Either way this is best-effort tuning: a database that will not take the
    pragmas is still a perfectly working database, so nothing here is allowed to fail the
    open.

    Args:
        conn: The freshly opened connection, before any schema work.
    """
    # Returns the resulting mode as a row — "wal" only if the switch actually took.
    try:
        mode = conn.execute("PRAGMA journal_mode = WAL;").fetchone()
    except sqlite3.Error:  # pragma: no cover - filesystem-dependent
        mode = None
    if mode is not None and str(mode[0]).lower() == "wal":
        conn.execute("PRAGMA synchronous = NORMAL;")
    try:
        conn.execute("PRAGMA temp_store = MEMORY;")
    except sqlite3.Error:  # pragma: no cover - defensive; no known failure mode
        pass


def connect(db_path: Path) -> sqlite3.Connection:
    """Open (creating if needed) the SQLite database and ensure the schema exists.

    Args:
        db_path: Filesystem location of the database. Parent directories are created.

    Returns:
        An open connection with ``Row`` factory, foreign keys enabled, and the
        durability/throughput pragmas applied (see :func:`_tune`).
    """
    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    _tune(conn)
    conn.execute("PRAGMA foreign_keys = ON;")
    conn.executescript(_SCHEMA)
    _migrate(conn)
    conn.execute(
        "INSERT OR REPLACE INTO schema_meta(key, value) VALUES ('version', ?)",
        (str(SCHEMA_VERSION),),
    )
    conn.commit()
    return conn


def _migrate(conn: sqlite3.Connection) -> None:
    """Bring an existing database's tables up to the current schema.

    ``executescript(_SCHEMA)`` only *creates* missing tables (``IF NOT EXISTS``); it cannot
    add a column to a table an older version already created. Column additions are applied
    here, guarded so the migration is idempotent and safe to run on every open. Indexes on
    added columns are created here too (not in ``_SCHEMA``), so ``executescript`` never
    references a column an older database hasn't grown yet.
    """
    message_cols = {row["name"] for row in conn.execute("PRAGMA table_info(messages)")}
    if "channel_id" not in message_cols:
        # v3 -> v4: channel history moved from being keyed by slot index to a
        # slot-independent channel identity. Existing rows are backfilled lazily at runtime
        # (see Repository.backfill_channel_ids) once the device's channels can be read.
        conn.execute("ALTER TABLE messages ADD COLUMN channel_id TEXT")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_messages_channel_id ON messages(channel_id)")

    observation_cols = {row["name"] for row in conn.execute("PRAGMA table_info(observations)")}
    if "node_type" not in observation_cols:
        # v4 -> v5: observations gained the transmitting node's advert type, so the map can
        # tell repeaters from leaf nodes. Older rows simply carry NULL (type unknown).
        conn.execute("ALTER TABLE observations ADD COLUMN node_type INTEGER")
    if "path" not in observation_cols:
        # v6 -> v7: observations gained the relay path of RX-logged packets, the passive
        # evidence the mesh topology graph is built from. Older rows carry NULL (no path).
        conn.execute("ALTER TABLE observations ADD COLUMN path TEXT")
    if "chan_hash" not in observation_cols:
        # v8 -> v9: overheard channel-text (GRP_TXT) frames keep the fields the packet
        # viewer needs to decrypt them — the channel-hash fingerprint, the 2-byte MAC, and
        # the ciphertext — so a channel we hold the key for stays readable even when the feed
        # is seeded from stored history (a live frame carries them in its raw payload; a
        # replayed one had nowhere to keep them). Older rows carry NULL (never decryptable).
        conn.execute("ALTER TABLE observations ADD COLUMN chan_hash TEXT")
        conn.execute("ALTER TABLE observations ADD COLUMN cipher_mac TEXT")
        conn.execute("ALTER TABLE observations ADD COLUMN crypted TEXT")
    if "payload_typename" not in observation_cols:
        # v9 -> v10: 'packet' RX-log rows keep their payload class (GRP_TXT, TRACE, PATH,
        # ACK, …), the one identifying thing a relayed flood carries when it names no
        # origin node — so the dashboard feed can read "channel text" / "trace" instead of
        # a bare "?" even when seeded from history. Older rows carry NULL (class unknown).
        conn.execute("ALTER TABLE observations ADD COLUMN payload_typename TEXT")
    if "dest" not in observation_cols:
        # v12 -> v13: 'packet' rows keep what the frame addressed — the recipient's key
        # hash, the sender's (a hash, or the whole key an anonymous request carries), and
        # the token a tokened class stands on: an ack's checksum, a trace's tag. Decoded
        # from the frame body (see meshterm.core.frames), which only a live event carries,
        # so — exactly like the channel-text crypto trio above — a replayed row would
        # otherwise have nowhere to keep it and the feed's subject lane would go blank the
        # moment it was seeded from history. Older rows carry NULL until re-heard.
        conn.execute("ALTER TABLE observations ADD COLUMN dest TEXT")
        conn.execute("ALTER TABLE observations ADD COLUMN src TEXT")
        conn.execute("ALTER TABLE observations ADD COLUMN tag TEXT")
    if "public_key" not in observation_cols:
        # v11 -> v12: adverts carry the transmitting node's full public key, but only its
        # 12-hex prefix was ever stored as the node id. Keep that prefix as the canonical id
        # (traces, contacts, and the topology graph all match on it) and record the whole key
        # alongside it, so a hash lane can show more than the twelve stored digits. Older rows
        # carry NULL until re-heard — the raw payloads the past keys arrived in were never
        # persisted, so history can't be completed from itself.
        conn.execute("ALTER TABLE observations ADD COLUMN public_key TEXT")
    if "trace_snrs" not in observation_cols:
        # v13 -> v14: a trace's header path field is not relay hashes but one signed SNR
        # byte per hop it traversed (see meshterm.core.frames.trace_link_snrs), so it gets
        # a column of its own rather than sitting in `path` pretending to be adjacency.
        # Live frames carried the readings in their raw payload; without somewhere to keep
        # them a replayed row would show a trace's links blank, the same gap the crypto
        # trio and the addressing columns above were added to close.
        conn.execute("ALTER TABLE observations ADD COLUMN trace_snrs TEXT")
    if "route" not in observation_cols:
        # v14 -> v15: a frame's route type, which says what its `path` field *means* — and
        # without it the field was being read one way for packets that meant it two.
        #
        # A FLOOD packet accumulates: every relay appends its hash, so the path is the route
        # the packet actually travelled to reach us. A DIRECT packet is the opposite — the
        # path is a routing *instruction* the sender wrote and the relays consume, so it
        # describes where the packet is going, not where it has been, and an empty one means
        # "route used up", not "arrived in zero hops".
        #
        # Reading every path as a travelled route therefore drew a direct-routed message as
        # having reached us out of nowhere, and drew our own outgoing route as if it were an
        # inbound one (JP, 2026-09-02). Older rows carry NULL and are read as unknown rather
        # than assumed flooded; the raw headers they arrived in were never persisted, so this
        # cannot be filled in from history.
        conn.execute("ALTER TABLE observations ADD COLUMN route TEXT")