# Copyright © 2025-26 l5yth & contributors # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # frozen_string_literal: true module PotatoMesh module App module Database # Schema-version marker that gates the one-shot #756 meshcore message # content-dedup backfill. Stored in SQLite's ``PRAGMA user_version``; # bump this constant when a new one-shot migration is appended and # check the previous value below to decide whether to skip. # Bumped 1→2: the purge now keys on the stable +channel_name+ (not the # per-receiver +channel+ slot index) and inherits the widened # +MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS+ (30→300 s), so it must re-run # once to clear the cross-slot, clock-skewed duplicates that accumulated # under the old window/key. MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION = 2 # Column definitions required for environment telemetry support. Each # entry pairs the column name with the SQL type used when backfilling # legacy databases that pre-date the extended telemetry schema. TELEMETRY_COLUMN_DEFINITIONS = [ ["gas_resistance", "REAL"], ["current", "REAL"], ["iaq", "INTEGER"], ["distance", "REAL"], ["lux", "REAL"], ["white_lux", "REAL"], ["ir_lux", "REAL"], ["uv_lux", "REAL"], ["wind_direction", "INTEGER"], ["wind_speed", "REAL"], ["weight", "REAL"], ["wind_gust", "REAL"], ["wind_lull", "REAL"], ["radiation", "REAL"], ["rainfall_1h", "REAL"], ["rainfall_24h", "REAL"], ["soil_moisture", "INTEGER"], ["soil_temperature", "REAL"], ].freeze # Open a connection to the application database applying common pragmas. # # @param readonly [Boolean] whether to open the database in read-only mode. # @return [SQLite3::Database] configured database handle. def open_database(readonly: false) SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: readonly).tap do |db| db.busy_timeout = PotatoMesh::Config.db_busy_timeout_ms db.execute("PRAGMA foreign_keys = ON") end end # Execute the provided block and retry when SQLite reports a busy error. # # @param max_retries [Integer] maximum number of retries when locked. # @param base_delay [Float] incremental back-off delay between retries. # @yield Executes the database operation. # @return [Object] result of the block. def with_busy_retry( max_retries: PotatoMesh::Config.db_busy_max_retries, base_delay: PotatoMesh::Config.db_busy_retry_delay ) attempts = 0 begin yield rescue SQLite3::BusyException attempts += 1 raise if attempts > max_retries sleep(base_delay * attempts) retry end end # Determine whether the database schema has already been provisioned. # # @return [Boolean] true when all required tables exist. def db_schema_present? return false unless File.exist?(PotatoMesh::Config.db_path) db = open_database(readonly: true) required = %w[nodes messages positions telemetry neighbors instances traces trace_hops ingestors] tables = db.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages','positions','telemetry','neighbors','instances','traces','trace_hops','ingestors')", ).flatten (required - tables).empty? rescue SQLite3::Exception false ensure db&.close end # Create the database schema using the bundled SQL files. # # @return [void] def init_db FileUtils.mkdir_p(File.dirname(PotatoMesh::Config.db_path)) db = open_database %w[nodes messages positions telemetry neighbors instances traces ingestors].each do |schema| sql_file = File.expand_path("../../../../data/#{schema}.sql", __dir__) db.execute_batch(File.read(sql_file)) end ensure db&.close end # Apply any schema migrations required for older installations. # # @return [void] def ensure_schema_upgrades FileUtils.mkdir_p(File.dirname(PotatoMesh::Config.db_path)) db = open_database node_table_exists = db.get_first_value( "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='nodes'", ).to_i > 0 if node_table_exists node_columns = db.execute("PRAGMA table_info(nodes)").map { |row| row[1] } unless node_columns.include?("precision_bits") db.execute("ALTER TABLE nodes ADD COLUMN precision_bits INTEGER") node_columns << "precision_bits" end unless node_columns.include?("lora_freq") db.execute("ALTER TABLE nodes ADD COLUMN lora_freq INTEGER") end unless node_columns.include?("modem_preset") db.execute("ALTER TABLE nodes ADD COLUMN modem_preset TEXT") end unless node_columns.include?("protocol") db.execute("ALTER TABLE nodes ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") db.execute("UPDATE nodes SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") end unless node_columns.include?("synthetic") db.execute("ALTER TABLE nodes ADD COLUMN synthetic BOOLEAN NOT NULL DEFAULT 0") end if node_columns.include?("long_name") existing_indexes = db.execute("SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='nodes'").flatten unless existing_indexes.include?("idx_nodes_long_name") db.execute("CREATE INDEX IF NOT EXISTS idx_nodes_long_name ON nodes(long_name)") end end # Backfill #747: ensure_unknown_node previously omitted the protocol # column and hardcoded role=CLIENT_HIDDEN, causing meshcore placeholder # nodes to be stored as meshtastic/CLIENT_HIDDEN. Fix both in one pass. if node_columns.include?("protocol") db.execute("UPDATE nodes SET protocol = 'meshcore' WHERE long_name LIKE 'Meshcore %' AND protocol = 'meshtastic'") db.execute("UPDATE nodes SET role = 'COMPANION' WHERE protocol = 'meshcore' AND role = 'CLIENT_HIDDEN'") end # Backfill #755: reconcile meshcore synthetic placeholder rows that # share a long_name with a real (pubkey-derived) meshcore node. # Earlier releases only merged synthetics at real-node upsert time; # if a synthetic arrived after the real was already stored (common # with co-operating ingestors that share this DB), the duplicate # persisted. Migrate messages to the real id, then drop the stray # synthetic rows. Idempotent — the EXISTS guards make repeated runs # a no-op. if node_columns.include?("protocol") && node_columns.include?("synthetic") # Only collapse synthetics whose long_name resolves to *exactly* # one real meshcore node. When two real devices share a # long_name, the placeholder is ambiguous — merging would risk # mis-attributing historical chat messages to the wrong radio. # Wrapped in a single transaction so that a crash between the # UPDATE and DELETE cannot leave messages redirected without the # corresponding synthetic row cleared. db.transaction do db.execute(<<~SQL) UPDATE messages SET from_id = ( SELECT real.node_id FROM nodes real JOIN nodes synth ON synth.long_name = real.long_name WHERE synth.node_id = messages.from_id AND synth.synthetic = 1 AND synth.protocol = 'meshcore' AND real.synthetic = 0 AND real.protocol = 'meshcore' LIMIT 1 ) WHERE from_id IN ( SELECT synth.node_id FROM nodes synth WHERE synth.synthetic = 1 AND synth.protocol = 'meshcore' AND ( SELECT COUNT(*) FROM nodes real WHERE real.long_name = synth.long_name AND real.synthetic = 0 AND real.protocol = 'meshcore' ) = 1 ) SQL db.execute(<<~SQL) DELETE FROM nodes WHERE synthetic = 1 AND protocol = 'meshcore' AND ( SELECT COUNT(*) FROM nodes real WHERE real.long_name = nodes.long_name AND real.synthetic = 0 AND real.protocol = 'meshcore' AND real.node_id != nodes.node_id ) = 1 SQL end end end message_table_exists = db.get_first_value( "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='messages'", ).to_i > 0 message_columns = message_table_exists ? db.execute("PRAGMA table_info(messages)").map { |row| row[1] } : [] if message_table_exists unless message_columns.include?("lora_freq") db.execute("ALTER TABLE messages ADD COLUMN lora_freq INTEGER") end unless message_columns.include?("modem_preset") db.execute("ALTER TABLE messages ADD COLUMN modem_preset TEXT") end unless message_columns.include?("channel_name") db.execute("ALTER TABLE messages ADD COLUMN channel_name TEXT") # Keep the cached column list current so the meshcore content-dedup # purge below (which keys on +channel_name+) runs on this same boot # rather than waiting for the next restart. message_columns << "channel_name" end unless message_columns.include?("reply_id") db.execute("ALTER TABLE messages ADD COLUMN reply_id INTEGER") message_columns << "reply_id" end unless message_columns.include?("emoji") db.execute("ALTER TABLE messages ADD COLUMN emoji TEXT") message_columns << "emoji" end unless message_columns.include?("ingestor") db.execute("ALTER TABLE messages ADD COLUMN ingestor TEXT") end unless message_columns.include?("protocol") db.execute("ALTER TABLE messages ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") db.execute("UPDATE messages SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") end reply_index_exists = db.get_first_value( "SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name='idx_messages_reply_id'", ).to_i > 0 unless reply_index_exists db.execute("CREATE INDEX IF NOT EXISTS idx_messages_reply_id ON messages(reply_id)") end # #756 — partial index backing the meshcore content-dedup lookup in # insert_message. Scoped to meshcore so the index stays small even # on meshtastic-heavy deployments. ``CREATE … IF NOT EXISTS`` is # cheap enough to run on every boot; the one-shot backfill below # is gated separately via ``PRAGMA user_version`` so it does not # repeat after the first successful pass. # +channel_name+ is required because the purge keys on it (the stable # cross-ingestor discriminator) rather than the per-receiver +channel+ # slot index; a table lacking the column degrades to no purge. meshcore_dedup_columns = %w[from_id to_id channel_name text rx_time protocol] if meshcore_dedup_columns.all? { |column| message_columns.include?(column) } db.execute(<<~SQL) CREATE INDEX IF NOT EXISTS idx_messages_meshcore_content ON messages(from_id, channel, rx_time) WHERE protocol = 'meshcore' SQL # #756 backfill — collapse pre-existing meshcore duplicate groups. # Delete any row that has an EARLIER same-content row # (from_id, to_id, channel_name, text) within # #{PotatoMesh::App::DataProcessing::MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS} s, # keeping the earliest (min rx_time, min id) copy. The key + window # match the runtime guard (+insert_message+): the per-receiver # +channel+ index differs across ingestors for one logical channel, # so the stable +channel_name+ is what clusters cross-ingestor copies. # # NOTE — this EXISTS predicate is TRANSITIVE and so does NOT behave # identically to the runtime guard: a chain of identical-content rows # each within the window of the previous collapses to a single row # even if the chain spans MORE than the window, whereas the runtime # guard (which only compares a new row against already-persisted rows) # keeps roughly one row per window gap. This is intentionally more # aggressive for the one-shot historical sweep — it also clears # repeated-identical-text backlogs (e.g. a beacon/bot re-posting the # same string on a cadence ≤ the window) on top of the cross-ingestor # duplicates — and is a one-time effect; the gentler runtime guard # governs every new row. See #756 and ``CONTRACTS.md``. # # Gated via ``PRAGMA user_version`` so this expensive self-join # runs exactly once after deploy. Post-fix the runtime guard # prevents new duplicates from accumulating, so re-running on # every boot would scan ``messages`` for no reason. current_version = db.get_first_value("PRAGMA user_version").to_i if current_version < MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION window = PotatoMesh::App::DataProcessing::MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS db.transaction do # Window bound via ``?`` to match the rest of the codebase's # parameter-binding style; the value is a Ruby integer constant # so SQL-injection was never at risk here — the switch is # purely for consistency. ``PRAGMA user_version`` cannot # accept bind params, so it keeps literal interpolation of # an internal constant. db.execute(<<~SQL, [window]) DELETE FROM messages WHERE protocol = 'meshcore' AND text IS NOT NULL AND text != '' AND from_id IS NOT NULL AND EXISTS ( SELECT 1 FROM messages AS earlier WHERE earlier.protocol = 'meshcore' AND earlier.from_id = messages.from_id AND earlier.to_id IS messages.to_id AND earlier.channel_name IS messages.channel_name AND earlier.text = messages.text AND messages.rx_time - earlier.rx_time >= 0 AND messages.rx_time - earlier.rx_time <= ? AND (earlier.rx_time < messages.rx_time OR earlier.id < messages.id) ) SQL db.execute("PRAGMA user_version = #{MESHCORE_CONTENT_DEDUP_BACKFILL_VERSION}") end end end end tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='instances'").flatten if tables.empty? sql_file = File.expand_path("../../../../data/instances.sql", __dir__) db.execute_batch(File.read(sql_file)) end instance_columns = db.execute("PRAGMA table_info(instances)").map { |row| row[1] } unless instance_columns.include?("contact_link") db.execute("ALTER TABLE instances ADD COLUMN contact_link TEXT") instance_columns << "contact_link" end unless instance_columns.include?("nodes_count") db.execute("ALTER TABLE instances ADD COLUMN nodes_count INTEGER") instance_columns << "nodes_count" end unless instance_columns.include?("meshcore_nodes_count") db.execute("ALTER TABLE instances ADD COLUMN meshcore_nodes_count INTEGER") instance_columns << "meshcore_nodes_count" end unless instance_columns.include?("meshtastic_nodes_count") db.execute("ALTER TABLE instances ADD COLUMN meshtastic_nodes_count INTEGER") instance_columns << "meshtastic_nodes_count" end telemetry_tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='telemetry'").flatten if telemetry_tables.empty? telemetry_schema = File.expand_path("../../../../data/telemetry.sql", __dir__) db.execute_batch(File.read(telemetry_schema)) end telemetry_columns = db.execute("PRAGMA table_info(telemetry)").map { |row| row[1] } TELEMETRY_COLUMN_DEFINITIONS.each do |name, type| next if telemetry_columns.include?(name) db.execute("ALTER TABLE telemetry ADD COLUMN #{name} #{type}") telemetry_columns << name end unless telemetry_columns.include?("ingestor") db.execute("ALTER TABLE telemetry ADD COLUMN ingestor TEXT") end unless telemetry_columns.include?("telemetry_type") db.execute("ALTER TABLE telemetry ADD COLUMN telemetry_type TEXT") end unless telemetry_columns.include?("protocol") db.execute("ALTER TABLE telemetry ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") db.execute("UPDATE telemetry SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") end position_tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='positions'").flatten if position_tables.empty? positions_schema = File.expand_path("../../../../data/positions.sql", __dir__) db.execute_batch(File.read(positions_schema)) end position_columns = db.execute("PRAGMA table_info(positions)").map { |row| row[1] } unless position_columns.include?("ingestor") db.execute("ALTER TABLE positions ADD COLUMN ingestor TEXT") end unless position_columns.include?("protocol") db.execute("ALTER TABLE positions ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") db.execute("UPDATE positions SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") end neighbor_tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='neighbors'").flatten if neighbor_tables.empty? neighbors_schema = File.expand_path("../../../../data/neighbors.sql", __dir__) db.execute_batch(File.read(neighbors_schema)) end neighbor_columns = db.execute("PRAGMA table_info(neighbors)").map { |row| row[1] } unless neighbor_columns.include?("ingestor") db.execute("ALTER TABLE neighbors ADD COLUMN ingestor TEXT") end unless neighbor_columns.include?("protocol") db.execute("ALTER TABLE neighbors ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") db.execute("UPDATE neighbors SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") end trace_tables = db.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('traces','trace_hops')", ).flatten unless trace_tables.include?("traces") && trace_tables.include?("trace_hops") traces_schema = File.expand_path("../../../../data/traces.sql", __dir__) db.execute_batch(File.read(traces_schema)) end trace_columns = db.execute("PRAGMA table_info(traces)").map { |row| row[1] } unless trace_columns.include?("ingestor") db.execute("ALTER TABLE traces ADD COLUMN ingestor TEXT") end unless trace_columns.include?("protocol") db.execute("ALTER TABLE traces ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") db.execute("UPDATE traces SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") end ingestor_tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ingestors'").flatten if ingestor_tables.empty? ingestors_schema = File.expand_path("../../../../data/ingestors.sql", __dir__) db.execute_batch(File.read(ingestors_schema)) else ingestor_columns = db.execute("PRAGMA table_info(ingestors)").map { |row| row[1] } unless ingestor_columns.include?("version") db.execute("ALTER TABLE ingestors ADD COLUMN version TEXT") end unless ingestor_columns.include?("lora_freq") db.execute("ALTER TABLE ingestors ADD COLUMN lora_freq INTEGER") end unless ingestor_columns.include?("modem_preset") db.execute("ALTER TABLE ingestors ADD COLUMN modem_preset TEXT") end unless ingestor_columns.include?("protocol") db.execute("ALTER TABLE ingestors ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'") db.execute("UPDATE ingestors SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''") end end rescue SQLite3::SQLException, Errno::ENOENT => e warn_log( "Failed to apply schema upgrade", context: "database.schema", error_class: e.class.name, error_message: e.message, ) ensure db&.close end end end end