mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-09-01 16:48:19 +00:00
gadgethd
87239bd4b8
UpsertIATADetails was a plain UPDATE that silently affected zero rows when the iata_codes row did not yet exist. Adding a new IATA to the config for the first time resulted in display_name, approx_lat, and approx_lng remaining NULL. Change the query to INSERT ... ON CONFLICT DO UPDATE so the row is atomically created (if missing) or updated (if present), matching the function's name and the pattern used by UpsertTransportScope.
939 lines
33 KiB
SQL
939 lines
33 KiB
SQL
-- Copyright 2026 Beacon Contributors
|
|
-- SPDX-License-Identifier: agpl
|
|
|
|
-- ============================================================
|
|
-- IATA CODES
|
|
-- ============================================================
|
|
|
|
-- name: UpsertIATA :exec
|
|
INSERT INTO iata_codes (iata)
|
|
VALUES ($1)
|
|
ON CONFLICT (iata) DO NOTHING;
|
|
|
|
-- name: GetIATA :one
|
|
SELECT * FROM iata_codes WHERE iata = $1;
|
|
|
|
-- name: ListIATAs :many
|
|
SELECT * FROM iata_codes ORDER BY iata;
|
|
|
|
-- name: UpsertIATADetails :exec
|
|
INSERT INTO iata_codes (iata, display_name, approx_lat, approx_lng)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (iata) DO UPDATE SET
|
|
display_name = EXCLUDED.display_name,
|
|
approx_lat = EXCLUDED.approx_lat,
|
|
approx_lng = EXCLUDED.approx_lng;
|
|
|
|
-- ============================================================
|
|
-- TRANSPORT CODES
|
|
-- ============================================================
|
|
|
|
-- name: UpsertTransportScope :exec
|
|
INSERT INTO transport_scopes (name, display_name, transport_key, key_fingerprint)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (name) DO UPDATE SET
|
|
display_name = EXCLUDED.display_name,
|
|
transport_key = EXCLUDED.transport_key,
|
|
key_fingerprint = EXCLUDED.key_fingerprint;
|
|
|
|
-- name: GetTransportScopes :many
|
|
SELECT name, transport_key, key_fingerprint FROM transport_scopes ORDER BY name;
|
|
|
|
-- name: GetTransportScopeByName :one
|
|
SELECT id FROM transport_scopes WHERE name = $1;
|
|
|
|
-- name: GetScopeNames :many
|
|
SELECT name FROM transport_scopes ORDER BY name;
|
|
|
|
-- name: GetScopesByIATAs :many
|
|
SELECT
|
|
ts.name,
|
|
COUNT(DISTINCT os.observer_id) AS observer_count,
|
|
COUNT(DISTINCT n.id) AS node_count,
|
|
COUNT(DISTINCT po.iata) AS iata_count
|
|
FROM transport_scopes ts
|
|
LEFT JOIN observer_scopes os ON os.scope_id = ts.id
|
|
LEFT JOIN observers o ON o.id = os.observer_id
|
|
LEFT JOIN packet_observations po ON po.observer_id = o.id
|
|
LEFT JOIN nodes n ON n.default_scope_id = ts.id
|
|
WHERE ($1::text = '' OR po.iata = ANY(string_to_array($1::text, ',')))
|
|
GROUP BY ts.name
|
|
ORDER BY ts.name;
|
|
|
|
-- name: GetScopeByName :one
|
|
SELECT
|
|
ts.name,
|
|
COUNT(DISTINCT p.packet_hash) AS packet_count,
|
|
COUNT(DISTINCT os.observer_id) AS observer_count,
|
|
COUNT(DISTINCT n.id) AS node_count,
|
|
COUNT(DISTINCT po.iata) AS iata_count,
|
|
array_remove(array_agg(DISTINCT po.iata ORDER BY po.iata), NULL)::text[] AS iatas
|
|
FROM transport_scopes ts
|
|
LEFT JOIN packets p ON p.scope_id = ts.id
|
|
LEFT JOIN observer_scopes os ON os.scope_id = ts.id
|
|
LEFT JOIN observers o ON o.id = os.observer_id
|
|
LEFT JOIN packet_observations po ON po.observer_id = o.id
|
|
LEFT JOIN nodes n ON n.default_scope_id = ts.id
|
|
WHERE ts.name = $1
|
|
GROUP BY ts.name;
|
|
|
|
-- ============================================================
|
|
-- OBSERVERS
|
|
-- ============================================================
|
|
|
|
-- name: UpsertObserver :one
|
|
INSERT INTO observers (public_key, observer_type, last_seen)
|
|
VALUES ($1, 'unknown', NOW())
|
|
ON CONFLICT (public_key) DO UPDATE SET
|
|
last_seen = NOW(),
|
|
observation_count = observers.observation_count + 1
|
|
RETURNING *;
|
|
|
|
-- name: UpdateObserverStatus :one
|
|
UPDATE observers SET
|
|
display_name = COALESCE(NULLIF($2, ''), display_name),
|
|
observer_type = COALESCE(NULLIF($3, ''), observer_type),
|
|
software_version = COALESCE($4, software_version),
|
|
hardware_model = COALESCE($5, hardware_model),
|
|
firmware_version = COALESCE($6, firmware_version),
|
|
firmware_build = COALESCE($7, firmware_build),
|
|
radio_freq_mhz = COALESCE($8, radio_freq_mhz),
|
|
radio_sf = COALESCE($9, radio_sf),
|
|
radio_bw_khz = COALESCE($10, radio_bw_khz),
|
|
radio_cr = COALESCE($11, radio_cr),
|
|
battery_level = COALESCE($12, battery_level),
|
|
uptime_seconds = COALESCE($13, uptime_seconds),
|
|
status_metadata = $14,
|
|
last_status_at = NOW(),
|
|
last_seen = NOW()
|
|
WHERE public_key = $1
|
|
RETURNING id;
|
|
|
|
-- name: UpsertObserverScope :exec
|
|
INSERT INTO observer_scopes (observer_id, scope_id, last_seen)
|
|
VALUES ($1, $2, NOW())
|
|
ON CONFLICT (observer_id, scope_id) DO UPDATE SET
|
|
last_seen = NOW();
|
|
|
|
-- name: GetObserverScopes :many
|
|
SELECT ts.name FROM observer_scopes os
|
|
JOIN transport_scopes ts ON ts.id = os.scope_id
|
|
WHERE os.observer_id = $1
|
|
ORDER BY ts.name;
|
|
|
|
-- name: GetObserverByPubkey :one
|
|
SELECT * FROM observers WHERE public_key = $1;
|
|
|
|
-- name: GetObserverByID :one
|
|
SELECT * FROM observers WHERE id = $1;
|
|
|
|
-- name: GetObserverBrokers :many
|
|
SELECT broker_name, last_seen, last_packet_at
|
|
FROM observer_brokers
|
|
WHERE observer_id = $1
|
|
ORDER BY last_seen DESC;
|
|
|
|
-- name: ListObservers :many
|
|
-- Pass cursor=0 to start from the beginning, or the last seen observer's rownum for pagination.
|
|
-- Note: observers use UUID PKs so we order by last_seen and use a keyset on last_seen+id.
|
|
SELECT
|
|
o.id,
|
|
o.display_name,
|
|
o.observer_type,
|
|
o.last_status_at,
|
|
o.radio_freq_mhz,
|
|
o.radio_sf,
|
|
o.radio_bw_khz,
|
|
array_remove(array_agg(DISTINCT ts.name ORDER BY ts.name), NULL)::text[] AS scopes,
|
|
COALESCE(CASE
|
|
WHEN o.last_status_at > NOW() - INTERVAL '5 minutes' THEN 'online'
|
|
ELSE 'offline'
|
|
END, 'offline')::text AS status,
|
|
COALESCE((
|
|
SELECT po.iata
|
|
FROM packet_observations po
|
|
WHERE po.observer_id = o.id
|
|
ORDER BY po.heard_at DESC
|
|
LIMIT 1
|
|
), '')::text AS iata
|
|
FROM observers o
|
|
LEFT JOIN observer_brokers ob ON ob.observer_id = o.id
|
|
LEFT JOIN observer_scopes os ON os.observer_id = o.id
|
|
LEFT JOIN transport_scopes ts ON ts.id = os.scope_id
|
|
WHERE
|
|
($1::text = '' OR (
|
|
SELECT po.iata FROM packet_observations po
|
|
WHERE po.observer_id = o.id
|
|
ORDER BY po.heard_at DESC LIMIT 1
|
|
) = ANY(string_to_array($1::text, ',')))
|
|
AND ($2 = '' OR o.observer_type = $2)
|
|
AND ($3 = '' OR ob.broker_name = $3)
|
|
AND ($4 = '' OR CASE
|
|
WHEN o.last_status_at > NOW() - INTERVAL '5 minutes' THEN 'online'
|
|
ELSE 'offline'
|
|
END = $4)
|
|
AND ($5 = '' OR o.display_name ILIKE '%' || $5 || '%')
|
|
AND ($6::timestamptz IS NULL OR o.last_seen < $6)
|
|
AND ($8::text = '' OR EXISTS (
|
|
SELECT 1 FROM observer_scopes os2
|
|
JOIN transport_scopes ts2 ON ts2.id = os2.scope_id
|
|
WHERE os2.observer_id = o.id AND ts2.name = $8::text
|
|
))
|
|
GROUP BY o.id
|
|
ORDER BY o.last_seen DESC
|
|
LIMIT $7;
|
|
|
|
-- name: GetObserverLastIATA :one
|
|
SELECT iata FROM packet_observations
|
|
WHERE observer_id = $1
|
|
ORDER BY heard_at DESC
|
|
LIMIT 1;
|
|
|
|
-- name: GetObserverRadio :one
|
|
SELECT radio_freq_mhz, radio_bw_khz, radio_sf, radio_cr
|
|
FROM observers
|
|
WHERE id = $1;
|
|
|
|
-- name: InsertObserverTelemetry :exec
|
|
-- Inserts a telemetry snapshot for an observer. The reported_at timestamp should
|
|
-- be truncated to the configured resolution before calling to ensure deduplication.
|
|
INSERT INTO observer_telemetry (
|
|
observer_id, reported_at, battery_voltage_mv, airtime_tx_pct,
|
|
airtime_rx_pct, noise_floor_db, uptime_seconds, queue_length,
|
|
debug_flags, receive_errors
|
|
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
|
|
ON CONFLICT (observer_id, reported_at) DO NOTHING;
|
|
|
|
-- name: GetObserverTelemetry :many
|
|
SELECT id, reported_at, battery_voltage_mv, airtime_tx_pct, airtime_rx_pct,
|
|
noise_floor_db, uptime_seconds, queue_length, debug_flags, receive_errors
|
|
FROM observer_telemetry
|
|
WHERE observer_id = $1
|
|
AND ($2::timestamptz IS NULL OR reported_at >= $2)
|
|
AND ($3::timestamptz IS NULL OR reported_at <= $3)
|
|
AND ($4 = 0 OR id > $4)
|
|
ORDER BY reported_at ASC;
|
|
|
|
-- name: GetObserverTelemetryBucketed :many
|
|
SELECT
|
|
(date_trunc('day', reported_at) +
|
|
(EXTRACT(HOUR FROM reported_at)::int / $4::int) * ($4::int * interval '1 hour'))::timestamptz AS bucket,
|
|
AVG(battery_voltage_mv)::int AS battery_voltage_mv,
|
|
GREATEST(MAX(airtime_tx_pct) - MIN(airtime_tx_pct), 0)::real AS airtime_tx_pct,
|
|
GREATEST(MAX(airtime_rx_pct) - MIN(airtime_rx_pct), 0)::real AS airtime_rx_pct,
|
|
AVG(noise_floor_db)::real AS noise_floor_db,
|
|
MAX(uptime_seconds)::bigint AS uptime_seconds,
|
|
AVG(queue_length)::int AS queue_length,
|
|
GREATEST(MAX(receive_errors) - MIN(receive_errors), 0)::int AS receive_errors
|
|
FROM observer_telemetry
|
|
WHERE observer_id = $1
|
|
AND ($2::timestamptz IS NULL OR reported_at >= $2)
|
|
AND ($3::timestamptz IS NULL OR reported_at <= $3)
|
|
GROUP BY bucket
|
|
ORDER BY bucket ASC;
|
|
|
|
-- name: ListObserverAdverts :many
|
|
-- Returns advert packets (payload_type=4) heard by a specific observer.
|
|
-- Pass cursor=0 to start from the beginning, or the last seen id for pagination.
|
|
SELECT
|
|
po.id,
|
|
encode(po.packet_hash, 'hex') AS packet_hash_hex,
|
|
p.payload_type,
|
|
po.iata,
|
|
po.heard_at,
|
|
po.rssi,
|
|
po.snr,
|
|
po.hop_count,
|
|
n.name AS node_name,
|
|
encode(p.origin_pubkey, 'hex') AS node_public_key
|
|
FROM packet_observations po
|
|
JOIN packets p ON p.packet_hash = po.packet_hash
|
|
LEFT JOIN nodes n ON n.public_key = p.origin_pubkey
|
|
WHERE po.observer_id = $1
|
|
AND p.payload_type = 4
|
|
AND ($2 = 0 OR po.id > $2)
|
|
ORDER BY po.id ASC
|
|
LIMIT $3;
|
|
|
|
-- name: DeleteOldTelemetry :exec
|
|
-- Deletes telemetry rows older than the given cutoff. Called by the cleanup goroutine.
|
|
DELETE FROM observer_telemetry WHERE reported_at < $1;
|
|
|
|
-- ============================================================
|
|
-- OBSERVER BROKERS
|
|
-- ============================================================
|
|
|
|
-- name: UpsertObserverBroker :exec
|
|
INSERT INTO observer_brokers (observer_id, broker_name, last_seen, last_packet_at)
|
|
VALUES ($1, $2, NOW(), NOW())
|
|
ON CONFLICT (observer_id, broker_name) DO UPDATE SET
|
|
last_seen = NOW(),
|
|
last_packet_at = NOW();
|
|
|
|
-- ============================================================
|
|
-- PACKETS
|
|
-- ============================================================
|
|
|
|
-- name: UpsertPacket :one
|
|
INSERT INTO packets (
|
|
packet_hash,
|
|
payload_type,
|
|
payload_version,
|
|
route_type,
|
|
transport_codes_present,
|
|
region_code,
|
|
sub_region_code,
|
|
origin_pubkey,
|
|
raw_payload,
|
|
raw_header,
|
|
parsed_payload,
|
|
channel_hash,
|
|
scope_id,
|
|
trace_tag,
|
|
first_heard_at,
|
|
last_heard_at
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, NOW(), NOW()
|
|
)
|
|
ON CONFLICT (packet_hash) DO UPDATE SET
|
|
last_heard_at = NOW()
|
|
RETURNING packet_hash, payload_type, payload_version, route_type, transport_codes_present, region_code, sub_region_code, origin_pubkey, raw_payload, raw_header, parsed_payload, decrypted, channel_hash, first_heard_at, last_heard_at, (xmax = 0)
|
|
AS inserted;
|
|
|
|
-- name: SetPacketDecrypted :exec
|
|
UPDATE packets SET decrypted = true WHERE packet_hash = $1;
|
|
|
|
-- name: GetPacketByHash :one
|
|
SELECT p.*, ts.name AS scope_name,
|
|
cm.sender_name AS cm_sender_name,
|
|
cm.content AS cm_content,
|
|
cm.sent_at AS cm_sent_at
|
|
FROM packets p
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
LEFT JOIN channel_messages cm ON cm.packet_hash = p.packet_hash
|
|
WHERE p.packet_hash = $1;
|
|
|
|
-- name: GetPacketsByTraceTag :many
|
|
-- Returns all packets for a given trace tag with observations.
|
|
SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex,
|
|
p.route_type,
|
|
p.first_heard_at,
|
|
p.last_heard_at,
|
|
p.parsed_payload,
|
|
p.scope_id,
|
|
ts.name AS scope_name
|
|
FROM packets p
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE p.trace_tag = decode($1, 'hex')
|
|
ORDER BY p.first_heard_at ASC;
|
|
|
|
-- name: GetPacketObservationCount :one
|
|
SELECT COUNT(*) FROM packet_observations WHERE packet_hash = $1;
|
|
|
|
-- name: ListPackets :many
|
|
-- Returns packets with the latest observation rolled in for display.
|
|
-- Pass cursor=0 to start from the beginning.
|
|
SELECT
|
|
p.packet_hash,
|
|
p.payload_type,
|
|
p.route_type,
|
|
p.first_heard_at,
|
|
p.last_heard_at,
|
|
p.scope_id,
|
|
ts.name AS scope_name,
|
|
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count,
|
|
po.observer_id AS latest_observer_id,
|
|
o.display_name AS latest_observer_name,
|
|
po.iata AS latest_observer_iata
|
|
FROM packets p
|
|
LEFT JOIN LATERAL (
|
|
SELECT observer_id, iata
|
|
FROM packet_observations
|
|
WHERE packet_hash = p.packet_hash
|
|
ORDER BY heard_at DESC
|
|
LIMIT 1
|
|
) po ON true
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE
|
|
($1::smallint = -1 OR p.payload_type = $1::smallint)
|
|
AND ($2::smallint = -1 OR p.route_type = $2::smallint)
|
|
AND ($3::text = '' OR EXISTS (
|
|
SELECT 1 FROM packet_observations po3
|
|
WHERE po3.packet_hash = p.packet_hash
|
|
AND po3.iata = ANY(string_to_array($3::text, ','))
|
|
))
|
|
AND ($4::timestamptz IS NULL OR p.first_heard_at >= $4)
|
|
AND ($5::timestamptz IS NULL OR p.first_heard_at <= $5)
|
|
AND ($6::timestamptz IS NULL OR p.last_heard_at < $6)
|
|
AND ($8::text = '' OR ts.name = $8::text)
|
|
ORDER BY p.last_heard_at DESC
|
|
LIMIT $7;
|
|
|
|
-- name: ListPacketsAfterID :many
|
|
-- Returns packets with observations after the given observation ID, ordered oldest first.
|
|
-- Used for WS reconnect backfill. Pass afterObservationId=0 to start from the beginning.
|
|
SELECT
|
|
p.packet_hash,
|
|
p.payload_type,
|
|
p.route_type,
|
|
p.first_heard_at,
|
|
p.last_heard_at,
|
|
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count,
|
|
po.observer_id AS latest_observer_id,
|
|
o.display_name AS latest_observer_name,
|
|
po.iata AS latest_observer_iata,
|
|
ts.name AS scope_name
|
|
FROM packets p
|
|
JOIN packet_observations po ON po.packet_hash = p.packet_hash
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE po.id > $1
|
|
AND ($2::smallint = -1 OR p.payload_type = $2::smallint)
|
|
AND ($3::smallint = -1 OR p.route_type = $3::smallint)
|
|
AND ($4::text = '' OR po.iata = ANY(string_to_array($4::text, ',')))
|
|
AND ($5::text = '' OR ts.name = $5::text)
|
|
ORDER BY po.id ASC
|
|
LIMIT $6;
|
|
|
|
|
|
-- name: DeleteOldPackets :exec
|
|
-- Deletes packets and their observations older than the given cutoff.
|
|
-- packet_observations cascade-delete via FK.
|
|
DELETE FROM packets WHERE last_heard_at < $1;
|
|
|
|
-- ============================================================
|
|
-- PACKET OBSERVATIONS
|
|
-- ============================================================
|
|
|
|
-- name: InsertObservation :one
|
|
INSERT INTO packet_observations (
|
|
packet_hash,
|
|
observer_id,
|
|
iata,
|
|
heard_at,
|
|
path_length_byte,
|
|
hash_size,
|
|
hop_count,
|
|
path_bytes,
|
|
rssi,
|
|
snr,
|
|
propagation_time_ms,
|
|
radio_freq_mhz,
|
|
spread_factor,
|
|
bandwidth_khz,
|
|
coding_rate,
|
|
source_broker
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16
|
|
)
|
|
ON CONFLICT (packet_hash, observer_id, heard_at) DO NOTHING
|
|
RETURNING *;
|
|
|
|
-- name: ListObservationsForPacket :many
|
|
SELECT po.*, o.display_name AS observer_name
|
|
FROM packet_observations po
|
|
LEFT JOIN observers o ON o.id = po.observer_id
|
|
WHERE po.packet_hash = $1
|
|
ORDER BY po.heard_at ASC;
|
|
|
|
-- ============================================================
|
|
-- NODES
|
|
-- ============================================================
|
|
|
|
-- name: UpsertNode :one
|
|
INSERT INTO nodes (public_key, node_type, name, latitude, longitude, location_source, last_advert_at, last_seen, radio_freq_mhz, radio_sf, radio_bw_khz)
|
|
VALUES ($1, $2, $3, $4, $5, 'advert', NOW(), NOW(), $6, $7, $8)
|
|
ON CONFLICT (public_key) DO UPDATE SET
|
|
node_type = EXCLUDED.node_type,
|
|
name = COALESCE(EXCLUDED.name, nodes.name),
|
|
latitude = COALESCE(EXCLUDED.latitude, nodes.latitude),
|
|
longitude = COALESCE(EXCLUDED.longitude, nodes.longitude),
|
|
location_source = CASE WHEN EXCLUDED.latitude IS NOT NULL THEN 'advert' ELSE nodes.location_source END,
|
|
last_advert_at = NOW(),
|
|
last_seen = NOW(),
|
|
radio_freq_mhz = EXCLUDED.radio_freq_mhz,
|
|
radio_sf = EXCLUDED.radio_sf,
|
|
radio_bw_khz = EXCLUDED.radio_bw_khz
|
|
RETURNING *;
|
|
|
|
-- name: SetNodeMultibytePaths :exec
|
|
UPDATE nodes SET supports_multibyte_paths = TRUE
|
|
WHERE id = $1 AND supports_multibyte_paths = FALSE;
|
|
|
|
-- name: SetNodeMultibyteTraces :exec
|
|
UPDATE nodes SET supports_multibyte_traces = TRUE
|
|
WHERE id = $1 AND supports_multibyte_traces = FALSE;
|
|
|
|
-- name: SetNodeDefaultScope :exec
|
|
UPDATE nodes SET default_scope_id = $2 WHERE id = $1;
|
|
|
|
-- name: GetNodeByID :one
|
|
SELECT n.*, ts.name AS default_scope_name,
|
|
EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer,
|
|
(SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id,
|
|
(SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC)
|
|
FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas,
|
|
(SELECT COUNT(*) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count
|
|
FROM nodes n
|
|
LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id
|
|
WHERE n.id = $1;
|
|
|
|
-- name: GetNodesByIDs :many
|
|
SELECT id, public_key, name, latitude, longitude
|
|
FROM nodes
|
|
WHERE id = ANY($1::uuid[]);
|
|
|
|
-- name: ListNodes :many
|
|
SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.last_seen,
|
|
n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz,
|
|
ts.name AS default_scope_name,
|
|
json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FILTER (WHERE ni.iata IS NOT NULL) AS iatas,
|
|
EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer,
|
|
(SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id,
|
|
(SELECT COUNT(*) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count
|
|
FROM nodes n
|
|
LEFT JOIN node_iatas ni ON ni.node_id = n.id
|
|
LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id
|
|
WHERE
|
|
($1 = 0 OR n.node_type = $1)
|
|
AND ($2::text = '' OR n.id IN (SELECT node_id FROM node_iatas WHERE iata = ANY(string_to_array($2::text, ','))))
|
|
AND (
|
|
$3::text = 'any'
|
|
OR ($3::text = 'true' AND n.supports_multibyte_paths = TRUE)
|
|
OR ($3::text = 'false' AND n.supports_multibyte_paths = FALSE)
|
|
)
|
|
AND (
|
|
$4::text = 'any'
|
|
OR ($4::text = 'true' AND n.supports_multibyte_traces = TRUE)
|
|
OR ($4::text = 'false' AND n.supports_multibyte_traces = FALSE)
|
|
)
|
|
AND ($5::bytea IS NULL OR n.public_key = $5)
|
|
AND ($6 = '' OR n.name ILIKE '%' || $6 || '%')
|
|
AND ($7::timestamptz IS NULL OR n.last_seen < $7)
|
|
AND ($9::text = '' OR ts.name = $9::text)
|
|
GROUP BY n.id, ts.name
|
|
ORDER BY n.last_seen DESC
|
|
LIMIT $8;
|
|
|
|
-- name: ListNodeObservations :many
|
|
SELECT po.id, encode(po.packet_hash, 'hex') AS packet_hash_hex,
|
|
p.payload_type, po.iata, po.heard_at, po.rssi, po.snr, po.hop_count
|
|
FROM packet_observations po
|
|
JOIN packets p ON p.packet_hash = po.packet_hash
|
|
JOIN nodes n ON n.public_key = p.origin_pubkey
|
|
WHERE n.id = $1
|
|
AND ($2 = 0 OR po.id < $2)
|
|
ORDER BY po.id DESC
|
|
LIMIT $3;
|
|
|
|
-- ============================================================
|
|
-- NODE IATAS
|
|
-- ============================================================
|
|
|
|
-- name: UpsertNodeIATA :exec
|
|
INSERT INTO node_iatas (node_id, iata, last_heard, observation_count)
|
|
VALUES ($1, $2, NOW(), 1)
|
|
ON CONFLICT (node_id, iata) DO UPDATE SET
|
|
last_heard = NOW(),
|
|
observation_count = node_iatas.observation_count + 1;
|
|
|
|
-- name: UpsertNodeShortID :exec
|
|
INSERT INTO node_short_ids (node_id, iata, prefix_4)
|
|
VALUES ($1, $2, $3)
|
|
ON CONFLICT (node_id, iata) DO NOTHING;
|
|
|
|
-- ============================================================
|
|
-- CHANNELS
|
|
-- ============================================================
|
|
|
|
-- name: UpsertChannel :one
|
|
-- Upsert a channel by (hash, key_fingerprint). Pass NULL fingerprint for
|
|
-- hash-only records (key unknown). Returns the channel row.
|
|
INSERT INTO channels (channel_hash, key_fingerprint, name, hashtag, is_hashtag, key_known, last_seen)
|
|
VALUES ($1, $2::bytea, $3, $4, $5, ($2 IS NOT NULL), NOW())
|
|
ON CONFLICT (channel_hash, key_fingerprint) DO UPDATE SET
|
|
last_seen = NOW(),
|
|
name = COALESCE(EXCLUDED.name, channels.name),
|
|
message_count = CASE WHEN $6 THEN channels.message_count + 1 ELSE channels.message_count END
|
|
RETURNING *;
|
|
|
|
-- name: UpsertChannelHashOnly :one
|
|
INSERT INTO channels (channel_hash, last_seen)
|
|
VALUES ($1, NOW())
|
|
ON CONFLICT (channel_hash) WHERE key_fingerprint IS NULL DO UPDATE SET
|
|
last_seen = NOW()
|
|
RETURNING id;
|
|
|
|
-- name: ListChannels :many
|
|
-- Returns channels ordered by last seen, optionally filtered by hash and/or IATA.
|
|
-- Pass NULL for hash to skip hash filtering. Pass empty string for iata to skip IATA filtering.
|
|
-- IATA filter returns channels that have active packets in that IATA (case-insensitive).
|
|
-- Pass cursor=0 to start from the beginning (cursor is last_seen epoch ms).
|
|
SELECT DISTINCT c.* FROM channels c
|
|
WHERE ($1::bytea IS NULL OR c.channel_hash = $1)
|
|
AND ($2 = '' OR EXISTS (
|
|
SELECT 1 FROM packets p
|
|
JOIN packet_observations po ON po.packet_hash = p.packet_hash
|
|
WHERE p.channel_hash = c.channel_hash
|
|
AND po.iata ILIKE $2
|
|
))
|
|
AND ($3::timestamptz IS NULL OR c.last_seen < $3)
|
|
ORDER BY c.last_seen DESC
|
|
LIMIT $4;
|
|
|
|
-- name: GetChannelByID :one
|
|
SELECT * FROM channels WHERE id = $1;
|
|
|
|
|
|
-- ============================================================
|
|
-- CHANNEL MESSAGES
|
|
-- ============================================================
|
|
|
|
-- name: InsertChannelMessage :one
|
|
INSERT INTO channel_messages (channel_id, packet_hash, sender_name, content, sent_at)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
ON CONFLICT (packet_hash) DO NOTHING
|
|
RETURNING id;
|
|
|
|
-- name: ListChannelMessages :many
|
|
-- Returns messages for a channel identified by integer ID.
|
|
-- Pass a zero/null timestamp for since to return all messages up to limit.
|
|
-- Pass empty string for iata to skip IATA filtering.
|
|
-- Pass cursor=0 to start from the beginning.
|
|
SELECT DISTINCT ON (cm.id) cm.*, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash,
|
|
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = cm.packet_hash) AS observation_count
|
|
FROM channel_messages cm
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE cm.channel_id = $1
|
|
AND ($2::timestamptz IS NULL OR cm.sent_at >= $2)
|
|
AND ($3::text = '' OR po.iata = ANY(string_to_array($3::text, ',')))
|
|
AND ($4::text = '' OR ts.name = $4::text)
|
|
AND ($5::bigint = 0 OR cm.id < $5::bigint)
|
|
ORDER BY cm.id DESC
|
|
LIMIT $6;
|
|
|
|
-- name: ListAllChannelMessages :many
|
|
-- Returns all messages across all channels with optional time, IATA, scope and cursor filters.
|
|
-- Pass empty string for iata or scope to skip those filters.
|
|
-- Pass cursor=0 to start from the beginning.
|
|
SELECT DISTINCT ON (cm.id) cm.*, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash,
|
|
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = cm.packet_hash) AS observation_count
|
|
FROM channel_messages cm
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE ($1::timestamptz IS NULL OR cm.sent_at >= $1)
|
|
AND ($2::text = '' OR po.iata = ANY(string_to_array($2::text, ',')))
|
|
AND ($3::text = '' OR ts.name = $3::text)
|
|
AND ($4 = 0 OR cm.id < $4)
|
|
ORDER BY cm.id DESC
|
|
LIMIT $5;
|
|
|
|
-- name: ListChannelMessagesByHash :many
|
|
-- Returns messages for all channels matching a hash byte.
|
|
-- May return messages from multiple channels if the hash collides across different keys.
|
|
-- Pass empty string for iata or scope to skip those filters.
|
|
-- Pass cursor=0 to start from the beginning.
|
|
SELECT DISTINCT ON (cm.id) cm.*, c.channel_hash,
|
|
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = cm.packet_hash) AS observation_count
|
|
FROM channel_messages cm
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE c.channel_hash = $1
|
|
AND ($2::timestamptz IS NULL OR cm.sent_at >= $2)
|
|
AND ($3::text = '' OR po.iata = ANY(string_to_array($3::text, ',')))
|
|
AND ($4::text = '' OR ts.name = $4::text)
|
|
AND ($5::bigint = 0 OR cm.id < $5::bigint)
|
|
ORDER BY cm.id DESC
|
|
LIMIT $6;
|
|
|
|
-- name: ListMessagesAfterID :many
|
|
-- Returns messages after the given message ID, ordered oldest first.
|
|
-- Used for WS reconnect backfill.
|
|
SELECT DISTINCT ON (cm.id) cm.*, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash,
|
|
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = cm.packet_hash) AS observation_count
|
|
FROM channel_messages cm
|
|
JOIN channels c ON c.id = cm.channel_id
|
|
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
|
|
JOIN packets p ON p.packet_hash = cm.packet_hash
|
|
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
|
|
WHERE cm.id > $1
|
|
AND ($2::text = '' OR po.iata = ANY(string_to_array($2::text, ',')))
|
|
AND ($3::text = '' OR ts.name = $3::text)
|
|
ORDER BY cm.id ASC
|
|
LIMIT $4;
|
|
|
|
-- ============================================================
|
|
-- STATS
|
|
-- ============================================================
|
|
|
|
-- name: GetStatsOverview :one
|
|
SELECT
|
|
COUNT(DISTINCT po.packet_hash) AS total_packets,
|
|
COUNT(*) AS total_observations,
|
|
COUNT(DISTINCT po.observer_id) AS active_observers,
|
|
COUNT(DISTINCT po.iata) AS active_iatas
|
|
FROM packet_observations po
|
|
WHERE po.heard_at > NOW() - INTERVAL '24 hours'
|
|
AND ($1::text = '' OR po.iata = ANY(string_to_array($1::text, ',')));
|
|
|
|
-- name: GetHourlyStats :many
|
|
SELECT iata, hour, observation_count, unique_packets, active_observers
|
|
FROM mv_hourly_iata_stats
|
|
WHERE ($1::text = '' OR iata = ANY(string_to_array($1::text, ',')))
|
|
AND hour >= NOW() - $2::interval
|
|
ORDER BY iata, hour;
|
|
|
|
-- name: GetTopNodes :many
|
|
SELECT * FROM mv_top_nodes_by_iata
|
|
WHERE ($1::text = '' OR iata = ANY(string_to_array($1::text, ',')))
|
|
ORDER BY observation_count DESC
|
|
LIMIT $2;
|
|
|
|
-- name: GetStatsPayloadBreakdown :many
|
|
-- Returns observation counts grouped by payload type for the given window and IATA.
|
|
SELECT
|
|
p.payload_type,
|
|
COUNT(*) AS count
|
|
FROM packet_observations po
|
|
JOIN packets p ON p.packet_hash = po.packet_hash
|
|
WHERE po.heard_at > $1
|
|
AND ($2::text = '' OR po.iata = ANY(string_to_array($2::text, ',')))
|
|
GROUP BY p.payload_type
|
|
ORDER BY count DESC;
|
|
|
|
-- name: GetStatsNodeTypes :many
|
|
-- Returns node counts grouped by type, optionally filtered by IATA.
|
|
SELECT
|
|
n.node_type,
|
|
COUNT(DISTINCT n.id)::bigint AS count
|
|
FROM nodes n
|
|
LEFT JOIN node_iatas ni ON ni.node_id = n.id
|
|
WHERE ($1::text = '' OR ni.iata = ANY(string_to_array($1::text, ',')))
|
|
GROUP BY n.node_type
|
|
ORDER BY count DESC;
|
|
|
|
-- name: GetStatsTopObservers :many
|
|
-- Returns the top N observers by observation count for the given window and IATA.
|
|
SELECT
|
|
o.id,
|
|
o.display_name,
|
|
o.observer_type,
|
|
COUNT(*) AS observation_count,
|
|
COALESCE((
|
|
SELECT po2.iata FROM packet_observations po2
|
|
WHERE po2.observer_id = o.id
|
|
ORDER BY po2.heard_at DESC LIMIT 1
|
|
), '') AS iata
|
|
FROM packet_observations po
|
|
JOIN observers o ON o.id = po.observer_id
|
|
WHERE po.heard_at > $1
|
|
AND ($2::text = '' OR po.iata = ANY(string_to_array($2::text, ',')))
|
|
GROUP BY o.id
|
|
ORDER BY observation_count DESC
|
|
LIMIT $3;
|
|
|
|
-- name: GetRadioPresets :many
|
|
SELECT preset, iata, source_type, count
|
|
FROM mv_radio_presets
|
|
WHERE ($1::text = '' OR preset = $1::text)
|
|
AND ($2::text = '' OR iata = ANY(string_to_array($2::text, ',')))
|
|
ORDER BY preset, iata, source_type;
|
|
|
|
-- name: GetScopeStats :many
|
|
SELECT
|
|
ts.name,
|
|
COUNT(DISTINCT p.packet_hash) AS packet_count,
|
|
COUNT(DISTINCT os.observer_id) AS observer_count,
|
|
COUNT(DISTINCT n.id) AS node_count
|
|
FROM transport_scopes ts
|
|
LEFT JOIN packets p ON p.scope_id = ts.id
|
|
LEFT JOIN observer_scopes os ON os.scope_id = ts.id
|
|
LEFT JOIN nodes n ON n.default_scope_id = ts.id
|
|
GROUP BY ts.name
|
|
ORDER BY ts.name;
|
|
|
|
-- ============================================================
|
|
-- REGIONS
|
|
-- ============================================================
|
|
|
|
-- name: ListRegions :many
|
|
SELECT id, slug, name
|
|
FROM regions
|
|
ORDER BY display_order, name;
|
|
|
|
-- name: GetRegion :one
|
|
SELECT id, slug, name, description, center_lat, center_lng, zoom_level
|
|
FROM regions
|
|
WHERE id = $1;
|
|
|
|
-- name: GetRegionBySlug :one
|
|
SELECT id, slug, name, description, center_lat, center_lng, zoom_level
|
|
FROM regions
|
|
WHERE slug = $1;
|
|
|
|
-- name: GetRegionIATAs :many
|
|
SELECT iata FROM region_iatas
|
|
WHERE region_id = $1
|
|
ORDER BY iata;
|
|
|
|
-- name: UpsertRegion :one
|
|
INSERT INTO regions (slug, name, description, display_order, center_lat, center_lng, zoom_level, updated_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())
|
|
ON CONFLICT (slug) DO UPDATE SET
|
|
name = EXCLUDED.name,
|
|
description = EXCLUDED.description,
|
|
display_order = EXCLUDED.display_order,
|
|
center_lat = EXCLUDED.center_lat,
|
|
center_lng = EXCLUDED.center_lng,
|
|
zoom_level = EXCLUDED.zoom_level,
|
|
updated_at = NOW()
|
|
RETURNING id;
|
|
|
|
-- name: UpsertRegionIATA :exec
|
|
INSERT INTO region_iatas (region_id, iata)
|
|
VALUES ($1, $2)
|
|
ON CONFLICT (region_id, iata) DO NOTHING;
|
|
|
|
-- ============================================================
|
|
-- TRACES
|
|
-- ============================================================
|
|
|
|
-- name: ListTraceTags :many
|
|
-- Returns distinct trace tags with summary info, ordered by most recent first.
|
|
SELECT
|
|
encode(p.trace_tag, 'hex') AS trace_tag,
|
|
MIN(p.first_heard_at)::timestamptz AS first_heard_at,
|
|
MAX(p.last_heard_at)::timestamptz AS last_heard_at,
|
|
COUNT(DISTINCT p.packet_hash) AS packet_count,
|
|
COUNT(DISTINCT po.iata) AS iata_count,
|
|
MAX(p.parsed_payload->>'type')::text AS trace_type,
|
|
best.parsed_payload AS best_payload
|
|
FROM packets p
|
|
LEFT JOIN packet_observations po ON po.packet_hash = p.packet_hash
|
|
LEFT JOIN LATERAL (
|
|
SELECT parsed_payload
|
|
FROM packets p2
|
|
WHERE p2.trace_tag = p.trace_tag
|
|
ORDER BY jsonb_array_length(p2.parsed_payload->'pathHashes') DESC
|
|
LIMIT 1
|
|
) best ON true
|
|
WHERE p.trace_tag IS NOT NULL
|
|
AND ($1::text = '' OR po.iata = ANY(string_to_array($1::text, ',')))
|
|
AND ($2::text = '' OR p.scope_id = (SELECT id FROM transport_scopes WHERE name = $2))
|
|
AND ($3::timestamptz IS NULL OR p.first_heard_at >= $3)
|
|
AND ($4::timestamptz IS NULL OR p.first_heard_at <= $4)
|
|
AND ($5::timestamptz IS NULL OR p.last_heard_at < $5)
|
|
AND ($7::text = '' OR p.parsed_payload->>'type' = $7)
|
|
GROUP BY p.trace_tag, best.parsed_payload
|
|
ORDER BY MAX(p.last_heard_at) DESC
|
|
LIMIT $6;
|
|
|
|
-- ============================================================
|
|
-- ROUTES
|
|
-- ============================================================
|
|
|
|
-- name: UpsertKnownRoute :exec
|
|
-- Inserts or updates a known route (all hops resolved to high confidence).
|
|
-- node_ids and hash_prefix are ordered arrays of the resolved node UUIDs and
|
|
-- their hash bytes. last_seen is bumped on conflict.
|
|
INSERT INTO known_routes (node_ids, hash_prefix, iata, hop_count)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (node_ids, iata) DO UPDATE SET
|
|
last_seen = NOW(),
|
|
observation_count = known_routes.observation_count + 1;
|
|
|
|
-- name: ListKnownRoutes :many
|
|
SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count
|
|
FROM known_routes
|
|
WHERE ($1 = '' OR iata = $1)
|
|
AND ($2 = 0 OR hop_count = $2)
|
|
AND ($3::timestamptz IS NULL OR last_seen < $3)
|
|
ORDER BY last_seen DESC
|
|
LIMIT $4;
|
|
|
|
-- name: SearchKnownRoutes :many
|
|
-- Returns known routes containing a subsequence from source to destination hash prefix.
|
|
-- Verifies source appears before destination in the route.
|
|
SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count
|
|
FROM known_routes
|
|
WHERE iata = $1
|
|
AND array_position(hash_prefix, $2::bytea) IS NOT NULL
|
|
AND array_position(hash_prefix, $3::bytea) IS NOT NULL
|
|
AND array_position(hash_prefix, $2::bytea) < array_position(hash_prefix, $3::bytea)
|
|
ORDER BY hop_count ASC, last_seen DESC;
|
|
|
|
-- name: GetKnownRoutesByNode :many
|
|
SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count
|
|
FROM known_routes
|
|
WHERE iata = $1
|
|
AND $2::uuid = ANY(node_ids)
|
|
ORDER BY hop_count ASC, last_seen DESC;
|
|
|
|
-- ============================================================
|
|
-- NEIGHBORS
|
|
-- ============================================================
|
|
|
|
-- name: UpsertNodeNeighbor :exec
|
|
-- Records or updates a neighbor relationship between two nodes observed in the same IATA.
|
|
-- node_id is the advertising node, neighbor_id is the first-hop forwarder.
|
|
INSERT INTO node_neighbors (node_id, neighbor_id, iata, observation_count)
|
|
VALUES ($1, $2, $3, 1)
|
|
ON CONFLICT (node_id, neighbor_id, iata) DO UPDATE SET
|
|
last_seen = NOW(),
|
|
observation_count = node_neighbors.observation_count + 1;
|
|
|
|
-- name: GetNodeNeighbors :many
|
|
-- Returns the neighbors of a node with details, ordered by most recently seen.
|
|
SELECT
|
|
n.id, n.public_key, n.name, n.node_type, n.latitude, n.longitude,
|
|
nn.iata, nn.observation_count, nn.first_seen, nn.last_seen
|
|
FROM node_neighbors nn
|
|
JOIN nodes n ON n.id = nn.neighbor_id
|
|
WHERE nn.node_id = $1
|
|
ORDER BY nn.last_seen DESC;
|
|
|
|
-- name: GetCrossIATANeighbors :many
|
|
-- Returns neighbors of a node that are in a different IATA.
|
|
SELECT
|
|
n.id, n.name, n.node_type, n.latitude, n.longitude,
|
|
nn.iata AS neighbor_iata, nn.observation_count, nn.last_seen
|
|
FROM node_neighbors nn
|
|
JOIN nodes n ON n.id = nn.neighbor_id
|
|
WHERE nn.node_id = $1
|
|
AND nn.iata != $2
|
|
ORDER BY nn.last_seen DESC;
|
|
|
|
-- ============================================================
|
|
-- HELPERS
|
|
-- ============================================================
|
|
|
|
-- name: ResolvePathHashes :many
|
|
SELECT ns.prefix_4 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key
|
|
FROM node_short_ids ns
|
|
JOIN nodes n ON n.id = ns.node_id
|
|
WHERE ns.iata = $1
|
|
AND CASE
|
|
WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 1 THEN ns.prefix_1 = ANY($2)
|
|
WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 2 THEN ns.prefix_2 = ANY($2)
|
|
WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 3 THEN ns.prefix_3 = ANY($2)
|
|
WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 4 THEN ns.prefix_4 = ANY($2)
|
|
ELSE FALSE
|
|
END;
|
|
|
|
-- name: RefreshHourlyStats :exec
|
|
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_hourly_iata_stats;
|
|
|
|
-- name: RefreshTopNodes :exec
|
|
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_top_nodes_by_iata;
|
|
|
|
-- name: RefreshRadioPresets :exec
|
|
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets;
|