mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-09-09 05:33:54 +00:00
init
This commit is contained in:
Enot (ded) Skelly
@@ -0,0 +1 @@
|
||||
.env
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
FROM golang:1.23-alpine AS builder
|
||||
WORKDIR /app
|
||||
COPY go.mod go.sum ./
|
||||
RUN go mod download
|
||||
COPY . .
|
||||
RUN go build -o tower ./cmd/tower
|
||||
|
||||
FROM alpine:3.19
|
||||
RUN apk add --no-cache ca-certificates tzdata
|
||||
WORKDIR /app
|
||||
COPY --from=builder /app/tower .
|
||||
ENTRYPOINT ["./tower"]
|
||||
@@ -0,0 +1,98 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"tower/internal/api/router"
|
||||
"tower/internal/hub"
|
||||
"tower/internal/ingest"
|
||||
|
||||
"github.com/joho/godotenv"
|
||||
)
|
||||
|
||||
func main() {
|
||||
_ = godotenv.Load()
|
||||
addr := os.Getenv("LISTEN_ADDR")
|
||||
if addr == "" {
|
||||
addr = ":8080"
|
||||
}
|
||||
|
||||
// ── Hub ──────────────────────────────────────────────────────────────────
|
||||
h := hub.New()
|
||||
go h.Run()
|
||||
|
||||
// ── MQTT ingest workers ──────────────────────────────────────────────────
|
||||
// TODO: wire in a real DB handle and ChannelKeyStore once those are built.
|
||||
// For now the workers start but immediately hit the decode TODO stub.
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
broker1 := ingest.New(
|
||||
ingest.Config{
|
||||
BrokerName: "mqtt1",
|
||||
URL: mustEnv("MQTT_BROKER_1_URL"),
|
||||
Username: mustEnv("MQTT_BROKER_1_USERNAME"),
|
||||
Password: mustEnv("MQTT_BROKER_1_PASSWORD"),
|
||||
},
|
||||
nil, // TODO: replace with *db.Queries or pgxpool handle
|
||||
h,
|
||||
nil, // TODO: replace with channel key store
|
||||
)
|
||||
|
||||
broker2 := ingest.New(
|
||||
ingest.Config{
|
||||
BrokerName: "mqtt2",
|
||||
URL: mustEnv("MQTT_BROKER_2_URL"),
|
||||
Username: mustEnv("MQTT_BROKER_2_USERNAME"),
|
||||
Password: mustEnv("MQTT_BROKER_2_PASSWORD"),
|
||||
},
|
||||
nil, // TODO: replace with *db.Queries or pgxpool handle
|
||||
h,
|
||||
nil, // TODO: replace with channel key store
|
||||
)
|
||||
|
||||
go broker1.Start(ctx)
|
||||
go broker2.Start(ctx)
|
||||
|
||||
// ── HTTP server ──────────────────────────────────────────────────────────
|
||||
r := router.New(h)
|
||||
|
||||
srv := &http.Server{
|
||||
Addr: addr,
|
||||
Handler: r,
|
||||
}
|
||||
|
||||
go func() {
|
||||
fmt.Printf("Tower listening on %s\n", addr)
|
||||
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||
log.Fatalf("server error: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// ── Graceful shutdown ────────────────────────────────────────────────────
|
||||
quit := make(chan os.Signal, 1)
|
||||
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
|
||||
<-quit
|
||||
|
||||
log.Println("shutting down...")
|
||||
cancel() // stops ingest workers
|
||||
if err := srv.Shutdown(context.Background()); err != nil {
|
||||
log.Printf("server shutdown error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// mustEnv returns the value of an env var, or empty string if unset.
|
||||
// Workers handle missing config gracefully (log + retry), so we don't fatal here.
|
||||
func mustEnv(key string) string {
|
||||
v := os.Getenv(key)
|
||||
if v == "" {
|
||||
log.Printf("warning: %s is not set", key)
|
||||
}
|
||||
return v
|
||||
}
|
||||
@@ -0,0 +1,322 @@
|
||||
-- ============================================================
|
||||
-- Tower schema migration
|
||||
-- ============================================================
|
||||
|
||||
-- ============================================================
|
||||
-- IATA CODES
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE iata_codes (
|
||||
iata CHAR(3) PRIMARY KEY,
|
||||
display_name TEXT,
|
||||
approx_lat DOUBLE PRECISION,
|
||||
approx_lng DOUBLE PRECISION,
|
||||
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
);
|
||||
|
||||
-- ============================================================
|
||||
-- REGIONS
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE regions (
|
||||
id SERIAL PRIMARY KEY,
|
||||
slug TEXT UNIQUE NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
description TEXT,
|
||||
display_order INT DEFAULT 0,
|
||||
center_lat DOUBLE PRECISION,
|
||||
center_lng DOUBLE PRECISION,
|
||||
zoom_level INT DEFAULT 8,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE TABLE region_iatas (
|
||||
region_id INT NOT NULL REFERENCES regions(id) ON DELETE CASCADE,
|
||||
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
|
||||
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
PRIMARY KEY (region_id, iata)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_region_iatas_iata ON region_iatas(iata);
|
||||
|
||||
-- ============================================================
|
||||
-- NODES (must come before observers due to observer_owners FK)
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE nodes (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
public_key BYTEA UNIQUE NOT NULL,
|
||||
node_type SMALLINT NOT NULL,
|
||||
name TEXT,
|
||||
latitude DOUBLE PRECISION,
|
||||
longitude DOUBLE PRECISION,
|
||||
location_source TEXT,
|
||||
last_advert_at TIMESTAMPTZ,
|
||||
supports_multibyte_paths BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
supports_multibyte_traces BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
min_firmware_version TEXT GENERATED ALWAYS AS (
|
||||
CASE
|
||||
WHEN supports_multibyte_paths THEN '1.14.0+'
|
||||
WHEN supports_multibyte_traces THEN '1.11.0+'
|
||||
ELSE NULL
|
||||
END
|
||||
) STORED,
|
||||
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
metadata JSONB
|
||||
);
|
||||
|
||||
CREATE INDEX idx_nodes_type_last_seen ON nodes(node_type, last_seen DESC);
|
||||
CREATE INDEX idx_nodes_location ON nodes(latitude, longitude)
|
||||
WHERE latitude IS NOT NULL AND longitude IS NOT NULL;
|
||||
CREATE INDEX idx_nodes_pubkey ON nodes(public_key);
|
||||
CREATE INDEX idx_nodes_multibyte_paths ON nodes(supports_multibyte_paths) WHERE supports_multibyte_paths;
|
||||
CREATE INDEX idx_nodes_multibyte_traces ON nodes(supports_multibyte_traces) WHERE supports_multibyte_traces;
|
||||
CREATE INDEX idx_nodes_min_firmware ON nodes(min_firmware_version) WHERE min_firmware_version IS NOT NULL;
|
||||
|
||||
-- ============================================================
|
||||
-- OBSERVERS
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE observers (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
public_key BYTEA UNIQUE NOT NULL,
|
||||
display_name TEXT,
|
||||
observer_type TEXT,
|
||||
software_version TEXT,
|
||||
hardware_model TEXT,
|
||||
firmware_version TEXT,
|
||||
firmware_build TEXT,
|
||||
radio_freq_mhz REAL,
|
||||
radio_sf SMALLINT,
|
||||
radio_bw_khz REAL,
|
||||
radio_cr SMALLINT,
|
||||
battery_level REAL,
|
||||
uptime_seconds BIGINT,
|
||||
status_metadata JSONB,
|
||||
last_status_at TIMESTAMPTZ,
|
||||
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
observation_count BIGINT DEFAULT 0,
|
||||
metadata JSONB
|
||||
);
|
||||
|
||||
CREATE INDEX idx_observers_last_seen ON observers(last_seen DESC);
|
||||
CREATE INDEX idx_observers_pubkey ON observers(public_key);
|
||||
CREATE INDEX idx_observers_type ON observers(observer_type) WHERE observer_type IS NOT NULL;
|
||||
|
||||
CREATE TABLE observer_brokers (
|
||||
observer_id UUID NOT NULL REFERENCES observers(id) ON DELETE CASCADE,
|
||||
broker_name TEXT NOT NULL,
|
||||
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
last_packet_at TIMESTAMPTZ,
|
||||
auth_ok BOOLEAN DEFAULT TRUE,
|
||||
PRIMARY KEY (observer_id, broker_name)
|
||||
);
|
||||
|
||||
CREATE TABLE observer_locations (
|
||||
observer_id UUID NOT NULL REFERENCES observers(id) ON DELETE CASCADE,
|
||||
iata CHAR(3) REFERENCES iata_codes(iata),
|
||||
latitude DOUBLE PRECISION,
|
||||
longitude DOUBLE PRECISION,
|
||||
reported_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
PRIMARY KEY (observer_id, reported_at)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_observer_locations_recent ON observer_locations(observer_id, reported_at DESC);
|
||||
|
||||
CREATE TABLE observer_telemetry (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
observer_id UUID NOT NULL REFERENCES observers(id) ON DELETE CASCADE,
|
||||
reported_at TIMESTAMPTZ NOT NULL,
|
||||
battery_voltage_mv INT,
|
||||
airtime_tx_pct REAL,
|
||||
airtime_rx_pct REAL,
|
||||
noise_floor_db REAL,
|
||||
uptime_seconds BIGINT,
|
||||
queue_length INT,
|
||||
debug_flags INT,
|
||||
receive_errors INT,
|
||||
UNIQUE (observer_id, reported_at)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_telemetry_reported_brin ON observer_telemetry USING BRIN (reported_at);
|
||||
CREATE INDEX idx_telemetry_observer_recent ON observer_telemetry(observer_id, reported_at DESC);
|
||||
|
||||
CREATE TABLE observer_owners (
|
||||
observer_id UUID PRIMARY KEY REFERENCES observers(id) ON DELETE CASCADE,
|
||||
owner_node_id UUID REFERENCES nodes(id),
|
||||
owner_pubkey BYTEA,
|
||||
contact_name TEXT,
|
||||
contact_email TEXT,
|
||||
notes TEXT,
|
||||
source TEXT,
|
||||
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX idx_observer_owners_node ON observer_owners(owner_node_id) WHERE owner_node_id IS NOT NULL;
|
||||
|
||||
-- ============================================================
|
||||
-- NODE IATAS AND SHORT IDS
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE node_iatas (
|
||||
node_id UUID NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
|
||||
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
|
||||
first_heard TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
last_heard TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
observation_count BIGINT DEFAULT 0,
|
||||
PRIMARY KEY (node_id, iata)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_node_iatas_iata ON node_iatas(iata, last_heard DESC);
|
||||
|
||||
CREATE TABLE node_short_ids (
|
||||
node_id UUID NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
|
||||
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE,
|
||||
prefix_4 BYTEA NOT NULL,
|
||||
prefix_1 BYTEA GENERATED ALWAYS AS (substring(prefix_4 from 1 for 1)) STORED,
|
||||
prefix_2 BYTEA GENERATED ALWAYS AS (substring(prefix_4 from 1 for 2)) STORED,
|
||||
prefix_3 BYTEA GENERATED ALWAYS AS (substring(prefix_4 from 1 for 3)) STORED,
|
||||
PRIMARY KEY (node_id, iata)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_short_ids_p1 ON node_short_ids(iata, prefix_1);
|
||||
CREATE INDEX idx_short_ids_p2 ON node_short_ids(iata, prefix_2);
|
||||
CREATE INDEX idx_short_ids_p3 ON node_short_ids(iata, prefix_3);
|
||||
CREATE INDEX idx_short_ids_p4 ON node_short_ids(iata, prefix_4);
|
||||
|
||||
-- ============================================================
|
||||
-- PACKETS
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE packets (
|
||||
packet_hash BYTEA PRIMARY KEY,
|
||||
payload_type SMALLINT NOT NULL,
|
||||
payload_version SMALLINT NOT NULL,
|
||||
route_type SMALLINT NOT NULL,
|
||||
transport_codes_present BOOLEAN DEFAULT FALSE,
|
||||
region_code INT,
|
||||
sub_region_code INT,
|
||||
origin_pubkey BYTEA,
|
||||
raw_payload BYTEA NOT NULL,
|
||||
parsed_payload JSONB,
|
||||
decrypted BOOLEAN DEFAULT FALSE,
|
||||
channel_hash BYTEA,
|
||||
first_heard_at TIMESTAMPTZ NOT NULL,
|
||||
last_heard_at TIMESTAMPTZ NOT NULL,
|
||||
observation_count INT DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE INDEX idx_packets_first_heard_brin ON packets USING BRIN (first_heard_at);
|
||||
CREATE INDEX idx_packets_payload_type ON packets(payload_type, first_heard_at DESC);
|
||||
CREATE INDEX idx_packets_route_type ON packets(route_type, first_heard_at DESC);
|
||||
CREATE INDEX idx_packets_origin ON packets(origin_pubkey, first_heard_at DESC)
|
||||
WHERE origin_pubkey IS NOT NULL;
|
||||
CREATE INDEX idx_packets_channel ON packets(channel_hash, first_heard_at DESC)
|
||||
WHERE channel_hash IS NOT NULL;
|
||||
|
||||
-- ============================================================
|
||||
-- PACKET OBSERVATIONS
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE packet_observations (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
packet_hash BYTEA NOT NULL REFERENCES packets(packet_hash) ON DELETE CASCADE,
|
||||
observer_id UUID NOT NULL REFERENCES observers(id),
|
||||
iata CHAR(3) NOT NULL REFERENCES iata_codes(iata),
|
||||
heard_at TIMESTAMPTZ NOT NULL,
|
||||
path_length_byte SMALLINT NOT NULL,
|
||||
hash_size SMALLINT NOT NULL,
|
||||
hop_count SMALLINT NOT NULL,
|
||||
path_bytes BYTEA,
|
||||
rssi SMALLINT,
|
||||
snr REAL,
|
||||
propagation_time_ms INT,
|
||||
radio_freq_mhz REAL,
|
||||
spread_factor SMALLINT,
|
||||
bandwidth_khz REAL,
|
||||
coding_rate SMALLINT,
|
||||
source_broker TEXT,
|
||||
UNIQUE (packet_hash, observer_id, heard_at)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_observations_heard_brin ON packet_observations USING BRIN (heard_at);
|
||||
CREATE INDEX idx_observations_iata_heard ON packet_observations(iata, heard_at DESC);
|
||||
CREATE INDEX idx_observations_observer ON packet_observations(observer_id, heard_at DESC);
|
||||
CREATE INDEX idx_observations_packet ON packet_observations(packet_hash);
|
||||
|
||||
-- ============================================================
|
||||
-- CHANNELS AND CHAT MESSAGES
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE channels (
|
||||
id SERIAL PRIMARY KEY,
|
||||
channel_hash BYTEA UNIQUE NOT NULL,
|
||||
name TEXT,
|
||||
is_hashtag BOOLEAN DEFAULT FALSE,
|
||||
is_public BOOLEAN DEFAULT FALSE,
|
||||
key_known BOOLEAN DEFAULT FALSE,
|
||||
first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
message_count BIGINT DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE INDEX idx_channels_last_seen ON channels(last_seen DESC);
|
||||
|
||||
CREATE TABLE channel_keys (
|
||||
channel_id INT PRIMARY KEY REFERENCES channels(id) ON DELETE CASCADE,
|
||||
key_bytes BYTEA NOT NULL,
|
||||
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
added_by TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE channel_messages (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
channel_id INT NOT NULL REFERENCES channels(id),
|
||||
packet_hash BYTEA NOT NULL REFERENCES packets(packet_hash),
|
||||
sender_name TEXT,
|
||||
sender_pubkey BYTEA,
|
||||
content TEXT,
|
||||
sent_at TIMESTAMPTZ NOT NULL,
|
||||
UNIQUE (packet_hash)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_channel_messages_channel ON channel_messages(channel_id, sent_at DESC);
|
||||
CREATE INDEX idx_channel_messages_sent_brin ON channel_messages USING BRIN (sent_at);
|
||||
|
||||
-- ============================================================
|
||||
-- MATERIALIZED VIEWS
|
||||
-- ============================================================
|
||||
|
||||
CREATE MATERIALIZED VIEW mv_hourly_iata_stats AS
|
||||
SELECT
|
||||
iata,
|
||||
date_trunc('hour', heard_at) AS hour,
|
||||
COUNT(*) AS observation_count,
|
||||
COUNT(DISTINCT packet_hash) AS unique_packets,
|
||||
COUNT(DISTINCT observer_id) AS active_observers
|
||||
FROM packet_observations
|
||||
WHERE heard_at > NOW() - INTERVAL '7 days'
|
||||
GROUP BY iata, date_trunc('hour', heard_at);
|
||||
|
||||
CREATE UNIQUE INDEX idx_mv_hourly_iata
|
||||
ON mv_hourly_iata_stats(iata, hour);
|
||||
|
||||
CREATE MATERIALIZED VIEW mv_top_nodes_by_iata AS
|
||||
SELECT
|
||||
ni.iata,
|
||||
ni.node_id,
|
||||
n.name,
|
||||
n.node_type,
|
||||
ni.observation_count,
|
||||
ni.last_heard
|
||||
FROM node_iatas ni
|
||||
JOIN nodes n ON n.id = ni.node_id
|
||||
WHERE ni.last_heard > NOW() - INTERVAL '7 days';
|
||||
|
||||
CREATE UNIQUE INDEX idx_mv_top_nodes
|
||||
ON mv_top_nodes_by_iata(iata, node_id);
|
||||
@@ -0,0 +1,266 @@
|
||||
-- ============================================================
|
||||
-- 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;
|
||||
|
||||
-- ============================================================
|
||||
-- OBSERVERS
|
||||
-- ============================================================
|
||||
|
||||
-- name: UpsertObserver :one
|
||||
INSERT INTO observers (public_key, last_seen)
|
||||
VALUES ($1, NOW())
|
||||
ON CONFLICT (public_key) DO UPDATE SET
|
||||
last_seen = NOW(),
|
||||
observation_count = observers.observation_count + 1
|
||||
RETURNING *;
|
||||
|
||||
-- name: UpdateObserverStatus :exec
|
||||
UPDATE observers SET
|
||||
display_name = COALESCE($2, display_name),
|
||||
observer_type = COALESCE($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;
|
||||
|
||||
-- name: GetObserverByPubkey :one
|
||||
SELECT * FROM observers WHERE public_key = $1;
|
||||
|
||||
-- name: ListObservers :many
|
||||
SELECT * FROM observers ORDER BY last_seen DESC;
|
||||
|
||||
-- ============================================================
|
||||
-- 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,
|
||||
parsed_payload,
|
||||
channel_hash,
|
||||
first_heard_at,
|
||||
last_heard_at,
|
||||
observation_count
|
||||
) VALUES (
|
||||
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, NOW(), NOW(), 1
|
||||
)
|
||||
ON CONFLICT (packet_hash) DO UPDATE SET
|
||||
last_heard_at = NOW(),
|
||||
observation_count = packets.observation_count + 1
|
||||
RETURNING *, (xmax = 0) AS inserted;
|
||||
|
||||
-- name: GetPacket :one
|
||||
SELECT * FROM packets WHERE packet_hash = $1;
|
||||
|
||||
-- name: ListPackets :many
|
||||
SELECT p.*
|
||||
FROM packets p
|
||||
WHERE
|
||||
($1::smallint IS NULL OR p.payload_type = $1)
|
||||
AND ($2::smallint IS NULL OR p.route_type = $2)
|
||||
AND ($3::timestamptz IS NULL OR p.first_heard_at >= $3)
|
||||
AND ($4::timestamptz IS NULL OR p.first_heard_at <= $4)
|
||||
ORDER BY p.last_heard_at DESC
|
||||
LIMIT $5;
|
||||
|
||||
-- name: ListPacketsAfterID :many
|
||||
SELECT p.*
|
||||
FROM packets p
|
||||
JOIN packet_observations po ON po.packet_hash = p.packet_hash
|
||||
WHERE po.id > $1
|
||||
ORDER BY po.id ASC
|
||||
LIMIT $2;
|
||||
|
||||
-- ============================================================
|
||||
-- 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 * FROM packet_observations
|
||||
WHERE packet_hash = $1
|
||||
ORDER BY heard_at ASC;
|
||||
|
||||
-- name: ListObservationsForObserver :many
|
||||
SELECT * FROM packet_observations
|
||||
WHERE observer_id = $1
|
||||
AND ($2::timestamptz IS NULL OR heard_at >= $2)
|
||||
ORDER BY heard_at DESC
|
||||
LIMIT $3;
|
||||
|
||||
-- ============================================================
|
||||
-- NODES
|
||||
-- ============================================================
|
||||
|
||||
-- name: UpsertNode :one
|
||||
INSERT INTO nodes (public_key, node_type, name, latitude, longitude, location_source, last_advert_at, last_seen)
|
||||
VALUES ($1, $2, $3, $4, $5, 'advert', NOW(), NOW())
|
||||
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()
|
||||
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: GetNodeByPubkey :one
|
||||
SELECT * FROM nodes WHERE public_key = $1;
|
||||
|
||||
-- name: ListNodes :many
|
||||
SELECT * FROM nodes
|
||||
WHERE
|
||||
($1::smallint IS NULL OR node_type = $1)
|
||||
ORDER BY last_seen DESC
|
||||
LIMIT $2;
|
||||
|
||||
-- ============================================================
|
||||
-- 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
|
||||
INSERT INTO channels (channel_hash, last_seen)
|
||||
VALUES ($1, NOW())
|
||||
ON CONFLICT (channel_hash) DO UPDATE SET
|
||||
last_seen = NOW(),
|
||||
message_count = CASE WHEN $2 THEN channels.message_count + 1 ELSE channels.message_count END
|
||||
RETURNING *;
|
||||
|
||||
-- name: SetChannelKeyKnown :exec
|
||||
UPDATE channels SET key_known = TRUE WHERE channel_hash = $1;
|
||||
|
||||
-- name: ListChannels :many
|
||||
SELECT * FROM channels ORDER BY last_seen DESC LIMIT $1;
|
||||
|
||||
-- name: GetChannel :one
|
||||
SELECT * FROM channels WHERE channel_hash = $1;
|
||||
|
||||
-- ============================================================
|
||||
-- CHANNEL MESSAGES
|
||||
-- ============================================================
|
||||
|
||||
-- name: InsertChannelMessage :exec
|
||||
INSERT INTO channel_messages (channel_id, packet_hash, sender_name, sender_pubkey, content, sent_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)
|
||||
ON CONFLICT (packet_hash) DO NOTHING;
|
||||
|
||||
-- name: ListChannelMessages :many
|
||||
SELECT * FROM channel_messages
|
||||
WHERE channel_id = $1
|
||||
AND ($2::timestamptz IS NULL OR sent_at >= $2)
|
||||
ORDER BY sent_at DESC
|
||||
LIMIT $3;
|
||||
|
||||
-- ============================================================
|
||||
-- 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::char(3) IS NULL OR po.iata = $1);
|
||||
|
||||
-- name: GetHourlyStats :many
|
||||
SELECT * FROM mv_hourly_iata_stats
|
||||
WHERE ($1::char(3) IS NULL OR iata = $1)
|
||||
AND hour >= NOW() - $2::interval
|
||||
ORDER BY iata, hour;
|
||||
|
||||
-- name: GetTopNodes :many
|
||||
SELECT * FROM mv_top_nodes_by_iata
|
||||
WHERE ($1::char(3) IS NULL OR iata = $1)
|
||||
ORDER BY observation_count DESC
|
||||
LIMIT $2;
|
||||
@@ -0,0 +1,62 @@
|
||||
services:
|
||||
app:
|
||||
build: .
|
||||
env_file: .env
|
||||
ports:
|
||||
- "8080:8080"
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
redis:
|
||||
condition: service_healthy
|
||||
restart: unless-stopped
|
||||
|
||||
postgres:
|
||||
image: postgres:16-alpine
|
||||
environment:
|
||||
POSTGRES_USER: tower
|
||||
POSTGRES_PASSWORD: tower
|
||||
POSTGRES_DB: tower
|
||||
volumes:
|
||||
- postgres_data:/var/lib/postgresql/data
|
||||
- ./db/migrations:/docker-entrypoint-initdb.d:ro
|
||||
ports:
|
||||
- "5432:5432"
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "pg_isready -U tower"]
|
||||
interval: 5s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
restart: unless-stopped
|
||||
|
||||
redis:
|
||||
image: redis:7-alpine
|
||||
volumes:
|
||||
- redis_data:/data
|
||||
ports:
|
||||
- "6379:6379"
|
||||
healthcheck:
|
||||
test: ["CMD", "redis-cli", "ping"]
|
||||
interval: 5s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
restart: unless-stopped
|
||||
|
||||
caddy:
|
||||
image: caddy:2-alpine
|
||||
ports:
|
||||
- "80:80"
|
||||
- "443:443"
|
||||
volumes:
|
||||
- ./Caddyfile:/etc/caddy/Caddyfile:ro
|
||||
- caddy_data:/data
|
||||
- caddy_config:/config
|
||||
depends_on:
|
||||
- app
|
||||
restart: unless-stopped
|
||||
|
||||
volumes:
|
||||
postgres_data:
|
||||
redis_data:
|
||||
caddy_data:
|
||||
caddy_config:
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
LISTEN_ADDR=:8080
|
||||
|
||||
POSTGRES_DSN=postgres://tower:tower@localhost:5432/tower?sslmode=disable
|
||||
|
||||
REDIS_ADDR=localhost:6379
|
||||
|
||||
MQTT_BROKER_1_URL=wss://mqtt1.meshcore.ca:443
|
||||
MQTT_BROKER_1_USERNAME=
|
||||
MQTT_BROKER_1_PASSWORD=
|
||||
|
||||
MQTT_BROKER_2_URL=wss://mqtt2.meshcore.ca:443
|
||||
MQTT_BROKER_2_USERNAME=
|
||||
MQTT_BROKER_2_PASSWORD=
|
||||
|
||||
PACKET_RETENTION_DAYS=30
|
||||
@@ -0,0 +1,19 @@
|
||||
module tower
|
||||
|
||||
go 1.26.1
|
||||
|
||||
require (
|
||||
github.com/eclipse/paho.mqtt.golang v1.5.1
|
||||
github.com/go-chi/chi/v5 v5.2.1
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/meshcore-go/meshcore-go v1.0.6
|
||||
)
|
||||
|
||||
require (
|
||||
filippo.io/edwards25519 v1.2.0 // indirect
|
||||
github.com/gorilla/websocket v1.5.3 // indirect
|
||||
github.com/joho/godotenv v1.5.1 // indirect
|
||||
golang.org/x/crypto v0.51.0 // indirect
|
||||
golang.org/x/net v0.53.0 // indirect
|
||||
golang.org/x/sync v0.17.0 // indirect
|
||||
)
|
||||
@@ -0,0 +1,20 @@
|
||||
filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
|
||||
filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
|
||||
github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
|
||||
github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
|
||||
github.com/go-chi/chi/v5 v5.2.1 h1:KOIHODQj58PmL80G2Eak4WdvUzjSJSm0vG72crDCqb8=
|
||||
github.com/go-chi/chi/v5 v5.2.1/go.mod h1:L2yAIGWB3H+phAw1NxKwWM+7eUH/lU8pOMm5hHcoops=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
|
||||
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
|
||||
github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
|
||||
github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
|
||||
github.com/meshcore-go/meshcore-go v1.0.6 h1:Vf/DC2vdr76lW/kF4TntjiUgBObAip8fdrwV10R6o0M=
|
||||
github.com/meshcore-go/meshcore-go v1.0.6/go.mod h1:u+Lvlg4Wy4blqCAJB+yGL3j2y7JRow+RQfQPjiI294Y=
|
||||
golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI=
|
||||
golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8=
|
||||
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
|
||||
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
|
||||
golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
|
||||
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
|
||||
@@ -0,0 +1,59 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// ChannelsRouter mounts all /channels routes onto a subrouter.
|
||||
//
|
||||
// GET /channels → ListChannels
|
||||
// GET /channels/{channelHash} → GetChannel
|
||||
// GET /channels/{channelHash}/messages → ListChannelMessages
|
||||
func ChannelsRouter() http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Get("/", ListChannels)
|
||||
|
||||
r.Route("/{channelHash}", func(r chi.Router) {
|
||||
r.Get("/", GetChannel)
|
||||
r.Get("/messages", ListChannelMessages)
|
||||
})
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// ListChannels handles GET /api/v1/channels
|
||||
//
|
||||
// Query params (all optional):
|
||||
// limit=50
|
||||
func ListChannels(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query channels ORDER BY last_seen DESC, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetChannel handles GET /api/v1/channels/{channelHash}
|
||||
//
|
||||
// Returns channel detail including key_known status and message count.
|
||||
// Channel keys are server-side config; key material is never exposed via the API.
|
||||
func GetChannel(w http.ResponseWriter, r *http.Request) {
|
||||
// channelHash := chi.URLParam(r, "channelHash")
|
||||
// TODO: fetch channel, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// ListChannelMessages handles GET /api/v1/channels/{channelHash}/messages
|
||||
//
|
||||
// Query params (all optional):
|
||||
// since=<epoch ms>
|
||||
// limit=50
|
||||
// cursor=<opaque>
|
||||
//
|
||||
// Returns paginated decrypted channel messages. Messages where key_known=false
|
||||
// will have content=null.
|
||||
func ListChannelMessages(w http.ResponseWriter, r *http.Request) {
|
||||
// channelHash := chi.URLParam(r, "channelHash")
|
||||
// TODO: fetch channel_messages, paginate, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// IATAsRouter mounts all /iatas routes onto a subrouter.
|
||||
//
|
||||
// GET /iatas → ListIATAs
|
||||
// GET /iatas/{iata} → GetIATA
|
||||
func IATAsRouter() http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Get("/", ListIATAs)
|
||||
r.Get("/{iata}", GetIATA)
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// ListIATAs handles GET /api/v1/iatas
|
||||
//
|
||||
// Returns all known IATA codes with display name and coordinates where set.
|
||||
// IATAs are auto-created on first packet arrival; config file overrides name/coords.
|
||||
func ListIATAs(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query iata_codes, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetIATA handles GET /api/v1/iatas/{iata}
|
||||
//
|
||||
// Returns detail for a single IATA code including associated region memberships
|
||||
// and basic recent stats.
|
||||
func GetIATA(w http.ResponseWriter, r *http.Request) {
|
||||
// iata := chi.URLParam(r, "iata")
|
||||
// TODO: fetch iata_codes row + region memberships, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// NodesRouter mounts all /nodes routes onto a subrouter.
|
||||
//
|
||||
// GET /nodes → ListNodes
|
||||
// GET /nodes/{nodeId} → GetNode
|
||||
// GET /nodes/{nodeId}/observations → ListNodeObservations
|
||||
func NodesRouter() http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Get("/", ListNodes)
|
||||
|
||||
r.Route("/{nodeId}", func(r chi.Router) {
|
||||
r.Get("/", GetNode)
|
||||
r.Get("/observations", ListNodeObservations)
|
||||
})
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// ListNodes handles GET /api/v1/nodes
|
||||
//
|
||||
// Query params (all optional):
|
||||
// type=2 (node_type: 1=companion, 2=repeater, 3=room server)
|
||||
// iata=YOW
|
||||
// firmwareTier=1.14.0
|
||||
// limit=50
|
||||
// cursor=<opaque>
|
||||
func ListNodes(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query nodes with optional filters, paginate, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetNode handles GET /api/v1/nodes/{nodeId}
|
||||
//
|
||||
// Returns full node detail including iatasHeardIn, firmware capability flags,
|
||||
// minFirmwareVersion, and the latest advert payload.
|
||||
func GetNode(w http.ResponseWriter, r *http.Request) {
|
||||
// nodeId := chi.URLParam(r, "nodeId")
|
||||
// TODO: fetch node, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// ListNodeObservations handles GET /api/v1/nodes/{nodeId}/observations
|
||||
//
|
||||
// Query params (all optional):
|
||||
// since=<epoch ms>
|
||||
// limit=50
|
||||
// cursor=<opaque>
|
||||
func ListNodeObservations(w http.ResponseWriter, r *http.Request) {
|
||||
// nodeId := chi.URLParam(r, "nodeId")
|
||||
// TODO: fetch observations for node, paginate, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// ObserversRouter mounts all /observers routes onto a subrouter.
|
||||
//
|
||||
// GET /observers → ListObservers
|
||||
// GET /observers/{observerId} → GetObserver
|
||||
// GET /observers/{observerId}/telemetry → GetObserverTelemetry
|
||||
// GET /observers/{observerId}/adverts → ListObserverAdverts
|
||||
func ObserversRouter() http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Get("/", ListObservers)
|
||||
|
||||
r.Route("/{observerId}", func(r chi.Router) {
|
||||
r.Get("/", GetObserver)
|
||||
r.Get("/telemetry", GetObserverTelemetry)
|
||||
r.Get("/adverts", ListObserverAdverts)
|
||||
})
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// ListObservers handles GET /api/v1/observers
|
||||
//
|
||||
// Query params (all optional):
|
||||
//
|
||||
// iata=YOW
|
||||
// type=meshcoretomqtt
|
||||
// broker=mqtt1
|
||||
// status=online
|
||||
func ListObservers(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query observers with optional filters, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetObserver handles GET /api/v1/observers/{observerId}
|
||||
//
|
||||
// Returns full observer detail including broker badges, type, and recent stats.
|
||||
// Note: observer_owners data is never exposed via the public API.
|
||||
func GetObserver(w http.ResponseWriter, r *http.Request) {
|
||||
// observerId := chi.URLParam(r, "observerId")
|
||||
// TODO: fetch observer (exclude observer_owners fields), write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetObserverTelemetry handles GET /api/v1/observers/{observerId}/telemetry
|
||||
//
|
||||
// Query params (all optional):
|
||||
//
|
||||
// range=24h duration string: 24h, 7d, 30d
|
||||
// afterId=<status id> for deterministic WS reconnection backfill
|
||||
// limit=100
|
||||
//
|
||||
// Returns a time-bucketed array of telemetry points suitable for charting
|
||||
// (battery, airtime, noise floor, uptime, queue depth, receive errors).
|
||||
func GetObserverTelemetry(w http.ResponseWriter, r *http.Request) {
|
||||
// observerId := chi.URLParam(r, "observerId")
|
||||
// TODO: query status_metadata history, bucket by interval, write JSON response.
|
||||
// afterId (int64): WHERE id > afterId ORDER BY id ASC LIMIT limit
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// ListObserverAdverts handles GET /api/v1/observers/{observerId}/adverts
|
||||
//
|
||||
// Query params (all optional):
|
||||
//
|
||||
// limit=50
|
||||
// cursor=<opaque>
|
||||
func ListObserverAdverts(w http.ResponseWriter, r *http.Request) {
|
||||
// observerId := chi.URLParam(r, "observerId")
|
||||
// TODO: fetch advert packets heard by this observer, paginate, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// PacketsRouter mounts all /packets routes onto a subrouter.
|
||||
//
|
||||
// GET /packets → ListPackets
|
||||
// GET /packets/{packetHash} → GetPacket
|
||||
func PacketsRouter() http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Get("/", ListPackets)
|
||||
r.Get("/{packetHash}", GetPacket)
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// ListPackets handles GET /api/v1/packets
|
||||
//
|
||||
// Query params (all optional):
|
||||
//
|
||||
// iata=YOW
|
||||
// payloadType=4
|
||||
// routeType=1
|
||||
// since=<epoch ms>
|
||||
// until=<epoch ms>
|
||||
// afterId=<observation id> used for deterministic WS reconnection backfill
|
||||
// limit=50
|
||||
// cursor=<opaque>
|
||||
//
|
||||
// Returns a paginated list of packet summaries with the latest observation
|
||||
// rolled in, newest first.
|
||||
func ListPackets(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: parse query params, query DB/cache, write JSON response.
|
||||
//
|
||||
// afterId (int64) takes precedence over cursor for reconnection backfill:
|
||||
// WHERE id > afterId ORDER BY id ASC LIMIT limit
|
||||
// Normal pagination uses cursor (opaque, encodes last seen id+timestamp).
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetPacket handles GET /api/v1/packets/{packetHash}
|
||||
//
|
||||
// Returns the full packet with all observations and each observation's
|
||||
// resolved path inline.
|
||||
func GetPacket(w http.ResponseWriter, r *http.Request) {
|
||||
// packetHash := chi.URLParam(r, "packetHash")
|
||||
// TODO: fetch packet + observations, resolve paths, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// RegionsRouter mounts all /regions routes onto a subrouter.
|
||||
//
|
||||
// GET /regions → ListRegions
|
||||
// GET /regions/{regionId} → GetRegion
|
||||
//
|
||||
// Note: region creation and IATA assignment are managed via the server config
|
||||
// file, not the API (v1). These endpoints are read-only.
|
||||
func RegionsRouter() http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Get("/", ListRegions)
|
||||
r.Get("/{regionId}", GetRegion)
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// ListRegions handles GET /api/v1/regions
|
||||
//
|
||||
// Returns all super-regions with their associated IATA codes, center
|
||||
// coordinates, and zoom level for map initialisation.
|
||||
func ListRegions(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query regions + region_iatas, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetRegion handles GET /api/v1/regions/{regionId}
|
||||
//
|
||||
// Returns detail for a single super-region including its full IATA membership
|
||||
// list and recent aggregate stats.
|
||||
func GetRegion(w http.ResponseWriter, r *http.Request) {
|
||||
// regionId := chi.URLParam(r, "regionId")
|
||||
// TODO: fetch region + iatas, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// StatsRouter mounts all /stats routes onto a subrouter.
|
||||
//
|
||||
// GET /stats/overview → GetStatsOverview
|
||||
// GET /stats/observations → GetStatsObservations
|
||||
// GET /stats/payloadBreakdown → GetStatsPayloadBreakdown
|
||||
// GET /stats/topNodes → GetStatsTopNodes
|
||||
// GET /stats/topObservers → GetStatsTopObservers
|
||||
//
|
||||
// All endpoints accept either iata= (one or comma-separated) or regionId=
|
||||
// (expands to all IATAs in that super-region via region_iatas).
|
||||
func StatsRouter() http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Get("/overview", GetStatsOverview)
|
||||
r.Get("/observations", GetStatsObservations)
|
||||
r.Get("/payloadBreakdown", GetStatsPayloadBreakdown)
|
||||
r.Get("/topNodes", GetStatsTopNodes)
|
||||
r.Get("/topObservers", GetStatsTopObservers)
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// GetStatsOverview handles GET /api/v1/stats/overview
|
||||
//
|
||||
// Query params (all optional):
|
||||
// iata=YOW (one or comma-separated)
|
||||
// regionId=<id>
|
||||
//
|
||||
// Returns top-line figures: total packets and observations last 24h,
|
||||
// active observers, active IATAs, unique nodes seen.
|
||||
func GetStatsOverview(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query mv_hourly_iata_stats + live aggregates, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetStatsObservations handles GET /api/v1/stats/observations
|
||||
//
|
||||
// Query params (all optional):
|
||||
// iata=YOW
|
||||
// regionId=<id>
|
||||
// range=24h (duration string: 24h, 7d, 30d)
|
||||
// interval=1h (bucket size: 5m, 1h, 1d)
|
||||
//
|
||||
// Returns a time series of observation counts bucketed by interval,
|
||||
// suitable for charting.
|
||||
func GetStatsObservations(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query mv_hourly_iata_stats, group by bucket, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetStatsPayloadBreakdown handles GET /api/v1/stats/payloadBreakdown
|
||||
//
|
||||
// Query params (all optional):
|
||||
// iata=YOW
|
||||
// regionId=<id>
|
||||
// range=24h
|
||||
//
|
||||
// Returns observation counts grouped by payload_type for the given window.
|
||||
func GetStatsPayloadBreakdown(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query packets + observations filtered by time/iata, group by payload_type.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetStatsTopNodes handles GET /api/v1/stats/topNodes
|
||||
//
|
||||
// Query params (all optional):
|
||||
// iata=YOW
|
||||
// regionId=<id>
|
||||
// range=24h
|
||||
// limit=10
|
||||
//
|
||||
// Returns the top N nodes by observation contribution count using
|
||||
// mv_top_nodes_by_iata.
|
||||
func GetStatsTopNodes(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query mv_top_nodes_by_iata, apply filters, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetStatsTopObservers handles GET /api/v1/stats/topObservers
|
||||
//
|
||||
// Query params (all optional):
|
||||
// iata=YOW
|
||||
// regionId=<id>
|
||||
// range=24h
|
||||
// limit=10
|
||||
//
|
||||
// Returns the top N observers by observation count for the given window.
|
||||
func GetStatsTopObservers(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: query packet_observations grouped by observer_id, apply filters.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package middleware
|
||||
|
||||
import "net/http"
|
||||
|
||||
// NoopAuth is a placeholder for the authentication middleware that will be
|
||||
// wired onto the private route group when auth is implemented (see Future
|
||||
// Features → Admin authentication in the design doc).
|
||||
//
|
||||
// Replace this with a real JWT/session validation middleware before shipping
|
||||
// any write endpoints or admin functionality.
|
||||
func NoopAuth(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: validate bearer token, set user in context, return 401 on failure.
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package router
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/go-chi/chi/v5/middleware"
|
||||
|
||||
"tower/internal/api/handlers"
|
||||
mw "tower/internal/api/middleware"
|
||||
"tower/internal/hub"
|
||||
"tower/internal/ws"
|
||||
)
|
||||
|
||||
// New builds and returns the top-level Chi router.
|
||||
//
|
||||
// Route shape:
|
||||
//
|
||||
// /ws → WebSocket (public in v1)
|
||||
// /api/v1/ → public group
|
||||
// /packets → packets subrouter
|
||||
// /nodes → nodes subrouter
|
||||
// /observers → observers subrouter
|
||||
// /channels → channels subrouter
|
||||
// /iatas → iatas subrouter
|
||||
// /regions → regions subrouter
|
||||
// /stats → stats subrouter
|
||||
//
|
||||
// The private group is stubbed and ready for the auth middleware drop-in
|
||||
// described in Future Features → Admin authentication.
|
||||
func New(h *hub.Hub) http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
// ── Global middleware ────────────────────────────────────────────────────
|
||||
r.Use(middleware.RequestID)
|
||||
r.Use(middleware.RealIP)
|
||||
r.Use(middleware.Logger)
|
||||
r.Use(middleware.Recoverer)
|
||||
r.Use(middleware.CleanPath)
|
||||
r.Use(middleware.StripSlashes)
|
||||
|
||||
// ── WebSocket ────────────────────────────────────────────────────────────
|
||||
r.Get("/ws", ws.Handler(h))
|
||||
|
||||
// ── Public REST API (v1) ─────────────────────────────────────────────────
|
||||
r.Route("/api/v1", func(r chi.Router) {
|
||||
// Public group — no authentication required (all of v1 is public).
|
||||
r.Group(func(r chi.Router) {
|
||||
r.Mount("/packets", handlers.PacketsRouter())
|
||||
r.Mount("/nodes", handlers.NodesRouter())
|
||||
r.Mount("/observers", handlers.ObserversRouter())
|
||||
r.Mount("/channels", handlers.ChannelsRouter())
|
||||
r.Mount("/iatas", handlers.IATAsRouter())
|
||||
r.Mount("/regions", handlers.RegionsRouter())
|
||||
r.Mount("/stats", handlers.StatsRouter())
|
||||
})
|
||||
|
||||
// Private group — auth middleware applied.
|
||||
// Stubbed for the admin endpoints described in Future Features.
|
||||
// Swap mw.NoopAuth for a real JWT/session middleware when ready.
|
||||
r.Group(func(r chi.Router) {
|
||||
r.Use(mw.NoopAuth)
|
||||
// r.Mount("/admin", handlers.AdminRouter())
|
||||
})
|
||||
})
|
||||
|
||||
return r
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
// Package hub provides the central fan-out broker between the MQTT ingest
|
||||
// goroutines and connected WebSocket clients.
|
||||
//
|
||||
// Design:
|
||||
// - A single Hub.Run() goroutine owns the client map; no mutexes needed.
|
||||
// - Ingest goroutines call hub.Broadcast() from any goroutine.
|
||||
// - Each WebSocket connection gets a *Client with a buffered send channel.
|
||||
// - If a client's send buffer is full it receives a lagged notification and
|
||||
// its buffer is drained so it doesn't stall the broadcast loop.
|
||||
//
|
||||
// Subscription filtering (by IATA, payload type, channel hash, etc.) is
|
||||
// enforced here before events are placed on a client's send channel.
|
||||
package hub
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
)
|
||||
|
||||
// EventType identifies the kind of server-push event. These match the
|
||||
// discriminator values in the WebSocket protocol ("packetObservation", etc.).
|
||||
type EventType string
|
||||
|
||||
const (
|
||||
EventPacketObservation EventType = "packetObservation"
|
||||
EventObserverStatus EventType = "observerStatus"
|
||||
EventNodeUpdate EventType = "nodeUpdate"
|
||||
EventChannelMessage EventType = "channelMessage"
|
||||
)
|
||||
|
||||
// Event is a single fan-out unit. Payload is pre-serialised JSON so the
|
||||
// broadcast loop never touches encoding — it's done once by the ingest path.
|
||||
type Event struct {
|
||||
Type EventType
|
||||
Payload json.RawMessage
|
||||
|
||||
// Routing metadata used by the hub to match subscriptions.
|
||||
// Populated by the ingest layer before calling Broadcast.
|
||||
IATA string
|
||||
PayloadType uint8
|
||||
ChannelHash string // hex string, non-empty only for channelMessage events
|
||||
}
|
||||
|
||||
// Scope mirrors the client-side subscribe message. All fields are optional:
|
||||
// nil/empty means "no filter on this dimension" (match everything).
|
||||
// An empty non-nil slice means "match nothing on this dimension".
|
||||
type Scope struct {
|
||||
IATAs []string
|
||||
RegionIATAs []string // pre-expanded from regionId by the WS handler
|
||||
PayloadTypes []uint8
|
||||
ChannelHashes []string
|
||||
Events []EventType
|
||||
}
|
||||
|
||||
// Client represents a connected WebSocket consumer.
|
||||
type Client struct {
|
||||
Send chan Event
|
||||
scope []Scope // OR semantics: event matches if it matches any scope entry
|
||||
}
|
||||
|
||||
// matches returns true if the event satisfies at least one of the client's
|
||||
// active subscriptions.
|
||||
func (c *Client) matches(e Event) bool {
|
||||
if len(c.scope) == 0 {
|
||||
return true // no subscriptions yet → receive nothing
|
||||
}
|
||||
for _, s := range c.scope {
|
||||
if scopeMatches(s, e) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func scopeMatches(s Scope, e Event) bool {
|
||||
if len(s.Events) > 0 && !containsEventType(s.Events, e.Type) {
|
||||
return false
|
||||
}
|
||||
if len(s.IATAs) > 0 || len(s.RegionIATAs) > 0 {
|
||||
allIATAs := append(s.IATAs, s.RegionIATAs...) //nolint:gocritic
|
||||
if !containsString(allIATAs, e.IATA) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
if len(s.PayloadTypes) > 0 && !containsUint8(s.PayloadTypes, e.PayloadType) {
|
||||
return false
|
||||
}
|
||||
if len(s.ChannelHashes) > 0 && !containsString(s.ChannelHashes, e.ChannelHash) {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Hub is the central event broker.
|
||||
type Hub struct {
|
||||
subscribe chan subscribeMsg
|
||||
unsubscribe chan unsubscribeMsg
|
||||
remove chan *Client
|
||||
broadcast chan Event
|
||||
}
|
||||
|
||||
type subscribeMsg struct {
|
||||
client *Client
|
||||
scope Scope
|
||||
hasScope bool // true when this is an AddScope call, false for NewClient registration
|
||||
}
|
||||
|
||||
type unsubscribeMsg struct {
|
||||
client *Client
|
||||
subscriptionID string
|
||||
}
|
||||
|
||||
// New creates a Hub. Call Run() in a goroutine before using it.
|
||||
func New() *Hub {
|
||||
return &Hub{
|
||||
subscribe: make(chan subscribeMsg, 64),
|
||||
unsubscribe: make(chan unsubscribeMsg, 64),
|
||||
remove: make(chan *Client, 64),
|
||||
broadcast: make(chan Event, 512),
|
||||
}
|
||||
}
|
||||
|
||||
// NewClient creates a Client and registers it with the hub.
|
||||
// The caller is responsible for calling Remove when the connection closes.
|
||||
func (h *Hub) NewClient() *Client {
|
||||
c := &Client{
|
||||
Send: make(chan Event, 256),
|
||||
}
|
||||
// We don't add it to the map here; we send it through the channel so
|
||||
// Run() is the only goroutine that touches the client map.
|
||||
h.subscribe <- subscribeMsg{client: c, hasScope: false}
|
||||
return c
|
||||
}
|
||||
|
||||
// AddScope appends a subscription scope to a client. Called by the WS handler
|
||||
// when it receives a "subscribe" message from the client.
|
||||
func (h *Hub) AddScope(c *Client, s Scope) {
|
||||
h.subscribe <- subscribeMsg{client: c, scope: s, hasScope: true}
|
||||
}
|
||||
|
||||
// Remove deregisters a client and closes its Send channel.
|
||||
// Safe to call from any goroutine (e.g. the WS handler's defer).
|
||||
func (h *Hub) Remove(c *Client) {
|
||||
h.remove <- c
|
||||
}
|
||||
|
||||
// Broadcast enqueues an event for fan-out. Safe to call from any goroutine.
|
||||
func (h *Hub) Broadcast(e Event) {
|
||||
select {
|
||||
case h.broadcast <- e:
|
||||
default:
|
||||
log.Println("hub: broadcast channel full, dropping event")
|
||||
}
|
||||
}
|
||||
|
||||
// Run is the hub's single-goroutine event loop. Call it in a dedicated
|
||||
// goroutine: go hub.Run().
|
||||
//
|
||||
// It processes registrations, removals, and broadcasts sequentially so the
|
||||
// clients map needs no locking.
|
||||
func (h *Hub) Run() {
|
||||
// clients maps a *Client to the set of subscription IDs it holds.
|
||||
// We use a map[*Client]struct{} for O(1) presence checks and O(n)
|
||||
// broadcast — fine at the scale Tower targets.
|
||||
clients := make(map[*Client]struct{})
|
||||
|
||||
for {
|
||||
select {
|
||||
|
||||
case msg := <-h.subscribe:
|
||||
if !msg.hasScope {
|
||||
// Registration with no scope yet (NewClient path).
|
||||
clients[msg.client] = struct{}{}
|
||||
} else {
|
||||
// AddScope path — client must already be registered.
|
||||
if _, ok := clients[msg.client]; ok {
|
||||
msg.client.scope = append(msg.client.scope, msg.scope)
|
||||
}
|
||||
}
|
||||
|
||||
case c := <-h.remove:
|
||||
if _, ok := clients[c]; ok {
|
||||
delete(clients, c)
|
||||
close(c.Send)
|
||||
}
|
||||
|
||||
case evt := <-h.broadcast:
|
||||
for c := range clients {
|
||||
if !c.matches(evt) {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case c.Send <- evt:
|
||||
default:
|
||||
// Client send buffer full. The WS write pump is responsible
|
||||
// for detecting its own lagged state and sending the lagged
|
||||
// message. We just drain one slot so the broadcast loop
|
||||
// doesn't block.
|
||||
select {
|
||||
case <-c.Send:
|
||||
default:
|
||||
}
|
||||
log.Printf("hub: client send buffer full, dropped event type=%s", evt.Type)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
func containsString(haystack []string, needle string) bool {
|
||||
for _, s := range haystack {
|
||||
if s == needle {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func containsUint8(haystack []uint8, needle uint8) bool {
|
||||
for _, v := range haystack {
|
||||
if v == needle {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func containsEventType(haystack []EventType, needle EventType) bool {
|
||||
for _, v := range haystack {
|
||||
if v == needle {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,345 @@
|
||||
// Package ingest subscribes to a single MeshCore MQTT broker and drives the
|
||||
// observation pipeline described in the design doc.
|
||||
//
|
||||
// Call Start() once per broker in a dedicated goroutine. Both broker instances
|
||||
// share the same *hub.Hub and *DB handle so dedup and fan-out are centralised.
|
||||
//
|
||||
// Pipeline per incoming /packets message:
|
||||
// 1. Parse topic → extract IATA + publisher pubkey
|
||||
// 2. Decode hex payload via meshcore-go PacketFromBytes
|
||||
// 3. Compute content-based packet hash (PacketHash)
|
||||
// 4. Upsert observers + observer_brokers + iata_codes
|
||||
// 5. Upsert packets row (ON CONFLICT bump last_heard_at + observation_count)
|
||||
// 6. Insert packet_observations (ON CONFLICT DO NOTHING for cross-broker dedup)
|
||||
// 7. If INSERT succeeded: capability detection, payload-type side effects, fan-out
|
||||
//
|
||||
// Pipeline per incoming /status message:
|
||||
// 1. Parse topic → extract publisher pubkey
|
||||
// 2. Upsert observers row (status_metadata, last_status_at, observer_type, etc.)
|
||||
// 3. Fan out observerStatus event to hub
|
||||
package ingest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||||
"github.com/meshcore-go/meshcore-go"
|
||||
|
||||
"tower/internal/hub"
|
||||
)
|
||||
|
||||
// Config holds the connection parameters for one broker.
|
||||
type Config struct {
|
||||
// BrokerName is a short human-readable label ("mqtt1", "mqtt2") used in
|
||||
// log messages and stored in packet_observations.source_broker.
|
||||
BrokerName string
|
||||
|
||||
// URL is the full broker WebSocket URL, e.g. "wss://mqtt1.meshcore.ca/mqtt"
|
||||
URL string
|
||||
|
||||
Username string
|
||||
Password string
|
||||
}
|
||||
|
||||
// DB is the minimal database interface the ingest pipeline depends on.
|
||||
// Wire in your real *pgxpool.Pool implementation here.
|
||||
type DB interface {
|
||||
// UpsertObserver upserts the observers row keyed on pubkey.
|
||||
UpsertObserver(ctx context.Context, pubkey []byte, iata string) error
|
||||
|
||||
// UpsertObserverBroker records that this observer was seen on brokerName.
|
||||
UpsertObserverBroker(ctx context.Context, pubkey []byte, brokerName string) error
|
||||
|
||||
// UpsertIATA auto-creates an iata_codes row if it doesn't exist yet.
|
||||
UpsertIATA(ctx context.Context, iata string) error
|
||||
|
||||
// UpsertPacket inserts or bumps the packets row. Returns (isNew, error).
|
||||
UpsertPacket(ctx context.Context, p UpsertPacketParams) (bool, error)
|
||||
|
||||
// InsertObservation inserts a packet_observations row.
|
||||
// Returns (inserted, error); inserted=false means ON CONFLICT DO NOTHING fired.
|
||||
InsertObservation(ctx context.Context, o InsertObservationParams) (bool, error)
|
||||
|
||||
// SetNodeCapability flips supports_multibyte_paths or supports_multibyte_traces
|
||||
// for a node, never downgrading an existing TRUE.
|
||||
SetNodeCapability(ctx context.Context, nodeID string, paths, traces bool) error
|
||||
|
||||
// UpsertNode upserts a nodes row from an advert payload.
|
||||
UpsertNode(ctx context.Context, n UpsertNodeParams) error
|
||||
|
||||
// UpsertNodeIATA upserts a node_iatas row.
|
||||
UpsertNodeIATA(ctx context.Context, nodeID string, iata string) error
|
||||
|
||||
// InsertChannelMessage stores a decrypted group text message.
|
||||
InsertChannelMessage(ctx context.Context, m InsertChannelMessageParams) error
|
||||
|
||||
// UpdateObserverStatus updates the observer row from a /status message.
|
||||
UpdateObserverStatus(ctx context.Context, p UpdateObserverStatusParams) error
|
||||
}
|
||||
|
||||
// UpsertPacketParams mirrors the columns written on packets upsert.
|
||||
type UpsertPacketParams struct {
|
||||
PacketHash []byte
|
||||
RouteType uint8
|
||||
PayloadType uint8
|
||||
PayloadVersion uint8
|
||||
TransportCodes []byte // nil if not FLOOD/DIRECT
|
||||
RawPayload []byte
|
||||
ParsedPayload json.RawMessage
|
||||
OriginPubkey []byte
|
||||
ChannelHash []byte
|
||||
}
|
||||
|
||||
// InsertObservationParams mirrors the columns written on packet_observations insert.
|
||||
type InsertObservationParams struct {
|
||||
PacketHash []byte
|
||||
ObserverID string
|
||||
IATA string
|
||||
HeardAt time.Time
|
||||
PathLengthByte uint8
|
||||
HashSize uint8
|
||||
HopCount uint8
|
||||
PathBytes []byte
|
||||
RSSI int16
|
||||
SNR float32
|
||||
PropagationTimeMs int32
|
||||
RadioFreqMHz float32
|
||||
SpreadFactor int16
|
||||
BandwidthKHz float32
|
||||
CodingRate int16
|
||||
SourceBroker string
|
||||
}
|
||||
|
||||
// UpsertNodeParams carries the fields extracted from a payload type 0x04 advert.
|
||||
type UpsertNodeParams struct {
|
||||
PublicKey []byte
|
||||
Name string
|
||||
NodeType uint8 // 1=companion, 2=repeater, 3=room server
|
||||
Latitude *float64
|
||||
Longitude *float64
|
||||
}
|
||||
|
||||
// InsertChannelMessageParams carries a decrypted group text message.
|
||||
type InsertChannelMessageParams struct {
|
||||
ChannelID int
|
||||
PacketHash []byte
|
||||
SenderName string
|
||||
SenderPubkey []byte
|
||||
Content string
|
||||
SentAt time.Time
|
||||
}
|
||||
|
||||
// UpdateObserverStatusParams carries the fields parsed from a /status message.
|
||||
type UpdateObserverStatusParams struct {
|
||||
PublicKey []byte
|
||||
StatusMetadata json.RawMessage
|
||||
LastStatusAt time.Time
|
||||
BatteryLevel *int
|
||||
UptimeSeconds *int64
|
||||
SoftwareVersion string
|
||||
ObserverType string // only set if we can detect it; never downgrade to unknown
|
||||
DisplayName string // only set if current value is NULL
|
||||
}
|
||||
|
||||
// ChannelKeyStore is a read-only view of the channel keys loaded from config.
|
||||
// The ingest layer calls Decrypt and never touches key material directly.
|
||||
type ChannelKeyStore interface {
|
||||
// Decrypt attempts to decrypt groupText bytes using the key for channelHash.
|
||||
// Returns ("", false) if the key is unknown.
|
||||
Decrypt(channelHash []byte, ciphertext []byte) (plaintext string, ok bool)
|
||||
}
|
||||
|
||||
// Worker holds the dependencies for one broker's ingest loop.
|
||||
type Worker struct {
|
||||
cfg Config
|
||||
db DB
|
||||
hub *hub.Hub
|
||||
keys ChannelKeyStore
|
||||
}
|
||||
|
||||
// New creates an ingest Worker. Call Start() to connect and begin processing.
|
||||
func New(cfg Config, db DB, h *hub.Hub, keys ChannelKeyStore) *Worker {
|
||||
return &Worker{cfg: cfg, db: db, hub: h, keys: keys}
|
||||
}
|
||||
|
||||
// Start connects to the broker and blocks until ctx is cancelled. It
|
||||
// reconnects automatically on transient failures using paho's built-in
|
||||
// reconnect logic.
|
||||
//
|
||||
// Intended usage: go worker.Start(ctx)
|
||||
func (w *Worker) Start(ctx context.Context) {
|
||||
opts := mqtt.NewClientOptions().
|
||||
AddBroker(w.cfg.URL).
|
||||
SetClientID(fmt.Sprintf("tower-%s", w.cfg.BrokerName)).
|
||||
SetUsername(w.cfg.Username).
|
||||
SetPassword(w.cfg.Password).
|
||||
SetAutoReconnect(true).
|
||||
SetMaxReconnectInterval(30 * time.Second).
|
||||
SetOnConnectHandler(func(c mqtt.Client) {
|
||||
log.Printf("ingest[%s]: connected to %s", w.cfg.BrokerName, w.cfg.URL)
|
||||
w.subscribe(c)
|
||||
}).
|
||||
SetConnectionLostHandler(func(_ mqtt.Client, err error) {
|
||||
log.Printf("ingest[%s]: connection lost: %v", w.cfg.BrokerName, err)
|
||||
})
|
||||
|
||||
client := mqtt.NewClient(opts)
|
||||
if tok := client.Connect(); tok.Wait() && tok.Error() != nil {
|
||||
log.Printf("ingest[%s]: initial connect failed: %v", w.cfg.BrokerName, tok.Error())
|
||||
// paho will retry; we fall through and wait for ctx
|
||||
}
|
||||
|
||||
<-ctx.Done()
|
||||
client.Disconnect(500)
|
||||
log.Printf("ingest[%s]: stopped", w.cfg.BrokerName)
|
||||
}
|
||||
|
||||
// subscribe registers the wildcard topic handler after (re)connect.
|
||||
func (w *Worker) subscribe(client mqtt.Client) {
|
||||
// meshcore/{IATA}/{pubkey}/packets
|
||||
// meshcore/{IATA}/{pubkey}/status
|
||||
// We do NOT subscribe to /internal (Role 2 access).
|
||||
tok := client.Subscribe("meshcore/#", 1, func(_ mqtt.Client, msg mqtt.Message) {
|
||||
w.handleMessage(msg)
|
||||
})
|
||||
if tok.Wait() && tok.Error() != nil {
|
||||
log.Printf("ingest[%s]: subscribe error: %v", w.cfg.BrokerName, tok.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// handleMessage dispatches incoming MQTT messages by subtopic.
|
||||
func (w *Worker) handleMessage(msg mqtt.Message) {
|
||||
// Topic shape: meshcore/{IATA}/{pubkey}/{subtopic}
|
||||
parts := strings.SplitN(msg.Topic(), "/", 4)
|
||||
if len(parts) != 4 || parts[0] != "meshcore" {
|
||||
return
|
||||
}
|
||||
iata, pubkeyHex, subtopic := parts[1], parts[2], parts[3]
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
switch subtopic {
|
||||
case "packets":
|
||||
w.handlePacket(ctx, iata, pubkeyHex, msg.Payload())
|
||||
case "status":
|
||||
w.handleStatus(ctx, pubkeyHex, msg.Payload())
|
||||
// "internal" is intentionally not handled (Role 2 access)
|
||||
}
|
||||
}
|
||||
|
||||
// handlePacket runs the full observation pipeline for a /packets message.
|
||||
func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw []byte) {
|
||||
var envelope struct {
|
||||
Raw string `json:"raw"` // hex-encoded raw LoRa packet bytes
|
||||
Timestamp string `json:"timestamp"`
|
||||
Hash string `json:"hash"`
|
||||
Origin string `json:"origin"`
|
||||
Type string `json:"type"`
|
||||
Direction string `json:"direction"`
|
||||
Time string `json:"time"`
|
||||
Date string `json:"date"`
|
||||
Len string `json:"len"`
|
||||
PacketType string `json:"packet_type"`
|
||||
Route string `json:"route"`
|
||||
PayloadLen string `json:"payload_len"`
|
||||
OriginID string `json:"origin_id"`
|
||||
SNR string `json:"SNR"`
|
||||
RSSI string `json:"RSSI"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &envelope); err != nil || envelope.Raw == "" {
|
||||
log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex)
|
||||
return
|
||||
}
|
||||
hexBytes, err := hex.DecodeString(envelope.Raw)
|
||||
if err != nil {
|
||||
log.Printf("ingest[%s]: invalid hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
|
||||
}
|
||||
|
||||
// ── Step 2: decode via meshcore-go ──────────────────────────────────────
|
||||
packet, err := meshcore.PacketFromBytes(hexBytes)
|
||||
if err != nil {
|
||||
log.Printf("ingest[%s]: error decoding packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
|
||||
return
|
||||
}
|
||||
packetHash := packet.PacketHash()
|
||||
pathHashes := packet.PathHashes()
|
||||
parsed := packet.PayloadTypeString()
|
||||
// NOTE:
|
||||
// For now we just log and return so the scaffolding compiles.
|
||||
log.Printf("ingest[%s]: packet from %s/%s", w.cfg.BrokerName, iata, pubkeyHex)
|
||||
log.Printf("hash[%x]: path[%x], payload type: %s", packetHash, pathHashes, parsed)
|
||||
|
||||
// ── Step 3–6: DB writes ──────────────────────────────────────────────────
|
||||
// TODO: fill in once meshcore-go is wired.
|
||||
//
|
||||
// _ = w.db.UpsertObserver(ctx, pubkeyBytes, iata)
|
||||
// _ = w.db.UpsertObserverBroker(ctx, pubkeyBytes, w.cfg.BrokerName)
|
||||
// _ = w.db.UpsertIATA(ctx, iata)
|
||||
// isNew, _ := w.db.UpsertPacket(ctx, UpsertPacketParams{...})
|
||||
// inserted, _ := w.db.InsertObservation(ctx, InsertObservationParams{...})
|
||||
|
||||
// ── Step 7: side effects (only if observation INSERT succeeded) ──────────
|
||||
// TODO:
|
||||
// if inserted {
|
||||
// w.runCapabilityDetection(ctx, packet, iata)
|
||||
// w.handlePayloadTypeSideEffects(ctx, packet, iata)
|
||||
// w.fanOut(packet, observation, isNew)
|
||||
// }
|
||||
|
||||
_ = ctx // suppress unused warning until TODOs are filled in
|
||||
}
|
||||
|
||||
// handleStatus processes a /status message and fans out an observerStatus event.
|
||||
func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) {
|
||||
// TODO: parse status JSON, call w.db.UpdateObserverStatus, fan out event.
|
||||
//
|
||||
// params := UpdateObserverStatusParams{...}
|
||||
// _ = w.db.UpdateObserverStatus(ctx, params)
|
||||
//
|
||||
// payload, _ := json.Marshal(statusEvent{...})
|
||||
// w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload})
|
||||
|
||||
log.Printf("ingest[%s]: status from %s (TODO)", w.cfg.BrokerName, pubkeyHex)
|
||||
_ = ctx
|
||||
}
|
||||
|
||||
// runCapabilityDetection checks hash sizes and flips firmware capability flags.
|
||||
// Called only when the observation INSERT succeeded (no dedup conflict).
|
||||
//
|
||||
// Rules (from design doc):
|
||||
// - hash_size == 1: do nothing (proves nothing about firmware)
|
||||
// - duplicate hash prefixes within the path: skip entirely
|
||||
// - non-trace + hash_size 2 or 3 → supports_multibyte_paths = TRUE
|
||||
// - trace (0x09) + hash_size 2 or 4 → supports_multibyte_traces = TRUE
|
||||
//
|
||||
// TODO: implement once meshcore-go PathHashes() is wired.
|
||||
func (w *Worker) runCapabilityDetection(ctx context.Context, payloadType uint8, hashSize uint8, resolvedNodeIDs []string) {
|
||||
if hashSize < 2 {
|
||||
return
|
||||
}
|
||||
for _, nodeID := range resolvedNodeIDs {
|
||||
switch {
|
||||
case payloadType != 0x09 && (hashSize == 2 || hashSize == 3):
|
||||
_ = w.db.SetNodeCapability(ctx, nodeID, true, false)
|
||||
case payloadType == 0x09 && (hashSize == 2 || hashSize == 4):
|
||||
_ = w.db.SetNodeCapability(ctx, nodeID, false, true)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// fanOut builds and broadcasts the packetObservation event to connected WS clients.
|
||||
func (w *Worker) fanOut(packetHash string, payloadType uint8, iata string, isFirst bool, observationCount int64, payload json.RawMessage) {
|
||||
evt := hub.Event{
|
||||
Type: hub.EventPacketObservation,
|
||||
Payload: payload,
|
||||
IATA: iata,
|
||||
PayloadType: payloadType,
|
||||
}
|
||||
w.hub.Broadcast(evt)
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
// Package ws handles the WebSocket endpoint at GET /ws.
|
||||
//
|
||||
// Protocol (from design doc):
|
||||
//
|
||||
// On connect: server sends hello { v:1, type:"hello", serverTime:<ms>, connectionId:"uuid" }
|
||||
//
|
||||
// Client → Server:
|
||||
// subscribe { v, type, id, scope } → server replies subscribed { v, type, id, subscriptionId }
|
||||
// unsubscribe { v, type, id, subscriptionId }
|
||||
// ping { v, type, id } → server replies pong { v, type, id }
|
||||
//
|
||||
// Server → Client events (unsolicited):
|
||||
// packetObservation, observerStatus, nodeUpdate, channelMessage
|
||||
// lagged { v, type, droppedCount, since, lastObservationId }
|
||||
// error { v, type, code, message }
|
||||
//
|
||||
// Idle connections (no ping) closed after 90s.
|
||||
// Client should ping every 30s.
|
||||
package ws
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"tower/internal/hub"
|
||||
)
|
||||
|
||||
const (
|
||||
pingTimeout = 90 * time.Second
|
||||
writeTimeout = 10 * time.Second
|
||||
)
|
||||
|
||||
// Handler returns an http.HandlerFunc that requires the hub to be injected.
|
||||
// Wire it via router.New(h) so the hub is available at startup.
|
||||
func Handler(h *hub.Hub) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: upgrade to WebSocket using nhooyr.io/websocket or gorilla/websocket.
|
||||
// The structure below shows the intended shape; swap the stub conn calls
|
||||
// for real ones once the library is chosen.
|
||||
|
||||
connID := uuid.NewString()
|
||||
client := h.NewClient()
|
||||
defer h.Remove(client)
|
||||
|
||||
ctx, cancel := context.WithCancel(r.Context())
|
||||
defer cancel()
|
||||
|
||||
// Send hello
|
||||
hello := map[string]any{
|
||||
"v": 1,
|
||||
"type": "hello",
|
||||
"serverTime": time.Now().UnixMilli(),
|
||||
"connectionId": connID,
|
||||
}
|
||||
helloBytes, _ := json.Marshal(hello)
|
||||
log.Printf("ws[%s]: connected, hello: %s", connID, helloBytes)
|
||||
// TODO: conn.Write(ctx, websocket.MessageText, helloBytes)
|
||||
|
||||
// Write pump: forward hub events to the WS connection.
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case evt, ok := <-client.Send:
|
||||
if !ok {
|
||||
return // hub closed the channel (client removed)
|
||||
}
|
||||
msg := map[string]any{
|
||||
"v": 1,
|
||||
"type": "event",
|
||||
"event": evt.Type,
|
||||
"data": json.RawMessage(evt.Payload),
|
||||
}
|
||||
msgBytes, _ := json.Marshal(msg)
|
||||
_ = msgBytes
|
||||
// TODO: conn.Write(ctx, websocket.MessageText, msgBytes)
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Read pump: handle subscribe / unsubscribe / ping from the client.
|
||||
// Drives the idle timeout; any message resets the deadline.
|
||||
for {
|
||||
// TODO: _, msgBytes, err := conn.Read(ctx)
|
||||
// if err != nil { return }
|
||||
// w.handleClientMessage(ctx, client, h, connID, msgBytes)
|
||||
|
||||
// Stub: block until context cancelled.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(pingTimeout):
|
||||
log.Printf("ws[%s]: idle timeout", connID)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// clientMessage is the shape of every client → server message.
|
||||
type clientMessage struct {
|
||||
V int `json:"v"`
|
||||
Type string `json:"type"`
|
||||
ID string `json:"id"`
|
||||
SubscriptionID string `json:"subscriptionId,omitempty"`
|
||||
Scope *subscribeScope `json:"scope,omitempty"`
|
||||
}
|
||||
|
||||
// subscribeScope mirrors the scope object in the subscribe message.
|
||||
type subscribeScope struct {
|
||||
IATAs []string `json:"iatas"`
|
||||
RegionIDs []string `json:"regionIds"`
|
||||
PayloadTypes []uint8 `json:"payloadTypes"`
|
||||
RouteTypes []uint8 `json:"routeTypes"`
|
||||
ChannelHashes []string `json:"channelHashes"`
|
||||
ObserverIDs []string `json:"observerIds"`
|
||||
Events []hub.EventType `json:"events"`
|
||||
}
|
||||
|
||||
// handleClientMessage dispatches a parsed client message.
|
||||
// TODO: call this from the read pump once the WS library is wired.
|
||||
func handleClientMessage(ctx context.Context, client *hub.Client, h *hub.Hub, connID string, raw []byte) {
|
||||
var msg clientMessage
|
||||
if err := json.Unmarshal(raw, &msg); err != nil {
|
||||
log.Printf("ws[%s]: bad message: %v", connID, err)
|
||||
return
|
||||
}
|
||||
|
||||
switch msg.Type {
|
||||
case "subscribe":
|
||||
if msg.Scope == nil {
|
||||
return
|
||||
}
|
||||
scope := hub.Scope{
|
||||
IATAs: msg.Scope.IATAs,
|
||||
PayloadTypes: msg.Scope.PayloadTypes,
|
||||
ChannelHashes: msg.Scope.ChannelHashes,
|
||||
Events: msg.Scope.Events,
|
||||
// RegionIATAs: TODO expand msg.Scope.RegionIDs → IATA list via DB/config lookup
|
||||
}
|
||||
h.AddScope(client, scope)
|
||||
subID := uuid.NewString()
|
||||
reply, _ := json.Marshal(map[string]any{
|
||||
"v": 1, "type": "subscribed", "id": msg.ID, "subscriptionId": subID,
|
||||
})
|
||||
log.Printf("ws[%s]: subscribed %s → %s", connID, msg.ID, subID)
|
||||
_ = reply
|
||||
// TODO: conn.Write(ctx, websocket.MessageText, reply)
|
||||
|
||||
case "unsubscribe":
|
||||
// TODO: remove the specific subscriptionId from client.scope.
|
||||
// For now scope entries are append-only; implement removal when needed.
|
||||
log.Printf("ws[%s]: unsubscribe %s (TODO)", connID, msg.SubscriptionID)
|
||||
|
||||
case "ping":
|
||||
reply, _ := json.Marshal(map[string]any{"v": 1, "type": "pong", "id": msg.ID})
|
||||
_ = reply
|
||||
// TODO: conn.Write(ctx, websocket.MessageText, reply)
|
||||
|
||||
default:
|
||||
log.Printf("ws[%s]: unknown message type %q", connID, msg.Type)
|
||||
}
|
||||
|
||||
_ = ctx
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
version: "2"
|
||||
sql:
|
||||
- engine: "postgresql"
|
||||
queries: "db/queries"
|
||||
schema: "db/migrations"
|
||||
gen:
|
||||
go:
|
||||
package: "db"
|
||||
out: "db/sqlc"
|
||||
emit_json_tags: true
|
||||
emit_pointers_for_null_types: true
|
||||
emit_empty_slices: true
|
||||
overrides:
|
||||
- db_type: "uuid"
|
||||
go_type: "github.com/google/uuid.UUID"
|
||||
- db_type: "jsonb"
|
||||
go_type: "encoding/json.RawMessage"
|
||||
- db_type: "bytea"
|
||||
go_type: "[]byte"
|
||||
Reference in New Issue
Block a user