From 657917142a6368ea2535f948508ebc29adcf9aad Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:17:18 +0100 Subject: [PATCH] Add visibility, product, and owner data migrations --- .../db/migrations/016_private_prefixes.sql | 255 ++++++++++++++++++ .../016_stale_mqtt_observer_cleanup.sql | 33 +++ .../017_packet_visibility_materialization.sql | 61 +++++ .../018_api_performance_indexes.sql | 22 ++ .../db/migrations/020_product_features.sql | 10 + .../db/migrations/021_spam_webhook_state.sql | 2 + .../db/migrations/022_owner_alert_rules.sql | 14 + 7 files changed, 397 insertions(+) create mode 100644 backend/src/db/migrations/016_private_prefixes.sql create mode 100644 backend/src/db/migrations/016_stale_mqtt_observer_cleanup.sql create mode 100644 backend/src/db/migrations/017_packet_visibility_materialization.sql create mode 100644 backend/src/db/migrations/018_api_performance_indexes.sql create mode 100644 backend/src/db/migrations/020_product_features.sql create mode 100644 backend/src/db/migrations/021_spam_webhook_state.sql create mode 100644 backend/src/db/migrations/022_owner_alert_rules.sql diff --git a/backend/src/db/migrations/016_private_prefixes.sql b/backend/src/db/migrations/016_private_prefixes.sql new file mode 100644 index 0000000..cd6d1ad --- /dev/null +++ b/backend/src/db/migrations/016_private_prefixes.sql @@ -0,0 +1,255 @@ +SET LOCAL statement_timeout = 0; + +ALTER TABLE packets ADD COLUMN IF NOT EXISTS topic_prefix TEXT NOT NULL DEFAULT ''; +ALTER TABLE packets ADD COLUMN IF NOT EXISTS iata TEXT; +ALTER TABLE packets ADD COLUMN IF NOT EXISTS is_private BOOLEAN NOT NULL DEFAULT FALSE; +ALTER TABLE packets ADD COLUMN IF NOT EXISTS visibility_ok BOOLEAN NOT NULL DEFAULT TRUE; + +ALTER TABLE nodes ADD COLUMN IF NOT EXISTS last_rx_at TIMESTAMPTZ; +ALTER TABLE nodes ADD COLUMN IF NOT EXISTS last_status_at TIMESTAMPTZ; +ALTER TABLE nodes ADD COLUMN IF NOT EXISTS observer_iata TEXT; + +CREATE TABLE IF NOT EXISTS private_node_prefixes ( + node_id TEXT NOT NULL REFERENCES nodes(node_id) ON DELETE CASCADE, + network TEXT NOT NULL, + prefix_size_bytes SMALLINT NOT NULL CHECK (prefix_size_bytes BETWEEN 1 AND 3), + prefix TEXT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (node_id, network, prefix_size_bytes) +); + +CREATE INDEX IF NOT EXISTS private_node_prefixes_lookup_idx + ON private_node_prefixes (network, prefix_size_bytes, prefix); + +CREATE OR REPLACE FUNCTION sync_private_node_prefixes() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + changed_node_id TEXT := COALESCE(NEW.node_id, OLD.node_id); + changed_network TEXT := COALESCE(NEW.network, OLD.network, 'ukmesh'); +BEGIN + IF TG_OP = 'UPDATE' + AND OLD.name IS NOT DISTINCT FROM NEW.name + AND OLD.network IS NOT DISTINCT FROM NEW.network THEN + RETURN NEW; + END IF; + + DELETE FROM private_node_prefixes WHERE node_id = changed_node_id; + + IF TG_OP <> 'DELETE' AND COALESCE(NEW.name, '') LIKE '%🚫%' THEN + INSERT INTO private_node_prefixes (node_id, network, prefix_size_bytes, prefix) + VALUES + (NEW.node_id, COALESCE(NEW.network, 'ukmesh'), 1, UPPER(LEFT(NEW.node_id, 2))), + (NEW.node_id, COALESCE(NEW.network, 'ukmesh'), 2, UPPER(LEFT(NEW.node_id, 4))), + (NEW.node_id, COALESCE(NEW.network, 'ukmesh'), 3, UPPER(LEFT(NEW.node_id, 6))) + ON CONFLICT (node_id, network, prefix_size_bytes) DO UPDATE SET + prefix = EXCLUDED.prefix, + updated_at = NOW(); + END IF; + + -- A privacy-name or network change must also rematerialize packets already + -- stored for this participant. Restrict the rewrite to rows that can mention + -- the changed identity, then recompute against the complete prefix index. + UPDATE packets p + SET is_private = EXISTS ( + SELECT 1 + FROM private_node_prefixes pp + WHERE ( + pp.network = p.network + OR ( + pp.network IN ('ukmesh', 'northeast', 'teesside') + AND p.network IN ('ukmesh', 'northeast', 'teesside') + ) + ) + AND ( + pp.node_id IN (p.rx_node_id, p.src_node_id) + OR ( + p.path_hash_size_bytes = pp.prefix_size_bytes + AND EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS packet_prefix + WHERE UPPER(packet_prefix) = pp.prefix + ) + ) + ) + ), + visibility_ok = ( + (COALESCE(cardinality(p.path_hashes), 0) = 0 OR p.path_hash_size_bytes BETWEEN 1 AND 3) + AND NOT EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS path_hash + WHERE path_hash IS NULL + OR length(path_hash) <> p.path_hash_size_bytes * 2 + OR path_hash !~ '^[0-9A-Fa-f]+$' + ) + AND NOT EXISTS ( + SELECT 1 + FROM private_node_prefixes pp + WHERE ( + pp.network = p.network + OR ( + pp.network IN ('ukmesh', 'northeast', 'teesside') + AND p.network IN ('ukmesh', 'northeast', 'teesside') + ) + ) + AND ( + pp.node_id IN (p.rx_node_id, p.src_node_id) + OR ( + p.path_hash_size_bytes = pp.prefix_size_bytes + AND EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS packet_prefix + WHERE UPPER(packet_prefix) = pp.prefix + ) + ) + ) + ) + ) + WHERE ( + p.rx_node_id = changed_node_id + OR p.src_node_id = changed_node_id + OR ( + p.network = changed_network + OR ( + p.network IN ('ukmesh', 'northeast', 'teesside') + AND changed_network IN ('ukmesh', 'northeast', 'teesside') + ) + ) + AND EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS packet_prefix + WHERE UPPER(packet_prefix) IN ( + UPPER(LEFT(changed_node_id, 2)), + UPPER(LEFT(changed_node_id, 4)), + UPPER(LEFT(changed_node_id, 6)) + ) + ) + ); + + IF TG_OP = 'DELETE' THEN RETURN OLD; END IF; + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS nodes_private_prefix_materialization ON nodes; +CREATE TRIGGER nodes_private_prefix_materialization +AFTER INSERT OR UPDATE OR DELETE ON nodes +FOR EACH ROW +EXECUTE FUNCTION sync_private_node_prefixes(); + +INSERT INTO private_node_prefixes (node_id, network, prefix_size_bytes, prefix) +SELECT n.node_id, n.network, sizes.size_bytes, UPPER(LEFT(n.node_id, sizes.size_bytes * 2)) +FROM nodes n +CROSS JOIN (VALUES (1), (2), (3)) AS sizes(size_bytes) +WHERE n.name LIKE '%🚫%' +ON CONFLICT (node_id, network, prefix_size_bytes) DO UPDATE SET + prefix = EXCLUDED.prefix, + updated_at = NOW(); + +UPDATE packets + SET topic_prefix = split_part(topic, '/', 1), + iata = NULLIF(UPPER(split_part(topic, '/', 2)), '') + WHERE topic_prefix = '' OR iata IS NULL; + +WITH latest_rx AS ( + SELECT DISTINCT ON (rx_node_id) + rx_node_id, time, iata + FROM packets + WHERE rx_node_id IS NOT NULL AND rx_node_id <> '' + ORDER BY rx_node_id, time DESC +) +UPDATE nodes n + SET last_rx_at = r.time, + observer_iata = COALESCE(r.iata, n.iata) + FROM latest_rx r + WHERE n.node_id = r.rx_node_id; + +WITH latest_status AS ( + SELECT DISTINCT ON (node_id) node_id, time + FROM node_status_samples + ORDER BY node_id, time DESC +) +UPDATE nodes n + SET last_status_at = s.time + FROM latest_status s + WHERE n.node_id = s.node_id; + +UPDATE packets p + SET is_private = EXISTS ( + SELECT 1 + FROM private_node_prefixes pp + WHERE ( + pp.network = p.network + OR ( + pp.network IN ('ukmesh', 'northeast', 'teesside') + AND p.network IN ('ukmesh', 'northeast', 'teesside') + ) + ) + AND ( + pp.node_id IN (p.rx_node_id, p.src_node_id) + OR ( + p.path_hash_size_bytes = pp.prefix_size_bytes + AND EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS packet_prefix + WHERE UPPER(packet_prefix) = pp.prefix + ) + ) + ) + ), + visibility_ok = ( + (COALESCE(cardinality(p.path_hashes), 0) = 0 OR p.path_hash_size_bytes BETWEEN 1 AND 3) + AND NOT EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS path_hash + WHERE path_hash IS NULL + OR length(path_hash) <> p.path_hash_size_bytes * 2 + OR path_hash !~ '^[0-9A-Fa-f]+$' + ) + AND NOT EXISTS ( + SELECT 1 + FROM private_node_prefixes pp + WHERE ( + pp.network = p.network + OR ( + pp.network IN ('ukmesh', 'northeast', 'teesside') + AND p.network IN ('ukmesh', 'northeast', 'teesside') + ) + ) + AND ( + pp.node_id IN (p.rx_node_id, p.src_node_id) + OR ( + p.path_hash_size_bytes = pp.prefix_size_bytes + AND EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS packet_prefix + WHERE UPPER(packet_prefix) = pp.prefix + ) + ) + ) + ) + ); + +DELETE FROM packet_daily_stats +WHERE day >= CURRENT_DATE - 30; + +INSERT INTO packet_daily_stats ( + network, day, max_hop_count, max_hop_hash, max_hop_seen_at, updated_at +) +SELECT DISTINCT ON (network, time::date) + network, + time::date, + hop_count, + packet_hash, + time, + NOW() +FROM packets +WHERE visibility_ok IS TRUE + AND hop_count IS NOT NULL + AND time >= CURRENT_DATE - 30 +ORDER BY network, time::date, hop_count DESC, time DESC; + +CREATE INDEX IF NOT EXISTS packets_public_visibility_idx + ON packets (network, time DESC) + WHERE visibility_ok IS TRUE; diff --git a/backend/src/db/migrations/016_stale_mqtt_observer_cleanup.sql b/backend/src/db/migrations/016_stale_mqtt_observer_cleanup.sql new file mode 100644 index 0000000..4c6e719 --- /dev/null +++ b/backend/src/db/migrations/016_stale_mqtt_observer_cleanup.sql @@ -0,0 +1,33 @@ +ALTER TABLE nodes + ADD COLUMN IF NOT EXISTS last_mqtt_observer_seen_at TIMESTAMPTZ; + +-- The observer rollup is small and contains the best currently available +-- bootstrap signal. Older stale rows were cleaned manually before rollout. +UPDATE nodes n + SET last_mqtt_observer_seen_at = latest.last_seen + FROM ( + SELECT rx_node_id, MAX(last_seen) AS last_seen + FROM observer_region_observer_sightings + GROUP BY rx_node_id + ) latest + WHERE n.node_id = latest.rx_node_id + AND ( + n.last_mqtt_observer_seen_at IS NULL + OR n.last_mqtt_observer_seen_at < latest.last_seen + ); + +CREATE INDEX IF NOT EXISTS nodes_mqtt_observer_stale_idx + ON nodes (last_mqtt_observer_seen_at) + WHERE role IS NULL OR role = 2; + +CREATE TABLE IF NOT EXISTS maintenance_removed_records ( + archive_id BIGSERIAL PRIMARY KEY, + batch_id TEXT NOT NULL, + source_table TEXT NOT NULL, + record_data JSONB NOT NULL, + reason TEXT NOT NULL, + archived_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS maintenance_removed_records_batch_idx + ON maintenance_removed_records (batch_id, source_table); diff --git a/backend/src/db/migrations/017_packet_visibility_materialization.sql b/backend/src/db/migrations/017_packet_visibility_materialization.sql new file mode 100644 index 0000000..3b0fe34 --- /dev/null +++ b/backend/src/db/migrations/017_packet_visibility_materialization.sql @@ -0,0 +1,61 @@ +ALTER TABLE packets + ADD COLUMN IF NOT EXISTS is_private BOOLEAN NOT NULL DEFAULT FALSE; + +ALTER TABLE packets + ADD COLUMN IF NOT EXISTS visibility_ok BOOLEAN NOT NULL DEFAULT TRUE; + +CREATE INDEX IF NOT EXISTS packets_visible_network_time_idx + ON packets (network, time DESC) + WHERE visibility_ok = TRUE; + +CREATE TABLE IF NOT EXISTS private_node_prefixes ( + node_id TEXT NOT NULL REFERENCES nodes(node_id) ON DELETE CASCADE, + network TEXT NOT NULL, + prefix_size_bytes SMALLINT NOT NULL CHECK (prefix_size_bytes BETWEEN 1 AND 3), + prefix TEXT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (node_id, network, prefix_size_bytes) +); + +CREATE INDEX IF NOT EXISTS private_node_prefix_lookup_idx + ON private_node_prefixes (network, prefix_size_bytes, prefix); + +INSERT INTO private_node_prefixes (node_id, network, prefix_size_bytes, prefix) +SELECT node_id, network, prefix_size_bytes, UPPER(LEFT(node_id, prefix_size_bytes * 2)) +FROM nodes +CROSS JOIN (VALUES (1), (2), (3)) AS lengths(prefix_size_bytes) +WHERE name LIKE '%🚫%' +ON CONFLICT (node_id, network, prefix_size_bytes) DO UPDATE SET + prefix = EXCLUDED.prefix, + updated_at = NOW(); + +-- One-time classification for historical rows. The UPDATE only rewrites +-- packets that are actually private; public rows retain the fast defaults. +UPDATE packets packet +SET is_private = TRUE, + visibility_ok = FALSE +WHERE EXISTS ( + SELECT 1 + FROM nodes private_node + WHERE private_node.name LIKE '%🚫%' + AND private_node.node_id IN (packet.rx_node_id, packet.src_node_id) + AND ( + private_node.network = packet.network + OR ( + private_node.network IN ('ukmesh', 'northeast', 'teesside') + AND packet.network IN ('ukmesh', 'northeast', 'teesside') + ) + ) +) +OR EXISTS ( + SELECT 1 + FROM unnest(COALESCE(packet.path_hashes, ARRAY[]::text[])) AS path_hash(hash) + JOIN private_node_prefixes private_prefix + ON private_prefix.prefix_size_bytes = packet.path_hash_size_bytes + AND private_prefix.prefix = UPPER(path_hash.hash) + WHERE private_prefix.network = packet.network + OR ( + private_prefix.network IN ('ukmesh', 'northeast', 'teesside') + AND packet.network IN ('ukmesh', 'northeast', 'teesside') + ) +); diff --git a/backend/src/db/migrations/018_api_performance_indexes.sql b/backend/src/db/migrations/018_api_performance_indexes.sql new file mode 100644 index 0000000..2837940 --- /dev/null +++ b/backend/src/db/migrations/018_api_performance_indexes.sql @@ -0,0 +1,22 @@ +ALTER TABLE packets + ADD COLUMN IF NOT EXISTS companion_sender TEXT; + +CREATE INDEX IF NOT EXISTS idx_nodes_role_coverage + ON nodes(role, lat, lon) + WHERE role = 2 + AND lat IS NOT NULL + AND lon IS NOT NULL + AND (name IS NULL OR name NOT LIKE '%🚫%'); + +CREATE INDEX IF NOT EXISTS idx_packets_companion_sender_recent + ON packets(network, time DESC, companion_sender) + WHERE packet_type = 5 AND companion_sender IS NOT NULL; + +-- Backfill in the migration so existing 24-hour companion activity is visible +-- immediately. Future packets populate the column once at ingest. +UPDATE packets +SET companion_sender = NULLIF(BTRIM(payload->'decrypted'->>'sender'), '') +WHERE packet_type = 5 + AND companion_sender IS NULL + AND payload->'decrypted'->>'sender' IS NOT NULL + AND time > NOW() - INTERVAL '24 hours'; diff --git a/backend/src/db/migrations/020_product_features.sql b/backend/src/db/migrations/020_product_features.sql new file mode 100644 index 0000000..722d361 --- /dev/null +++ b/backend/src/db/migrations/020_product_features.sql @@ -0,0 +1,10 @@ +CREATE TABLE IF NOT EXISTS observer_registration_requests ( + id BIGSERIAL PRIMARY KEY, + public_key TEXT NOT NULL UNIQUE, + iata TEXT NOT NULL, + display_name TEXT, + contact TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'approved', 'rejected')), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/backend/src/db/migrations/021_spam_webhook_state.sql b/backend/src/db/migrations/021_spam_webhook_state.sql new file mode 100644 index 0000000..a044dea --- /dev/null +++ b/backend/src/db/migrations/021_spam_webhook_state.sql @@ -0,0 +1,2 @@ +ALTER TABLE spam_message_incidents + ADD COLUMN IF NOT EXISTS webhook_alerted_at TIMESTAMPTZ; diff --git a/backend/src/db/migrations/022_owner_alert_rules.sql b/backend/src/db/migrations/022_owner_alert_rules.sql new file mode 100644 index 0000000..83e30c2 --- /dev/null +++ b/backend/src/db/migrations/022_owner_alert_rules.sql @@ -0,0 +1,14 @@ +CREATE TABLE IF NOT EXISTS owner_alert_rules ( + id BIGSERIAL PRIMARY KEY, + owner_username TEXT NOT NULL, + node_id TEXT NOT NULL, + rule_type TEXT NOT NULL CHECK (rule_type IN ('offline_minutes', 'battery_below_mv', 'link_loss_above_db')), + threshold DOUBLE PRECISION NOT NULL, + channels JSONB NOT NULL DEFAULT '{}'::jsonb, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + last_triggered_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (owner_username, node_id, rule_type) +); +CREATE INDEX IF NOT EXISTS owner_alert_rules_enabled_idx ON owner_alert_rules(enabled, updated_at);