commit 84df3949d1ea803cfdbaa4aa06917fe7de1ad6d1
Author: Enot (ded) Skelly <enot@mailbox.org>
Date:   Wed May 20 11:40:04 2026 -0700

    init

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..4c49bd7
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+.env
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..eeb68f3
--- /dev/null
+++ b/Dockerfile
@@ -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"]
diff --git a/cmd/tower/main.go b/cmd/tower/main.go
new file mode 100644
index 0000000..0800e03
--- /dev/null
+++ b/cmd/tower/main.go
@@ -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
+}
diff --git a/db/migrations/001_schema.sql b/db/migrations/001_schema.sql
new file mode 100644
index 0000000..38f5092
--- /dev/null
+++ b/db/migrations/001_schema.sql
@@ -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);
diff --git a/db/queries/queries.sql b/db/queries/queries.sql
new file mode 100644
index 0000000..d00a6df
--- /dev/null
+++ b/db/queries/queries.sql
@@ -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;
diff --git a/docker-compose.yml b/docker-compose.yml
new file mode 100644
index 0000000..cf87462
--- /dev/null
+++ b/docker-compose.yml
@@ -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:
diff --git a/env.example b/env.example
new file mode 100644
index 0000000..893ffb6
--- /dev/null
+++ b/env.example
@@ -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
diff --git a/go.mod b/go.mod
new file mode 100644
index 0000000..0944636
--- /dev/null
+++ b/go.mod
@@ -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
+)
diff --git a/go.sum b/go.sum
new file mode 100644
index 0000000..cb95e96
--- /dev/null
+++ b/go.sum
@@ -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=
diff --git a/internal/api/handlers/channels.go b/internal/api/handlers/channels.go
new file mode 100644
index 0000000..e814f0a
--- /dev/null
+++ b/internal/api/handlers/channels.go
@@ -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)
+}
diff --git a/internal/api/handlers/iatas.go b/internal/api/handlers/iatas.go
new file mode 100644
index 0000000..1ec816a
--- /dev/null
+++ b/internal/api/handlers/iatas.go
@@ -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)
+}
diff --git a/internal/api/handlers/nodes.go b/internal/api/handlers/nodes.go
new file mode 100644
index 0000000..da679d8
--- /dev/null
+++ b/internal/api/handlers/nodes.go
@@ -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)
+}
diff --git a/internal/api/handlers/observers.go b/internal/api/handlers/observers.go
new file mode 100644
index 0000000..c83a5a8
--- /dev/null
+++ b/internal/api/handlers/observers.go
@@ -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)
+}
diff --git a/internal/api/handlers/packets.go b/internal/api/handlers/packets.go
new file mode 100644
index 0000000..7548588
--- /dev/null
+++ b/internal/api/handlers/packets.go
@@ -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)
+}
diff --git a/internal/api/handlers/regions.go b/internal/api/handlers/regions.go
new file mode 100644
index 0000000..ac2fa55
--- /dev/null
+++ b/internal/api/handlers/regions.go
@@ -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)
+}
diff --git a/internal/api/handlers/stats.go b/internal/api/handlers/stats.go
new file mode 100644
index 0000000..7481a25
--- /dev/null
+++ b/internal/api/handlers/stats.go
@@ -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)
+}
diff --git a/internal/api/middleware/auth.go b/internal/api/middleware/auth.go
new file mode 100644
index 0000000..5f47d92
--- /dev/null
+++ b/internal/api/middleware/auth.go
@@ -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)
+	})
+}
diff --git a/internal/api/router/router.go b/internal/api/router/router.go
new file mode 100644
index 0000000..26e174e
--- /dev/null
+++ b/internal/api/router/router.go
@@ -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
+}
diff --git a/internal/hub/hub.go b/internal/hub/hub.go
new file mode 100644
index 0000000..90eb893
--- /dev/null
+++ b/internal/hub/hub.go
@@ -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
+}
diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go
new file mode 100644
index 0000000..d38529e
--- /dev/null
+++ b/internal/ingest/ingest.go
@@ -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)
+}
diff --git a/internal/ws/handler.go b/internal/ws/handler.go
new file mode 100644
index 0000000..7f386b7
--- /dev/null
+++ b/internal/ws/handler.go
@@ -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
+}
diff --git a/sqlc.yaml b/sqlc.yaml
new file mode 100644
index 0000000..6be8b1a
--- /dev/null
+++ b/sqlc.yaml
@@ -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"