mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-08-16 15:40:10 +00:00
agessaman
ebc66992dd
Port the observer firmware's neighbours feature into the bot's packet capture
service, by way of meshcore-packet-capture (upstream PRs #42/#43). On a long
interval the bot asks which repeaters it hears directly and records each
confirmed link with its measured SNR.
This is the strongest link evidence the bot collects: a first-party RF
measurement between two full 32-byte public keys. Path inference works from
1-3 byte prefixes with no keys, and complete_contact_tracking.hop_count
over-claims zero-hop (800 claimed vs 68 corroborated on the live database).
modules/neighbors_discovery.py keeps upstream's public names so its fixes and
tests stay portable. Two deliberate divergences:
- No command_lock plumbing. _SerializedCommands in modules/core.py already
serialises and paces every radio command, strictly more than upstream's
reentrant lock did.
- neighbors_collect_scopes defaults off. Upstream's zero-hop scope probe
relies on a neighbour not being a known contact; this bot tracks contacts,
and for a repeater with no stored path the library reaches zero-hop by
calling change_contact_path() then reset_path() -- mutating the device's
contact table per neighbour. Scope requests also hold the radio lock for
their whole round trip (~25s), stalling bot replies. The default cycle is
one command plus a passive listen window, during which the bot stays
responsive.
Evidence lands in neighbor_links and neighbor_observations (migration 22)
rather than mesh_connections, which cannot persist provenance. The viewer
exposes it as evidence=neighbors on /api/mesh/edges and a Neighbours Only
mode on the mesh page, with populated public keys and real SNR; confirmed
neighbours also relabel edges in the combined view and count as
provenance-trusted when framing the initial map.
neighbors_enabled is the single switch. Every enabled broker publishes once
it is on (mqttN_neighbors defaults true; set false to hold one back). The
topic derives from each broker's packets topic with the last segment swapped,
so a templated broker gets meshcore/{IATA}/{PUBLIC_KEY}/neighbors -- the
topic the firmware uses -- instead of an unrelated flat one. A derived
location-routed topic is skipped with a warning when no iata is set, rather
than publishing into meshcore/XYZ/... on a shared namespace. Snapshots are
non-retained: heard_secs_ago is relative to publish time, so a retained copy
would read as current days later.
Also adds a DM-gated `neighbors` command (the 12h interval floor makes
waiting for the scheduler impractical), which acks immediately and reports in
a second message once the window closes.
Requires meshcore >= 2.3.8 for send_node_discover_req / req_regions_sync.
903 lines
39 KiB
Python
903 lines
39 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Database migration versioning for MeshCore Bot.
|
||
|
||
Migrations are numbered functions applied exactly once and recorded in the
|
||
``schema_version`` table. New installs run all migrations in order;
|
||
upgraded installs skip any already-applied version.
|
||
|
||
Adding a migration
|
||
------------------
|
||
1. Write a function ``_mNNNN_short_description(cursor)`` below.
|
||
2. Append it to ``MIGRATIONS`` as ``(NNNN, "short description", _mNNNN_...)``.
|
||
|
||
Never modify or remove an existing migration — add a new one instead.
|
||
"""
|
||
|
||
import logging
|
||
import re
|
||
import sqlite3
|
||
from typing import Callable
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helper utilities
|
||
# ---------------------------------------------------------------------------
|
||
|
||
VALID_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
||
|
||
# Allowed column definition pattern: type keyword(s) optionally followed by
|
||
# DEFAULT and a literal value. This prevents SQL injection through the
|
||
# definition parameter of _add_column().
|
||
_VALID_COL_DEF = re.compile(
|
||
r"^[A-Z]+(?:\s+[A-Z]+)*" # type name, e.g. "TEXT", "INTEGER", "BOOLEAN"
|
||
r"(?:\s+DEFAULT\s+(?:'[^']*'|[0-9.]+|NULL|CURRENT_TIMESTAMP))?" # optional DEFAULT clause
|
||
r"$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def _validate_ident(name: str, kind: str) -> None:
|
||
if not VALID_IDENT.match(name):
|
||
raise ValueError(f"Invalid {kind} identifier: {name!r}")
|
||
|
||
|
||
def _table_exists(cursor: sqlite3.Cursor, table: str) -> bool:
|
||
_validate_ident(table, "table")
|
||
cursor.execute(
|
||
"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
|
||
(table,),
|
||
)
|
||
return cursor.fetchone() is not None
|
||
|
||
|
||
def _column_exists(cursor: sqlite3.Cursor, table: str, column: str) -> bool:
|
||
"""Return True if *column* already exists in *table*."""
|
||
_validate_ident(table, "table")
|
||
_validate_ident(column, "column")
|
||
cursor.execute(f'PRAGMA table_info("{table}")')
|
||
return any(row[1] == column for row in cursor.fetchall())
|
||
|
||
|
||
def _validate_col_definition(definition: str) -> None:
|
||
"""Ensure *definition* matches a safe SQLite column definition pattern."""
|
||
if not _VALID_COL_DEF.match(definition.strip()):
|
||
raise ValueError(f"Invalid column definition: {definition!r}")
|
||
|
||
|
||
def _add_column(
|
||
cursor: sqlite3.Cursor, table: str, column: str, definition: str
|
||
) -> None:
|
||
"""Add *column* to *table* if it does not already exist."""
|
||
_validate_ident(table, "table")
|
||
_validate_ident(column, "column")
|
||
_validate_col_definition(definition)
|
||
if not _column_exists(cursor, table, column):
|
||
cursor.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Individual migrations
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _m0001_initial_schema(cursor: sqlite3.Cursor) -> None:
|
||
"""Create all base tables. No-op for tables that already exist."""
|
||
cursor.executescript("""
|
||
CREATE TABLE IF NOT EXISTS geocoding_cache (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
query TEXT UNIQUE NOT NULL,
|
||
latitude REAL NOT NULL,
|
||
longitude REAL NOT NULL,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
expires_at TIMESTAMP NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS generic_cache (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
cache_key TEXT UNIQUE NOT NULL,
|
||
cache_value TEXT NOT NULL,
|
||
cache_type TEXT NOT NULL,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
expires_at TIMESTAMP NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS bot_metadata (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS feed_subscriptions (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
feed_type TEXT NOT NULL,
|
||
feed_url TEXT NOT NULL,
|
||
channel_name TEXT NOT NULL,
|
||
feed_name TEXT,
|
||
last_item_id TEXT,
|
||
last_check_time TIMESTAMP,
|
||
check_interval_seconds INTEGER DEFAULT 300,
|
||
enabled BOOLEAN DEFAULT 1,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
api_config TEXT,
|
||
rss_config TEXT,
|
||
UNIQUE(feed_url, channel_name)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS feed_activity (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
feed_id INTEGER NOT NULL,
|
||
item_id TEXT NOT NULL,
|
||
item_title TEXT,
|
||
processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
message_sent BOOLEAN DEFAULT 1,
|
||
FOREIGN KEY (feed_id) REFERENCES feed_subscriptions(id) ON DELETE CASCADE
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS feed_errors (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
feed_id INTEGER NOT NULL,
|
||
error_type TEXT NOT NULL,
|
||
error_message TEXT,
|
||
occurred_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
resolved_at TIMESTAMP,
|
||
FOREIGN KEY (feed_id) REFERENCES feed_subscriptions(id) ON DELETE CASCADE
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS channels (
|
||
channel_idx INTEGER PRIMARY KEY,
|
||
channel_name TEXT NOT NULL,
|
||
channel_type TEXT,
|
||
channel_key_hex TEXT,
|
||
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
UNIQUE(channel_idx)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS channel_operations (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
operation_type TEXT NOT NULL,
|
||
channel_idx INTEGER,
|
||
channel_name TEXT,
|
||
channel_key_hex TEXT,
|
||
status TEXT DEFAULT 'pending',
|
||
error_message TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS feed_message_queue (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
feed_id INTEGER NOT NULL,
|
||
channel_name TEXT NOT NULL,
|
||
message TEXT NOT NULL,
|
||
queued_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
sent_at TIMESTAMP,
|
||
FOREIGN KEY (feed_id) REFERENCES feed_subscriptions(id) ON DELETE CASCADE
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_geocoding_query ON geocoding_cache(query);
|
||
CREATE INDEX IF NOT EXISTS idx_geocoding_expires ON geocoding_cache(expires_at);
|
||
CREATE INDEX IF NOT EXISTS idx_generic_key ON generic_cache(cache_key);
|
||
CREATE INDEX IF NOT EXISTS idx_generic_type ON generic_cache(cache_type);
|
||
CREATE INDEX IF NOT EXISTS idx_generic_expires ON generic_cache(expires_at);
|
||
CREATE INDEX IF NOT EXISTS idx_feed_sub_enabled ON feed_subscriptions(enabled);
|
||
CREATE INDEX IF NOT EXISTS idx_feed_sub_type ON feed_subscriptions(feed_type);
|
||
CREATE INDEX IF NOT EXISTS idx_feed_sub_last_check ON feed_subscriptions(last_check_time);
|
||
CREATE INDEX IF NOT EXISTS idx_feed_act_feed_id ON feed_activity(feed_id);
|
||
CREATE INDEX IF NOT EXISTS idx_feed_act_proc_at ON feed_activity(processed_at);
|
||
CREATE INDEX IF NOT EXISTS idx_feed_err_feed_id ON feed_errors(feed_id);
|
||
CREATE INDEX IF NOT EXISTS idx_feed_err_occur_at ON feed_errors(occurred_at);
|
||
CREATE INDEX IF NOT EXISTS idx_feed_err_resolved ON feed_errors(resolved_at);
|
||
CREATE INDEX IF NOT EXISTS idx_channels_name ON channels(channel_name);
|
||
CREATE INDEX IF NOT EXISTS idx_chan_ops_status ON channel_operations(status, created_at);
|
||
CREATE INDEX IF NOT EXISTS idx_fmq_feed_id ON feed_message_queue(feed_id);
|
||
CREATE INDEX IF NOT EXISTS idx_fmq_sent_at ON feed_message_queue(sent_at);
|
||
""")
|
||
|
||
|
||
def _m0002_feed_subscriptions_output_format(cursor: sqlite3.Cursor) -> None:
|
||
"""Add output_format and message_send_interval_seconds to feed_subscriptions."""
|
||
_add_column(cursor, "feed_subscriptions", "output_format", "TEXT")
|
||
_add_column(
|
||
cursor,
|
||
"feed_subscriptions",
|
||
"message_send_interval_seconds",
|
||
"REAL DEFAULT 2.0",
|
||
)
|
||
|
||
|
||
def _m0003_feed_subscriptions_filter_sort(cursor: sqlite3.Cursor) -> None:
|
||
"""Add filter_config and sort_config to feed_subscriptions."""
|
||
_add_column(cursor, "feed_subscriptions", "filter_config", "TEXT")
|
||
_add_column(cursor, "feed_subscriptions", "sort_config", "TEXT")
|
||
|
||
|
||
def _m0004_channel_operations_result_processed(cursor: sqlite3.Cursor) -> None:
|
||
"""Add result_data and processed_at to channel_operations."""
|
||
_add_column(cursor, "channel_operations", "result_data", "TEXT")
|
||
_add_column(cursor, "channel_operations", "processed_at", "TIMESTAMP")
|
||
|
||
|
||
def _m0005_feed_message_queue_item_fields(cursor: sqlite3.Cursor) -> None:
|
||
"""Add item_id, item_title, and priority to feed_message_queue."""
|
||
_add_column(cursor, "feed_message_queue", "item_id", "TEXT")
|
||
_add_column(cursor, "feed_message_queue", "item_title", "TEXT")
|
||
_add_column(cursor, "feed_message_queue", "priority", "INTEGER DEFAULT 0")
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_fmq_priority "
|
||
"ON feed_message_queue(priority DESC, queued_at ASC)"
|
||
)
|
||
|
||
|
||
def _m0006_channel_operations_payload_data(cursor: sqlite3.Cursor) -> None:
|
||
"""Add payload_data to channel_operations for firmware config read/write operations."""
|
||
_add_column(cursor, "channel_operations", "payload_data", "TEXT")
|
||
|
||
|
||
# NOTE: Higher-numbered migrations can safely depend on tables created by other
|
||
# subsystems (e.g., repeater manager) by checking for table existence and then
|
||
# applying idempotent ALTER/CREATE INDEX statements.
|
||
|
||
|
||
def _m0007_packet_stream_table(cursor: sqlite3.Cursor) -> None:
|
||
"""Create packet_stream table and indexes (shared DB with web viewer)."""
|
||
cursor.execute(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS packet_stream (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
timestamp REAL NOT NULL,
|
||
data TEXT NOT NULL,
|
||
type TEXT NOT NULL
|
||
)
|
||
"""
|
||
)
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_packet_stream_timestamp ON packet_stream(timestamp)"
|
||
)
|
||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_packet_stream_type ON packet_stream(type)")
|
||
|
||
|
||
def _m0008_repeater_tables_optional_columns(cursor: sqlite3.Cursor) -> None:
|
||
"""Bring repeater/graph tables up to date if they exist."""
|
||
# repeater_contacts: add location columns only if table exists
|
||
if _table_exists(cursor, "repeater_contacts"):
|
||
for column_name, column_def in [
|
||
("latitude", "REAL"),
|
||
("longitude", "REAL"),
|
||
("city", "TEXT"),
|
||
("state", "TEXT"),
|
||
("country", "TEXT"),
|
||
]:
|
||
_add_column(cursor, "repeater_contacts", column_name, column_def)
|
||
|
||
# complete_contact_tracking: path columns + is_starred and out_bytes_per_hop
|
||
if _table_exists(cursor, "complete_contact_tracking"):
|
||
for column_name, column_def in [
|
||
("out_path", "TEXT"),
|
||
("out_path_len", "INTEGER"),
|
||
("snr", "REAL"),
|
||
("is_starred", "BOOLEAN DEFAULT 0"),
|
||
("out_bytes_per_hop", "INTEGER"),
|
||
]:
|
||
_add_column(cursor, "complete_contact_tracking", column_name, column_def)
|
||
|
||
# observed_paths: packet_hash + bytes_per_hop
|
||
if _table_exists(cursor, "observed_paths"):
|
||
for column_name, column_def in [
|
||
("packet_hash", "TEXT"),
|
||
("bytes_per_hop", "INTEGER"),
|
||
]:
|
||
_add_column(cursor, "observed_paths", column_name, column_def)
|
||
|
||
# mesh_connections: graph/viewer columns
|
||
if _table_exists(cursor, "mesh_connections"):
|
||
for column_name, column_def in [
|
||
("from_public_key", "TEXT"),
|
||
("to_public_key", "TEXT"),
|
||
("avg_hop_position", "REAL"),
|
||
("geographic_distance", "REAL"),
|
||
]:
|
||
_add_column(cursor, "mesh_connections", column_name, column_def)
|
||
|
||
|
||
def _m0009_repeater_optional_indexes(cursor: sqlite3.Cursor) -> None:
|
||
"""Create optional indexes for repeater/graph tables if they exist."""
|
||
if _table_exists(cursor, "unique_advert_packets"):
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_unique_advert_date_pubkey ON unique_advert_packets(date, public_key)"
|
||
)
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_unique_advert_hash ON unique_advert_packets(packet_hash)"
|
||
)
|
||
|
||
if _table_exists(cursor, "mesh_connections"):
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_from_prefix ON mesh_connections(from_prefix)"
|
||
)
|
||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_to_prefix ON mesh_connections(to_prefix)")
|
||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_last_seen ON mesh_connections(last_seen)")
|
||
|
||
if _table_exists(cursor, "observed_paths"):
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_observed_paths_public_key ON observed_paths(public_key, packet_type)"
|
||
)
|
||
if _column_exists(cursor, "observed_paths", "packet_hash"):
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_observed_paths_packet_hash ON observed_paths(packet_hash) WHERE packet_hash IS NOT NULL"
|
||
)
|
||
cursor.execute(
|
||
"CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_advert_unique ON observed_paths(public_key, path_hex, packet_type) WHERE public_key IS NOT NULL"
|
||
)
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_observed_paths_endpoints ON observed_paths(from_prefix, to_prefix, packet_type)"
|
||
)
|
||
cursor.execute(
|
||
"CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_message_unique ON observed_paths(from_prefix, to_prefix, path_hex, packet_type) WHERE public_key IS NULL"
|
||
)
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_observed_paths_last_seen ON observed_paths(last_seen)"
|
||
)
|
||
cursor.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_observed_paths_type_seen ON observed_paths(packet_type, last_seen)"
|
||
)
|
||
|
||
|
||
def _m0010_create_repeater_and_graph_tables(cursor: sqlite3.Cursor) -> None:
|
||
"""Create repeater/graph tables used by the web viewer and repeater manager."""
|
||
cursor.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS repeater_contacts (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
public_key TEXT UNIQUE NOT NULL,
|
||
name TEXT NOT NULL,
|
||
device_type TEXT NOT NULL,
|
||
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
contact_data TEXT,
|
||
latitude REAL,
|
||
longitude REAL,
|
||
city TEXT,
|
||
state TEXT,
|
||
country TEXT,
|
||
is_active BOOLEAN DEFAULT 1,
|
||
purge_count INTEGER DEFAULT 0
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS complete_contact_tracking (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
public_key TEXT NOT NULL,
|
||
name TEXT NOT NULL,
|
||
role TEXT NOT NULL,
|
||
device_type TEXT,
|
||
first_heard TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
last_heard TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
advert_count INTEGER DEFAULT 1,
|
||
latitude REAL,
|
||
longitude REAL,
|
||
city TEXT,
|
||
state TEXT,
|
||
country TEXT,
|
||
raw_advert_data TEXT,
|
||
signal_strength REAL,
|
||
snr REAL,
|
||
hop_count INTEGER,
|
||
is_currently_tracked BOOLEAN DEFAULT 0,
|
||
last_advert_timestamp TIMESTAMP,
|
||
location_accuracy REAL,
|
||
contact_source TEXT DEFAULT 'advertisement',
|
||
out_path TEXT,
|
||
out_path_len INTEGER,
|
||
out_bytes_per_hop INTEGER,
|
||
is_starred INTEGER DEFAULT 0
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS daily_stats (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
date DATE NOT NULL,
|
||
public_key TEXT NOT NULL,
|
||
advert_count INTEGER DEFAULT 1,
|
||
first_advert_time TIMESTAMP,
|
||
last_advert_time TIMESTAMP,
|
||
UNIQUE(date, public_key)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS unique_advert_packets (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
date DATE NOT NULL,
|
||
public_key TEXT NOT NULL,
|
||
packet_hash TEXT NOT NULL,
|
||
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
UNIQUE(date, public_key, packet_hash)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS purging_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
action TEXT NOT NULL,
|
||
public_key TEXT NOT NULL,
|
||
name TEXT NOT NULL,
|
||
reason TEXT
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS mesh_connections (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
from_prefix TEXT NOT NULL,
|
||
to_prefix TEXT NOT NULL,
|
||
from_public_key TEXT,
|
||
to_public_key TEXT,
|
||
observation_count INTEGER DEFAULT 1,
|
||
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
avg_hop_position REAL,
|
||
geographic_distance REAL,
|
||
UNIQUE(from_prefix, to_prefix)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS observed_paths (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
public_key TEXT,
|
||
packet_hash TEXT,
|
||
from_prefix TEXT NOT NULL,
|
||
to_prefix TEXT NOT NULL,
|
||
path_hex TEXT NOT NULL,
|
||
path_length INTEGER NOT NULL,
|
||
bytes_per_hop INTEGER,
|
||
packet_type TEXT NOT NULL,
|
||
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
observation_count INTEGER DEFAULT 1
|
||
);
|
||
"""
|
||
)
|
||
|
||
|
||
def _m0011_repeater_and_graph_indexes(cursor: sqlite3.Cursor) -> None:
|
||
"""Create indexes for repeater/graph tables (safe to run repeatedly)."""
|
||
cursor.executescript(
|
||
"""
|
||
CREATE INDEX IF NOT EXISTS idx_public_key ON repeater_contacts(public_key);
|
||
CREATE INDEX IF NOT EXISTS idx_device_type ON repeater_contacts(device_type);
|
||
CREATE INDEX IF NOT EXISTS idx_last_seen ON repeater_contacts(last_seen);
|
||
CREATE INDEX IF NOT EXISTS idx_is_active ON repeater_contacts(is_active);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_complete_public_key ON complete_contact_tracking(public_key);
|
||
CREATE INDEX IF NOT EXISTS idx_complete_role ON complete_contact_tracking(role);
|
||
CREATE INDEX IF NOT EXISTS idx_complete_last_heard ON complete_contact_tracking(last_heard);
|
||
CREATE INDEX IF NOT EXISTS idx_complete_currently_tracked ON complete_contact_tracking(is_currently_tracked);
|
||
CREATE INDEX IF NOT EXISTS idx_complete_location ON complete_contact_tracking(latitude, longitude);
|
||
CREATE INDEX IF NOT EXISTS idx_complete_role_tracked ON complete_contact_tracking(role, is_currently_tracked);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_unique_advert_date_pubkey ON unique_advert_packets(date, public_key);
|
||
CREATE INDEX IF NOT EXISTS idx_unique_advert_hash ON unique_advert_packets(packet_hash);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_from_prefix ON mesh_connections(from_prefix);
|
||
CREATE INDEX IF NOT EXISTS idx_to_prefix ON mesh_connections(to_prefix);
|
||
CREATE INDEX IF NOT EXISTS idx_last_seen ON mesh_connections(last_seen);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_observed_paths_public_key ON observed_paths(public_key, packet_type);
|
||
CREATE INDEX IF NOT EXISTS idx_observed_paths_packet_hash ON observed_paths(packet_hash) WHERE packet_hash IS NOT NULL;
|
||
CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_advert_unique ON observed_paths(public_key, path_hex, packet_type) WHERE public_key IS NOT NULL;
|
||
CREATE INDEX IF NOT EXISTS idx_observed_paths_endpoints ON observed_paths(from_prefix, to_prefix, packet_type);
|
||
CREATE UNIQUE INDEX IF NOT EXISTS idx_observed_paths_message_unique ON observed_paths(from_prefix, to_prefix, path_hex, packet_type) WHERE public_key IS NULL;
|
||
CREATE INDEX IF NOT EXISTS idx_observed_paths_last_seen ON observed_paths(last_seen);
|
||
CREATE INDEX IF NOT EXISTS idx_observed_paths_type_seen ON observed_paths(packet_type, last_seen);
|
||
"""
|
||
)
|
||
|
||
|
||
def _m0012_purging_log_details_column(cursor: sqlite3.Cursor) -> None:
|
||
"""Add details column for newer purging log entries."""
|
||
if _table_exists(cursor, "purging_log"):
|
||
_add_column(cursor, "purging_log", "details", "TEXT")
|
||
|
||
|
||
def _m0013_observed_paths_advert_covering_index(cursor: sqlite3.Cursor) -> None:
|
||
"""Covering index for the contacts-page recent-advert-paths window query.
|
||
|
||
Lets the ROW_NUMBER() OVER (PARTITION BY public_key ORDER BY last_seen DESC)
|
||
scan in the contacts API run as an ordered index scan over advert rows,
|
||
avoiding a full materialization + temp B-tree sort of observed_paths.
|
||
"""
|
||
if not _table_exists(cursor, "observed_paths"):
|
||
return
|
||
cursor.executescript(
|
||
"""
|
||
CREATE INDEX IF NOT EXISTS idx_observed_paths_advert_pk_seen
|
||
ON observed_paths(public_key, last_seen DESC, path_hex, path_length,
|
||
bytes_per_hop, observation_count)
|
||
WHERE packet_type = 'advert';
|
||
"""
|
||
)
|
||
|
||
|
||
def _m0014_observed_paths_multibyte_covering_index(cursor: sqlite3.Cursor) -> None:
|
||
"""Partial covering index for the mesh-graph multibyte evidence query.
|
||
|
||
/api/mesh/edges?evidence=multibyte reads every multi-byte path row
|
||
(bytes_per_hop >= 2, roughly a fifth of observed_paths); without an index
|
||
that is a full scan of the widest table in the database (~2s at 700k
|
||
rows). This partial index covers the query's columns, so it never touches
|
||
the table, and last_seen in slot 2 keeps the optional days filter indexed.
|
||
"""
|
||
if not _table_exists(cursor, "observed_paths"):
|
||
return
|
||
cursor.executescript(
|
||
"""
|
||
CREATE INDEX IF NOT EXISTS idx_observed_paths_multibyte
|
||
ON observed_paths(bytes_per_hop, last_seen, path_hex,
|
||
observation_count, first_seen)
|
||
WHERE bytes_per_hop >= 2;
|
||
"""
|
||
)
|
||
|
||
|
||
def _m0015_channel_operations_claimed_at(cursor: sqlite3.Cursor) -> None:
|
||
"""Record when a queued hardware/config operation is durably claimed.
|
||
|
||
A ``processing`` row is deliberately not auto-requeued: after a process
|
||
crash the device may already have applied the operation, so retrying it
|
||
automatically could execute a non-idempotent command twice. ``claimed_at``
|
||
gives operators enough information to diagnose and explicitly resolve such
|
||
an ambiguous operation.
|
||
"""
|
||
if _table_exists(cursor, "channel_operations"):
|
||
_add_column(cursor, "channel_operations", "claimed_at", "TIMESTAMP")
|
||
|
||
|
||
def _m0016_channel_operations_claim_owner(cursor: sqlite3.Cursor) -> None:
|
||
"""Persist enough local-process identity to recover only provably dead claims."""
|
||
if not _table_exists(cursor, "channel_operations"):
|
||
return
|
||
_add_column(cursor, "channel_operations", "claim_owner_host", "TEXT")
|
||
_add_column(cursor, "channel_operations", "claim_owner_pid", "INTEGER")
|
||
_add_column(cursor, "channel_operations", "claim_owner_boot_id", "TEXT")
|
||
|
||
|
||
def _m0017_feed_queue_item_uniqueness(cursor: sqlite3.Cursor) -> None:
|
||
"""Deduplicate identifiable queue items and prevent future duplicates.
|
||
|
||
Blank and NULL item IDs deliberately remain unconstrained: without a stable
|
||
provider identifier they cannot safely be treated as the same feed item.
|
||
For duplicate valid IDs, retain a sent row when one exists (otherwise the
|
||
oldest queued row) so migration does not resurrect already-delivered work.
|
||
"""
|
||
if not _table_exists(cursor, "feed_message_queue"):
|
||
return
|
||
cursor.execute(
|
||
"""
|
||
DELETE FROM feed_message_queue
|
||
WHERE item_id IS NOT NULL AND trim(item_id) <> ''
|
||
AND id NOT IN (
|
||
SELECT COALESCE(
|
||
MIN(CASE WHEN sent_at IS NOT NULL THEN id END),
|
||
MIN(id)
|
||
)
|
||
FROM feed_message_queue
|
||
WHERE item_id IS NOT NULL AND trim(item_id) <> ''
|
||
GROUP BY feed_id, item_id
|
||
)
|
||
"""
|
||
)
|
||
cursor.execute(
|
||
"""
|
||
CREATE UNIQUE INDEX IF NOT EXISTS idx_fmq_feed_item_unique
|
||
ON feed_message_queue(feed_id, item_id)
|
||
WHERE item_id IS NOT NULL AND trim(item_id) <> ''
|
||
"""
|
||
)
|
||
|
||
|
||
def _m0018_dashboard_rollup_tables(cursor: sqlite3.Cursor) -> None:
|
||
"""Durable dashboard storage: per-day rollups plus a current-state snapshot.
|
||
|
||
``daily_rollup`` holds one row per local date for the metrics whose raw
|
||
sources are pruned long before the dashboard's 30-day window (message_stats
|
||
and friends at 7 days, packet_stream at 3). Signal metrics are stored as
|
||
sums and counts rather than means so an arbitrary window re-aggregates
|
||
correctly — averaging daily averages weights a quiet day the same as a busy
|
||
one. Every numeric column is nullable on purpose: NULL means "no source
|
||
data for this day", which the UI must render as a gap, while 0 means "the
|
||
source was present and the count really was zero".
|
||
|
||
``dashboard_snapshot`` is a single row of JSON. The current-state half of
|
||
the dashboard is ~40 heterogeneous scalars and small arrays that are read
|
||
together and never queried by field, so columnizing it would cost a
|
||
migration per new tile for no query benefit.
|
||
"""
|
||
cursor.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS daily_rollup (
|
||
date TEXT PRIMARY KEY,
|
||
messages_total INTEGER, messages_dm INTEGER, messages_channel INTEGER,
|
||
unique_senders INTEGER, unique_channels INTEGER,
|
||
commands_total INTEGER, commands_replied INTEGER, unique_command_users INTEGER,
|
||
path_obs_total INTEGER, path_len_max INTEGER, path_len_sum INTEGER,
|
||
snr_sum REAL, snr_count INTEGER, rssi_sum REAL, rssi_count INTEGER,
|
||
hops_sum INTEGER, hops_count INTEGER,
|
||
packets_total INTEGER, packets_flood INTEGER, packets_direct INTEGER,
|
||
packets_multibyte INTEGER,
|
||
adverts_total INTEGER, nodes_active INTEGER, nodes_new INTEGER,
|
||
adverts_from_multibyte INTEGER, adverts_from_singlebyte INTEGER,
|
||
contacts_known INTEGER, contacts_tracked INTEGER,
|
||
sources_present INTEGER NOT NULL DEFAULT 0,
|
||
is_backfilled INTEGER NOT NULL DEFAULT 0,
|
||
is_final INTEGER NOT NULL DEFAULT 0,
|
||
computed_at TEXT
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_daily_rollup_computed ON daily_rollup(computed_at);
|
||
|
||
CREATE TABLE IF NOT EXISTS dashboard_snapshot (
|
||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||
generated_at REAL NOT NULL,
|
||
duration_ms INTEGER,
|
||
schema_rev INTEGER NOT NULL DEFAULT 1,
|
||
payload TEXT NOT NULL
|
||
);
|
||
"""
|
||
)
|
||
|
||
|
||
def _m0019_packet_stream_denorm_dims(cursor: sqlite3.Cursor) -> None:
|
||
"""Denormalize the four packet dimensions the dashboard aggregates on.
|
||
|
||
Counting packets by bytes_per_hop or route type via
|
||
``json_extract(data, '$.…')`` is a full scan that parses every row's JSON:
|
||
~3s for bytes_per_hop and ~6s for route_type_name on a 178k-row table.
|
||
The writer already holds these values as parsed fields, so storing them as
|
||
columns removes the parse entirely.
|
||
|
||
Deliberately no bulk UPDATE here. packet_stream rows average ~1KB, so
|
||
widening 178k of them rewrites ~180MB into the WAL inside MigrationRunner's
|
||
transaction — a multi-minute stall blocking bot startup on an SD-card Pi.
|
||
The dashboard refresher backfills a bounded batch per tick instead, and the
|
||
table's 3-day retention makes the gap self-healing regardless.
|
||
"""
|
||
if not _table_exists(cursor, "packet_stream"):
|
||
return
|
||
_add_column(cursor, "packet_stream", "route_type_name", "TEXT")
|
||
_add_column(cursor, "packet_stream", "payload_type_name", "TEXT")
|
||
_add_column(cursor, "packet_stream", "path_len", "INTEGER")
|
||
_add_column(cursor, "packet_stream", "bytes_per_hop", "INTEGER")
|
||
cursor.executescript(
|
||
"""
|
||
CREATE INDEX IF NOT EXISTS idx_packet_stream_dims
|
||
ON packet_stream(timestamp, bytes_per_hop, route_type_name, payload_type_name)
|
||
WHERE type = 'packet';
|
||
|
||
-- Worklist for the refresher's incremental backfill. Without it, the
|
||
-- "any rows left to dimension?" probe is a full table scan that costs
|
||
-- ~4.6s on a 180MB packet_stream — and it costs that on every tick
|
||
-- *after* the backlog is empty, because finding nothing still means
|
||
-- reading everything. The index shrinks toward empty as rows are
|
||
-- dimensioned, so the probe converges to a no-op. It must carry
|
||
-- ``type`` as well: non-packet rows keep a NULL route type forever and
|
||
-- would otherwise leave the index permanently non-empty.
|
||
CREATE INDEX IF NOT EXISTS idx_packet_stream_undimensioned
|
||
ON packet_stream(type, id)
|
||
WHERE route_type_name IS NULL;
|
||
"""
|
||
)
|
||
|
||
|
||
def _m0020_mesh_connections_last_seen_index(cursor: sqlite3.Cursor) -> None:
|
||
"""Add the table-specific time index used by graph windows and retention.
|
||
|
||
Older migrations attempted to create ``idx_last_seen`` for several tables.
|
||
SQLite index names are database-global, so the first table claimed that
|
||
name and ``mesh_connections`` was left without its intended index.
|
||
"""
|
||
if not _table_exists(cursor, "mesh_connections"):
|
||
return
|
||
|
||
# Some older databases already have the generic ``idx_last_seen`` attached
|
||
# to this table. Keep that usable index instead of building an identical
|
||
# second B-tree (which costs both storage and write amplification).
|
||
cursor.execute('PRAGMA index_list("mesh_connections")')
|
||
for index_row in cursor.fetchall():
|
||
index_name = index_row[1]
|
||
is_partial = bool(index_row[4]) if len(index_row) > 4 else False
|
||
cursor.execute(
|
||
"SELECT name FROM pragma_index_info(?) ORDER BY seqno",
|
||
(index_name,),
|
||
)
|
||
indexed_columns = [row[0] for row in cursor.fetchall()]
|
||
if not is_partial and indexed_columns == ["last_seen"]:
|
||
return
|
||
|
||
cursor.execute(
|
||
"""
|
||
CREATE INDEX IF NOT EXISTS idx_mesh_connections_last_seen
|
||
ON mesh_connections(last_seen)
|
||
"""
|
||
)
|
||
|
||
|
||
def _m0021_daily_rollup_packet_type_encoding(cursor: sqlite3.Cursor) -> None:
|
||
"""Record the per-payload-type multibyte split on each rollup day.
|
||
|
||
``packet_stream`` is pruned at three days, so the dashboard's 30-day
|
||
encoding trend cannot be recomputed from it after the fact — the split has
|
||
to be written as each day is rolled up, the same accumulate-forward shape
|
||
the advert share already uses.
|
||
|
||
One JSON column rather than a count pair per type: the payload-type
|
||
vocabulary belongs to the firmware, not to us, so a type appearing on the
|
||
mesh should not need a schema migration to be charted.
|
||
"""
|
||
if not _table_exists(cursor, "daily_rollup"):
|
||
return
|
||
_add_column(cursor, "daily_rollup", "packet_type_encoding", "TEXT")
|
||
|
||
|
||
def _m0022_neighbor_tables(cursor: sqlite3.Cursor) -> None:
|
||
"""Tables for zero-hop neighbor discovery (see modules/neighbors_discovery.py).
|
||
|
||
A discover response is the strongest link evidence this codebase has: a
|
||
confirmed direct RF reception between two *full* 32-byte public keys, with a
|
||
measured SNR. Every other edge source is weaker — ``observed_paths`` carries
|
||
1–3 byte prefixes and no keys, and ``complete_contact_tracking.hop_count``
|
||
over-claims zero-hop. So these keep the full keys rather than prefixes, and
|
||
stay separate from ``mesh_connections``, which cannot represent provenance
|
||
(its ``confirmed_2byte`` flag is memory-only and never persisted).
|
||
|
||
Two tables because they answer different questions: ``neighbor_links`` is the
|
||
current adjacency the mesh graph reads, ``neighbor_observations`` is the
|
||
per-cycle history a signal trend needs. SNR is stored as sum+count rather
|
||
than a mean, matching ``daily_rollup``, so any window re-aggregates exactly.
|
||
|
||
Index names are database-global in SQLite (see the note on migration 20), so
|
||
every name here is table-qualified.
|
||
"""
|
||
cursor.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS neighbor_links (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
self_public_key TEXT NOT NULL,
|
||
neighbor_public_key TEXT NOT NULL,
|
||
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
observation_count INTEGER DEFAULT 1,
|
||
snr_sum REAL DEFAULT 0,
|
||
snr_count INTEGER DEFAULT 0,
|
||
best_snr REAL,
|
||
last_snr REAL,
|
||
last_status TEXT,
|
||
scopes TEXT,
|
||
UNIQUE(self_public_key, neighbor_public_key)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS neighbor_observations (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
observed_at TIMESTAMP NOT NULL,
|
||
self_public_key TEXT NOT NULL,
|
||
neighbor_public_key TEXT NOT NULL,
|
||
snr REAL,
|
||
heard_secs_ago INTEGER,
|
||
scopes TEXT,
|
||
status TEXT NOT NULL
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_neighbor_links_last_seen
|
||
ON neighbor_links(last_seen);
|
||
CREATE INDEX IF NOT EXISTS idx_neighbor_links_neighbor
|
||
ON neighbor_links(neighbor_public_key);
|
||
CREATE INDEX IF NOT EXISTS idx_neighbor_observations_observed_at
|
||
ON neighbor_observations(observed_at);
|
||
CREATE INDEX IF NOT EXISTS idx_neighbor_observations_neighbor
|
||
ON neighbor_observations(neighbor_public_key, observed_at);
|
||
"""
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Migration registry — append new entries here, never remove or reorder.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
MigrationEntry = tuple[int, str, Callable[[sqlite3.Cursor], None]]
|
||
|
||
MIGRATIONS: list[MigrationEntry] = [
|
||
(1, "initial schema", _m0001_initial_schema),
|
||
(2, "feed_subscriptions: output_format, message_send_interval_seconds", _m0002_feed_subscriptions_output_format),
|
||
(3, "feed_subscriptions: filter_config, sort_config", _m0003_feed_subscriptions_filter_sort),
|
||
(4, "channel_operations: result_data, processed_at", _m0004_channel_operations_result_processed),
|
||
(5, "feed_message_queue: item_id, item_title, priority", _m0005_feed_message_queue_item_fields),
|
||
(6, "channel_operations: payload_data", _m0006_channel_operations_payload_data),
|
||
(7, "packet_stream table for web viewer", _m0007_packet_stream_table),
|
||
(8, "optional repeater/graph columns", _m0008_repeater_tables_optional_columns),
|
||
(9, "optional repeater/graph indexes", _m0009_repeater_optional_indexes),
|
||
(10, "create repeater/graph tables", _m0010_create_repeater_and_graph_tables),
|
||
(11, "repeater/graph indexes", _m0011_repeater_and_graph_indexes),
|
||
(12, "purging_log: add details column", _m0012_purging_log_details_column),
|
||
(13, "observed_paths: advert covering index for contacts page", _m0013_observed_paths_advert_covering_index),
|
||
(14, "observed_paths: multibyte covering index for mesh graph", _m0014_observed_paths_multibyte_covering_index),
|
||
(15, "channel_operations: claimed_at", _m0015_channel_operations_claimed_at),
|
||
(16, "channel_operations: claim owner identity", _m0016_channel_operations_claim_owner),
|
||
(17, "feed_message_queue: unique identifiable items", _m0017_feed_queue_item_uniqueness),
|
||
(18, "dashboard rollup and snapshot tables", _m0018_dashboard_rollup_tables),
|
||
(19, "packet_stream: denormalized packet dimensions", _m0019_packet_stream_denorm_dims),
|
||
(20, "mesh_connections: table-specific last_seen index", _m0020_mesh_connections_last_seen_index),
|
||
(21, "daily_rollup: per-payload-type multibyte split", _m0021_daily_rollup_packet_type_encoding),
|
||
(22, "neighbor discovery tables", _m0022_neighbor_tables),
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Migration runner
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class MigrationRunner:
|
||
"""Apply pending numbered migrations to a SQLite connection.
|
||
|
||
Usage::
|
||
|
||
with db_manager.connection() as conn:
|
||
runner = MigrationRunner(conn, logger)
|
||
runner.run()
|
||
conn.commit()
|
||
"""
|
||
|
||
def __init__(self, conn: sqlite3.Connection, logger: logging.Logger) -> None:
|
||
self.conn = conn
|
||
self.logger = logger
|
||
|
||
def _ensure_version_table(self) -> None:
|
||
self.conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS schema_version (
|
||
version INTEGER NOT NULL,
|
||
description TEXT,
|
||
applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
""")
|
||
# Legacy DBs may have been created without a uniqueness constraint.
|
||
self.conn.execute(
|
||
"CREATE UNIQUE INDEX IF NOT EXISTS idx_schema_version_version ON schema_version(version)"
|
||
)
|
||
|
||
def _applied_versions(self) -> set[int]:
|
||
cursor = self.conn.execute("SELECT version FROM schema_version")
|
||
return {int(row[0]) for row in cursor.fetchall() if row and row[0] is not None}
|
||
|
||
def _validate_versions(self, applied: set[int]) -> None:
|
||
known = {v for v, _, _ in MIGRATIONS}
|
||
unknown = sorted(v for v in applied if v not in known)
|
||
if unknown:
|
||
raise RuntimeError(
|
||
"Database schema is newer or inconsistent with this codebase. "
|
||
f"Unknown applied migration version(s): {unknown}. "
|
||
"Upgrade the bot to a newer version that includes these migrations."
|
||
)
|
||
|
||
def _apply(self, version: int, description: str, fn: Callable[[sqlite3.Cursor], None]) -> None:
|
||
cursor = self.conn.cursor()
|
||
fn(cursor)
|
||
cursor.execute(
|
||
"INSERT INTO schema_version (version, description) VALUES (?, ?)",
|
||
(version, description),
|
||
)
|
||
self.logger.info(f"DB migration {version:04d} applied: {description}")
|
||
|
||
def run(self) -> None:
|
||
"""Apply all pending migrations in version order."""
|
||
self._ensure_version_table()
|
||
applied = self._applied_versions()
|
||
self._validate_versions(applied)
|
||
pending = [(v, d, f) for v, d, f in MIGRATIONS if v not in applied]
|
||
pending.sort(key=lambda x: x[0])
|
||
if not pending:
|
||
self.logger.debug("Database schema is up to date")
|
||
return
|
||
|
||
try:
|
||
self.conn.execute("BEGIN IMMEDIATE")
|
||
for version, description, fn in pending:
|
||
self._apply(version, description, fn)
|
||
self.conn.commit()
|
||
except Exception:
|
||
self.conn.rollback()
|
||
raise
|
||
|
||
self.logger.info(
|
||
f"Database migrations complete: {len(pending)} applied, "
|
||
f"schema now at version {pending[-1][0]}"
|
||
)
|