ingestion + ws complete

cmd/tower/main.go

@@ -9,10 +9,13 @@ import (
 	"os/signal"
 	"syscall"
 
+	"tower/db"
 	"tower/internal/api/router"
 	"tower/internal/hub"
 	"tower/internal/ingest"
+	"tower/internal/keystore"
 
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/joho/godotenv"
 )
 
@@ -28,11 +31,18 @@ func main() {
 	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()
 
+	pool, err := pgxpool.New(ctx, mustEnv("POSTGRES_DSN"))
+	if err != nil {
+		log.Fatalf("failed to connect to postgres at %s: %v", os.Getenv("POSTGRES_DSN_HOST"), err)
+	}
+	defer pool.Close()
+
+	store := db.New(pool)
+	keys := keystore.NewMapKeyStore(map[string][]byte{})
+
 	broker1 := ingest.New(
 		ingest.Config{
 			BrokerName: "mqtt1",
@@ -40,9 +50,9 @@ func main() {
 			Username:   mustEnv("MQTT_BROKER_1_USERNAME"),
 			Password:   mustEnv("MQTT_BROKER_1_PASSWORD"),
 		},
-		nil, // TODO: replace with *db.Queries or pgxpool handle
+		store,
 		h,
-		nil, // TODO: replace with channel key store
+		keys,
 	)
 
 	broker2 := ingest.New(
@@ -52,9 +62,9 @@ func main() {
 			Username:   mustEnv("MQTT_BROKER_2_USERNAME"),
 			Password:   mustEnv("MQTT_BROKER_2_PASSWORD"),
 		},
-		nil, // TODO: replace with *db.Queries or pgxpool handle
+		store,
 		h,
-		nil, // TODO: replace with channel key store
+		keys,
 	)
 
 	go broker1.Start(ctx)
@@ -87,8 +97,9 @@ func main() {
 	}
 }
 
-// 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.
+// mustEnv returns the value of an env var and logs a warning if it is unset.
+// Callers that require the value to be non-empty should fatal themselves;
+// ingest workers tolerate missing broker config and will fail on connect instead.
 func mustEnv(key string) string {
 	v := os.Getenv(key)
 	if v == "" {

db/queries/queries.sql

@@ -25,10 +25,10 @@ ON CONFLICT (public_key) DO UPDATE SET
   observation_count = observers.observation_count + 1
 RETURNING *;
 
--- name: UpdateObserverStatus :exec
+-- name: UpdateObserverStatus :one
 UPDATE observers SET
-  display_name     = COALESCE($2, display_name),
-  observer_type    = COALESCE($3, observer_type),
+  display_name     = COALESCE(NULLIF($2, ''), display_name),
+  observer_type    = COALESCE(NULLIF($3, ''), observer_type),
   software_version = COALESCE($4, software_version),
   hardware_model   = COALESCE($5, hardware_model),
   firmware_version = COALESCE($6, firmware_version),
@@ -42,7 +42,8 @@ UPDATE observers SET
   status_metadata  = $14,
   last_status_at   = NOW(),
   last_seen        = NOW()
-WHERE public_key = $1;
+WHERE public_key = $1
+RETURNING id;
 
 -- name: GetObserverByPubkey :one
 SELECT * FROM observers WHERE public_key = $1;
@@ -50,6 +51,17 @@ SELECT * FROM observers WHERE public_key = $1;
 -- name: ListObservers :many
 SELECT * FROM observers ORDER BY last_seen DESC;
 
+-- name: GetObserverLastIATA :one
+SELECT iata FROM packet_observations
+WHERE observer_id = $1
+ORDER BY heard_at DESC
+LIMIT 1;
+
+-- name: GetObserverRadio :one
+SELECT radio_freq_mhz, radio_bw_khz, radio_sf, radio_cr
+FROM observers
+WHERE id = $1;
+
 -- ============================================================
 -- OBSERVER BROKERS
 -- ============================================================
@@ -264,3 +276,20 @@ SELECT * FROM mv_top_nodes_by_iata
 WHERE ($1::char(3) IS NULL OR iata = $1)
 ORDER BY observation_count DESC
 LIMIT $2;
+
+-- ============================================================
+-- HELPERS
+-- ============================================================
+
+-- name: ResolvePathHashes :many
+SELECT DISTINCT n.id
+FROM node_short_ids ns
+JOIN nodes n ON n.id = ns.node_id
+WHERE ns.iata = $1
+  AND CASE
+    WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 1 THEN ns.prefix_1 = ANY($2)
+    WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 2 THEN ns.prefix_2 = ANY($2)
+    WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 3 THEN ns.prefix_3 = ANY($2)
+    WHEN cardinality($2::bytea[]) > 0 AND length($2[1]) = 4 THEN ns.prefix_4 = ANY($2)
+    ELSE FALSE
+  END;

db/sqlc/db.go

@@ -0,0 +1,32 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.31.1
+
+package db
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+type DBTX interface {
+	Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
+	Query(context.Context, string, ...interface{}) (pgx.Rows, error)
+	QueryRow(context.Context, string, ...interface{}) pgx.Row
+}
+
+func New(db DBTX) *Queries {
+	return &Queries{db: db}
+}
+
+type Queries struct {
+	db DBTX
+}
+
+func (q *Queries) WithTx(tx pgx.Tx) *Queries {
+	return &Queries{
+		db: tx,
+	}
+}

db/sqlc/models.go

@@ -0,0 +1,221 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.31.1
+
+package db
+
+import (
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+)
+
+type Channel struct {
+	ID           int32              `json:"id"`
+	ChannelHash  []byte             `json:"channel_hash"`
+	Name         *string            `json:"name"`
+	IsHashtag    *bool              `json:"is_hashtag"`
+	IsPublic     *bool              `json:"is_public"`
+	KeyKnown     *bool              `json:"key_known"`
+	FirstSeen    pgtype.Timestamptz `json:"first_seen"`
+	LastSeen     pgtype.Timestamptz `json:"last_seen"`
+	MessageCount *int64             `json:"message_count"`
+}
+
+type ChannelKey struct {
+	ChannelID int32              `json:"channel_id"`
+	KeyBytes  []byte             `json:"key_bytes"`
+	AddedAt   pgtype.Timestamptz `json:"added_at"`
+	AddedBy   *string            `json:"added_by"`
+}
+
+type ChannelMessage struct {
+	ID           int64              `json:"id"`
+	ChannelID    int32              `json:"channel_id"`
+	PacketHash   []byte             `json:"packet_hash"`
+	SenderName   *string            `json:"sender_name"`
+	SenderPubkey []byte             `json:"sender_pubkey"`
+	Content      *string            `json:"content"`
+	SentAt       pgtype.Timestamptz `json:"sent_at"`
+}
+
+type IataCode struct {
+	Iata        string             `json:"iata"`
+	DisplayName *string            `json:"display_name"`
+	ApproxLat   *float64           `json:"approx_lat"`
+	ApproxLng   *float64           `json:"approx_lng"`
+	AddedAt     pgtype.Timestamptz `json:"added_at"`
+}
+
+type MvHourlyIataStat struct {
+	Iata             string          `json:"iata"`
+	Hour             pgtype.Interval `json:"hour"`
+	ObservationCount int64           `json:"observation_count"`
+	UniquePackets    int64           `json:"unique_packets"`
+	ActiveObservers  int64           `json:"active_observers"`
+}
+
+type MvTopNodesByIatum struct {
+	Iata             string             `json:"iata"`
+	NodeID           uuid.UUID          `json:"node_id"`
+	Name             *string            `json:"name"`
+	NodeType         int16              `json:"node_type"`
+	ObservationCount *int64             `json:"observation_count"`
+	LastHeard        pgtype.Timestamptz `json:"last_heard"`
+}
+
+type Node struct {
+	ID                      uuid.UUID          `json:"id"`
+	PublicKey               []byte             `json:"public_key"`
+	NodeType                int16              `json:"node_type"`
+	Name                    *string            `json:"name"`
+	Latitude                *float64           `json:"latitude"`
+	Longitude               *float64           `json:"longitude"`
+	LocationSource          *string            `json:"location_source"`
+	LastAdvertAt            pgtype.Timestamptz `json:"last_advert_at"`
+	SupportsMultibytePaths  bool               `json:"supports_multibyte_paths"`
+	SupportsMultibyteTraces bool               `json:"supports_multibyte_traces"`
+	MinFirmwareVersion      *string            `json:"min_firmware_version"`
+	FirstSeen               pgtype.Timestamptz `json:"first_seen"`
+	LastSeen                pgtype.Timestamptz `json:"last_seen"`
+	Metadata                []byte             `json:"metadata"`
+}
+
+type NodeIata struct {
+	NodeID           uuid.UUID          `json:"node_id"`
+	Iata             string             `json:"iata"`
+	FirstHeard       pgtype.Timestamptz `json:"first_heard"`
+	LastHeard        pgtype.Timestamptz `json:"last_heard"`
+	ObservationCount *int64             `json:"observation_count"`
+}
+
+type NodeShortID struct {
+	NodeID  uuid.UUID `json:"node_id"`
+	Iata    string    `json:"iata"`
+	Prefix4 []byte    `json:"prefix_4"`
+	Prefix1 []byte    `json:"prefix_1"`
+	Prefix2 []byte    `json:"prefix_2"`
+	Prefix3 []byte    `json:"prefix_3"`
+}
+
+type Observer struct {
+	ID               uuid.UUID          `json:"id"`
+	PublicKey        []byte             `json:"public_key"`
+	DisplayName      *string            `json:"display_name"`
+	ObserverType     *string            `json:"observer_type"`
+	SoftwareVersion  *string            `json:"software_version"`
+	HardwareModel    *string            `json:"hardware_model"`
+	FirmwareVersion  *string            `json:"firmware_version"`
+	FirmwareBuild    *string            `json:"firmware_build"`
+	RadioFreqMhz     *float32           `json:"radio_freq_mhz"`
+	RadioSf          *int16             `json:"radio_sf"`
+	RadioBwKhz       *float32           `json:"radio_bw_khz"`
+	RadioCr          *int16             `json:"radio_cr"`
+	BatteryLevel     *float32           `json:"battery_level"`
+	UptimeSeconds    *int64             `json:"uptime_seconds"`
+	StatusMetadata   []byte             `json:"status_metadata"`
+	LastStatusAt     pgtype.Timestamptz `json:"last_status_at"`
+	FirstSeen        pgtype.Timestamptz `json:"first_seen"`
+	LastSeen         pgtype.Timestamptz `json:"last_seen"`
+	ObservationCount *int64             `json:"observation_count"`
+	Metadata         []byte             `json:"metadata"`
+}
+
+type ObserverBroker struct {
+	ObserverID   uuid.UUID          `json:"observer_id"`
+	BrokerName   string             `json:"broker_name"`
+	FirstSeen    pgtype.Timestamptz `json:"first_seen"`
+	LastSeen     pgtype.Timestamptz `json:"last_seen"`
+	LastPacketAt pgtype.Timestamptz `json:"last_packet_at"`
+	AuthOk       *bool              `json:"auth_ok"`
+}
+
+type ObserverLocation struct {
+	ObserverID uuid.UUID          `json:"observer_id"`
+	Iata       *string            `json:"iata"`
+	Latitude   *float64           `json:"latitude"`
+	Longitude  *float64           `json:"longitude"`
+	ReportedAt pgtype.Timestamptz `json:"reported_at"`
+}
+
+type ObserverOwner struct {
+	ObserverID   uuid.UUID          `json:"observer_id"`
+	OwnerNodeID  pgtype.UUID        `json:"owner_node_id"`
+	OwnerPubkey  []byte             `json:"owner_pubkey"`
+	ContactName  *string            `json:"contact_name"`
+	ContactEmail *string            `json:"contact_email"`
+	Notes        *string            `json:"notes"`
+	Source       *string            `json:"source"`
+	AddedAt      pgtype.Timestamptz `json:"added_at"`
+	UpdatedAt    pgtype.Timestamptz `json:"updated_at"`
+}
+
+type ObserverTelemetry struct {
+	ID               int64              `json:"id"`
+	ObserverID       uuid.UUID          `json:"observer_id"`
+	ReportedAt       pgtype.Timestamptz `json:"reported_at"`
+	BatteryVoltageMv *int32             `json:"battery_voltage_mv"`
+	AirtimeTxPct     *float32           `json:"airtime_tx_pct"`
+	AirtimeRxPct     *float32           `json:"airtime_rx_pct"`
+	NoiseFloorDb     *float32           `json:"noise_floor_db"`
+	UptimeSeconds    *int64             `json:"uptime_seconds"`
+	QueueLength      *int32             `json:"queue_length"`
+	DebugFlags       *int32             `json:"debug_flags"`
+	ReceiveErrors    *int32             `json:"receive_errors"`
+}
+
+type Packet struct {
+	PacketHash            []byte             `json:"packet_hash"`
+	PayloadType           int16              `json:"payload_type"`
+	PayloadVersion        int16              `json:"payload_version"`
+	RouteType             int16              `json:"route_type"`
+	TransportCodesPresent *bool              `json:"transport_codes_present"`
+	RegionCode            *int32             `json:"region_code"`
+	SubRegionCode         *int32             `json:"sub_region_code"`
+	OriginPubkey          []byte             `json:"origin_pubkey"`
+	RawPayload            []byte             `json:"raw_payload"`
+	ParsedPayload         []byte             `json:"parsed_payload"`
+	Decrypted             *bool              `json:"decrypted"`
+	ChannelHash           []byte             `json:"channel_hash"`
+	FirstHeardAt          pgtype.Timestamptz `json:"first_heard_at"`
+	LastHeardAt           pgtype.Timestamptz `json:"last_heard_at"`
+	ObservationCount      *int32             `json:"observation_count"`
+}
+
+type PacketObservation struct {
+	ID                int64              `json:"id"`
+	PacketHash        []byte             `json:"packet_hash"`
+	ObserverID        uuid.UUID          `json:"observer_id"`
+	Iata              string             `json:"iata"`
+	HeardAt           pgtype.Timestamptz `json:"heard_at"`
+	PathLengthByte    int16              `json:"path_length_byte"`
+	HashSize          int16              `json:"hash_size"`
+	HopCount          int16              `json:"hop_count"`
+	PathBytes         []byte             `json:"path_bytes"`
+	Rssi              *int16             `json:"rssi"`
+	Snr               *float32           `json:"snr"`
+	PropagationTimeMs *int32             `json:"propagation_time_ms"`
+	RadioFreqMhz      *float32           `json:"radio_freq_mhz"`
+	SpreadFactor      *int16             `json:"spread_factor"`
+	BandwidthKhz      *float32           `json:"bandwidth_khz"`
+	CodingRate        *int16             `json:"coding_rate"`
+	SourceBroker      *string            `json:"source_broker"`
+}
+
+type Region struct {
+	ID           int32              `json:"id"`
+	Slug         string             `json:"slug"`
+	Name         string             `json:"name"`
+	Description  *string            `json:"description"`
+	DisplayOrder *int32             `json:"display_order"`
+	CenterLat    *float64           `json:"center_lat"`
+	CenterLng    *float64           `json:"center_lng"`
+	ZoomLevel    *int32             `json:"zoom_level"`
+	CreatedAt    pgtype.Timestamptz `json:"created_at"`
+	UpdatedAt    pgtype.Timestamptz `json:"updated_at"`
+}
+
+type RegionIata struct {
+	RegionID int32              `json:"region_id"`
+	Iata     string             `json:"iata"`
+	AddedAt  pgtype.Timestamptz `json:"added_at"`
+}

db/store.go

@@ -0,0 +1,210 @@
+// Package db implements the ingest.DB interface using sqlc-generated queries
+// over a pgx/v5 connection pool. Each method is a thin mapping layer between
+// the ingest param structs and the sqlc-generated param structs.
+package db
+
+import (
+	"context"
+	"encoding/binary"
+	"errors"
+
+	sqlc "tower/db/sqlc"
+	"tower/internal/ingest"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type Store struct {
+	q *sqlc.Queries
+}
+
+func New(pool *pgxpool.Pool) *Store {
+	return &Store{q: sqlc.New(pool)}
+}
+
+// UpsertObserver upserts the observers row keyed on pubkey.
+func (s *Store) UpsertObserver(ctx context.Context, pubkey []byte) (uuid.UUID, error) {
+	row, err := s.q.UpsertObserver(ctx, pubkey)
+	if err != nil {
+		return uuid.Nil, err
+	}
+	return row.ID, err
+}
+
+// UpsertObserverBroker records that this observer was seen on brokerName.
+func (s *Store) UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string) error {
+	params := sqlc.UpsertObserverBrokerParams{
+		ObserverID: observerID,
+		BrokerName: brokerName,
+	}
+	return s.q.UpsertObserverBroker(ctx, params)
+}
+
+// UpsertIATA auto-creates an iata_codes row if it doesn't exist yet.
+func (s *Store) UpsertIATA(ctx context.Context, iata string) error {
+	return s.q.UpsertIATA(ctx, iata)
+}
+
+// UpsertPacket inserts or bumps the packets row. Returns (isNew, error).
+func (s *Store) UpsertPacket(ctx context.Context, p ingest.UpsertPacketParams) (bool, error) {
+	var regionCode, subRegionCode *int32
+	hasTransportCodes := len(p.TransportCodes) == 4
+	if hasTransportCodes {
+		r := int32(binary.LittleEndian.Uint16(p.TransportCodes[0:2]))
+		s := int32(binary.LittleEndian.Uint16(p.TransportCodes[2:4]))
+		regionCode = &r
+		subRegionCode = &s
+	}
+	params := sqlc.UpsertPacketParams{
+		PacketHash:            p.PacketHash,
+		PayloadType:           int16(p.PayloadType),
+		PayloadVersion:        int16(p.PayloadVersion),
+		RouteType:             int16(p.RouteType),
+		TransportCodesPresent: &hasTransportCodes,
+		RegionCode:            regionCode,
+		SubRegionCode:         subRegionCode,
+		OriginPubkey:          p.OriginPubkey,
+		RawPayload:            p.RawPayload,
+		ParsedPayload:         p.ParsedPayload,
+		ChannelHash:           p.ChannelHash,
+	}
+	row, err := s.q.UpsertPacket(ctx, params)
+	if err != nil {
+		return false, err
+	}
+	return row.Inserted, nil
+}
+
+// InsertObservation inserts a packet_observations row.
+// Returns (inserted, error); inserted=false means ON CONFLICT DO NOTHING fired.
+func (s *Store) InsertObservation(ctx context.Context, o ingest.InsertObservationParams) (bool, error) {
+	params := sqlc.InsertObservationParams{
+		PacketHash:        o.PacketHash,
+		ObserverID:        o.ObserverID,
+		Iata:              o.IATA,
+		HeardAt:           pgtype.Timestamptz{Time: o.HeardAt, Valid: true},
+		PathLengthByte:    int16(o.PathLengthByte),
+		HashSize:          int16(o.HashSize),
+		HopCount:          int16(o.HopCount),
+		PathBytes:         o.PathBytes,
+		Rssi:              &o.RSSI,
+		Snr:               &o.SNR,
+		PropagationTimeMs: &o.PropagationTimeMs,
+		RadioFreqMhz:      &o.RadioFreqMHz,
+		SpreadFactor:      &o.SpreadFactor,
+		BandwidthKhz:      &o.BandwidthKHz,
+		CodingRate:        &o.CodingRate,
+		SourceBroker:      &o.SourceBroker,
+	}
+	row, err := s.q.InsertObservation(ctx, params)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil // conflict, not an error
+	}
+	return row.ID != 0, nil
+}
+
+// SetNodeCapability flips supports_multibyte_paths or supports_multibyte_traces
+// for a node, never downgrading an existing TRUE.
+func (s *Store) SetNodeCapability(ctx context.Context, nodeID uuid.UUID, paths, traces bool) error {
+	var errs []error
+	if paths {
+		errs = append(errs, s.q.SetNodeMultibytePaths(ctx, nodeID))
+	}
+	if traces {
+		errs = append(errs, s.q.SetNodeMultibyteTraces(ctx, nodeID))
+	}
+	return errors.Join(errs...)
+}
+
+// UpsertNode upserts a nodes row from an advert payload.
+func (s *Store) UpsertNode(ctx context.Context, n ingest.UpsertNodeParams) (uuid.UUID, error) {
+	params := sqlc.UpsertNodeParams{
+		PublicKey: n.PublicKey,
+		NodeType:  int16(n.NodeType),
+		Name:      &n.Name,
+		Latitude:  n.Latitude,
+		Longitude: n.Longitude,
+	}
+	row, err := s.q.UpsertNode(ctx, params)
+	if err != nil {
+		return uuid.Nil, err
+	}
+	return row.ID, nil
+}
+
+// UpsertNodeIATA upserts a node_iatas row.
+func (s *Store) UpsertNodeIATA(ctx context.Context, nodeID uuid.UUID, iata string) error {
+	params := sqlc.UpsertNodeIATAParams{NodeID: nodeID, Iata: iata}
+	return s.q.UpsertNodeIATA(ctx, params)
+}
+
+// InsertChannelMessage stores a decrypted group text message.
+func (s *Store) InsertChannelMessage(ctx context.Context, m ingest.InsertChannelMessageParams) error {
+	params := sqlc.InsertChannelMessageParams{ChannelID: int32(m.ChannelID), PacketHash: m.PacketHash, SenderName: &m.SenderName, Content: &m.Content, SentAt: pgtype.Timestamptz{Time: m.SentAt, Valid: true}}
+	return s.q.InsertChannelMessage(ctx, params)
+}
+
+// UpdateObserverStatus updates the observer row from a /status message.
+// Column2 = display_name, Column3 = observer_type (sqlc loses names inside CASE expressions).
+func (s *Store) UpdateObserverStatus(ctx context.Context, p ingest.UpdateObserverStatusParams) (uuid.UUID, error) {
+	params := sqlc.UpdateObserverStatusParams{PublicKey: p.PublicKey, Column2: p.DisplayName, Column3: p.ObserverType, SoftwareVersion: &p.SoftwareVersion, HardwareModel: &p.HardwareModel, FirmwareVersion: &p.FirmwareVersion, FirmwareBuild: &p.FirmwareBuild, RadioFreqMhz: &p.RadioFreqMHz, RadioSf: &p.RadioSF, RadioBwKhz: &p.RadioBWKHz, RadioCr: &p.RadioCR, BatteryLevel: p.BatteryLevel, UptimeSeconds: p.UptimeSeconds, StatusMetadata: p.StatusMetadata}
+	return s.q.UpdateObserverStatus(ctx, params)
+}
+
+// GetObserverLastIATA returns the IATA from the most recent observation for the given observer.
+func (s *Store) GetObserverLastIATA(ctx context.Context, observerID uuid.UUID) (string, error) {
+	return s.q.GetObserverLastIATA(ctx, observerID)
+}
+
+// GetObserverRadio returns the current radio settings for the given observer.
+func (s *Store) GetObserverRadio(ctx context.Context, observerID uuid.UUID) (ingest.RadioSettings, error) {
+	row, err := s.q.GetObserverRadio(ctx, observerID)
+	if err != nil {
+		return ingest.RadioSettings{}, err
+	}
+	var settings ingest.RadioSettings
+	if row.RadioFreqMhz != nil {
+		settings.FreqMHz = *row.RadioFreqMhz
+	}
+	if row.RadioSf != nil {
+		settings.SF = *row.RadioSf
+	}
+	if row.RadioBwKhz != nil {
+		settings.BWKHz = *row.RadioBwKhz
+	}
+	if row.RadioCr != nil {
+		settings.CR = *row.RadioCr
+	}
+	return settings, nil
+}
+
+// ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA.
+// Hash size is inferred from the length of the first element in hashes.
+func (s *Store) ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) ([]uuid.UUID, error) {
+	rows, err := s.q.ResolvePathHashes(ctx, sqlc.ResolvePathHashesParams{
+		Iata:    iata,
+		Column2: hashes,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	ids := make([]uuid.UUID, len(rows))
+	copy(ids, rows)
+	return ids, nil
+}
+
+// UpsertChannel upserts a channel row and returns its integer ID.
+func (s *Store) UpsertChannel(ctx context.Context, channelHash []byte) (int, error) {
+	row, err := s.q.UpsertChannel(ctx, sqlc.UpsertChannelParams{
+		ChannelHash:  channelHash,
+		MessageCount: nil, // message count bumped separately by InsertChannelMessage
+	})
+	if err != nil {
+		return 0, err
+	}
+	return int(row.ID), nil
+}

go.mod

@@ -3,17 +3,23 @@ module tower
 go 1.26.1
 
 require (
+	github.com/coder/websocket v1.8.14
 	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/jackc/pgx/v5 v5.9.2
+	github.com/joho/godotenv v1.5.1
 	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
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // 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
+	golang.org/x/sync v0.20.0 // indirect
+	golang.org/x/text v0.37.0 // indirect
 )

go.sum

@@ -1,5 +1,10 @@
 filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
 filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
+github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
+github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 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=
@@ -8,13 +13,34 @@ 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/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw=
+github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 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=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 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=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
+golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/ingest/ingest.go

@@ -21,14 +21,17 @@ package ingest
 
 import (
 	"context"
+	"encoding/binary"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"log"
+	"strconv"
 	"strings"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/google/uuid"
 	"github.com/meshcore-go/meshcore-go"
 
 	"tower/internal/hub"
@@ -51,10 +54,10 @@ type Config struct {
 // 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
+	UpsertObserver(ctx context.Context, pubkey []byte) (uuid.UUID, error)
 
 	// UpsertObserverBroker records that this observer was seen on brokerName.
-	UpsertObserverBroker(ctx context.Context, pubkey []byte, brokerName string) error
+	UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string) error
 
 	// UpsertIATA auto-creates an iata_codes row if it doesn't exist yet.
 	UpsertIATA(ctx context.Context, iata string) error
@@ -68,19 +71,32 @@ type DB interface {
 
 	// 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
+	SetNodeCapability(ctx context.Context, nodeID uuid.UUID, paths, traces bool) error
 
 	// UpsertNode upserts a nodes row from an advert payload.
-	UpsertNode(ctx context.Context, n UpsertNodeParams) error
+	UpsertNode(ctx context.Context, n UpsertNodeParams) (uuid.UUID, error)
 
 	// UpsertNodeIATA upserts a node_iatas row.
-	UpsertNodeIATA(ctx context.Context, nodeID string, iata string) error
+	UpsertNodeIATA(ctx context.Context, nodeID uuid.UUID, 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
+	// UpdateObserverStatus updates the observer row from a /status message. Returns the OberserID
+	// and any error.
+	UpdateObserverStatus(ctx context.Context, p UpdateObserverStatusParams) (uuid.UUID, error)
+
+	// GetObserverLastIATA returns the IATA from the most recent observation for the given observer.
+	GetObserverLastIATA(ctx context.Context, observerID uuid.UUID) (string, error)
+
+	// GetObserverRadio returns the current radio settings for the given observer.
+	GetObserverRadio(ctx context.Context, observerID uuid.UUID) (RadioSettings, error)
+
+	// ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA.
+	ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) ([]uuid.UUID, error)
+
+	// UpsertChannel upserts a channel row and returns its integer ID.
+	UpsertChannel(ctx context.Context, channelHash []byte) (int, error)
 }
 
 // UpsertPacketParams mirrors the columns written on packets upsert.
@@ -99,7 +115,7 @@ type UpsertPacketParams struct {
 // InsertObservationParams mirrors the columns written on packet_observations insert.
 type InsertObservationParams struct {
 	PacketHash        []byte
-	ObserverID        string
+	ObserverID        uuid.UUID
 	IATA              string
 	HeardAt           time.Time
 	PathLengthByte    uint8
@@ -127,12 +143,11 @@ type UpsertNodeParams struct {
 
 // InsertChannelMessageParams carries a decrypted group text message.
 type InsertChannelMessageParams struct {
-	ChannelID    int
-	PacketHash   []byte
-	SenderName   string
-	SenderPubkey []byte
-	Content      string
-	SentAt       time.Time
+	ChannelID  int
+	PacketHash []byte
+	SenderName string
+	Content    string
+	SentAt     time.Time
 }
 
 // UpdateObserverStatusParams carries the fields parsed from a /status message.
@@ -140,19 +155,66 @@ type UpdateObserverStatusParams struct {
 	PublicKey       []byte
 	StatusMetadata  json.RawMessage
 	LastStatusAt    time.Time
-	BatteryLevel    *int
+	BatteryLevel    *float32
 	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
+	HardwareModel   string
+	FirmwareVersion string
+	FirmwareBuild   string
+	RadioFreqMHz    float32
+	RadioSF         int16
+	RadioBWKHz      float32
+	RadioCR         int16
+}
+
+// RadioSettings holds the radio configuration for an observer, populated from
+// /status messages and copied onto each observation row for RF analysis.
+type RadioSettings struct {
+	FreqMHz float32 // MHz, e.g. 910.525
+	SF      int16   // LoRa spreading factor, e.g. 7
+	BWKHz   float32 // bandwidth in kHz, e.g. 62.5
+	CR      int16   // coding rate denominator, e.g. 5 means 4/5
+}
+
+// statusEvent is the JSON payload for an observerStatus WS event.
+// Shape matches the design doc § Server → Client events.
+type statusEvent struct {
+	ObserverID    string `json:"observerId"`
+	DisplayName   string `json:"displayName"`
+	IATA          string `json:"iata,omitempty"`
+	Online        bool   `json:"online"`
+	BatteryMV     int    `json:"batteryMv,omitempty"`
+	UptimeSeconds int64  `json:"uptimeSeconds"`
+	LastStatusAt  int64  `json:"lastStatusAt"` // epoch ms
+}
+
+// packetObservationEvent is the JSON payload for a packetObservation WS event.
+// Shape matches the design doc § Server → Client events.
+type packetObservationEvent struct {
+	PacketHash string `json:"packetHash"`
+	Packet     struct {
+		PayloadType        uint8  `json:"payloadType"`
+		PayloadTypeName    string `json:"payloadTypeName"`
+		RouteType          uint8  `json:"routeType"`
+		IsFirstObservation bool   `json:"isFirstObservation"`
+	} `json:"packet"`
+	Observation struct {
+		ObserverID   string  `json:"observerId"`
+		IATA         string  `json:"iata"`
+		HeardAt      int64   `json:"heardAt"`
+		RSSI         int16   `json:"rssi"`
+		SNR          float32 `json:"snr"`
+		SourceBroker string  `json:"sourceBroker"`
+	} `json:"observation"`
 }
 
 // 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)
+	// GetKey returns the channel key for the given hash, or nil if unknown
+	GetKey(channelHash []byte) []byte
 }
 
 // Worker holds the dependencies for one broker's ingest loop.
@@ -236,77 +298,264 @@ func (w *Worker) handleMessage(msg mqtt.Message) {
 // 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"`
+		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        json.RawMessage `json:"SNR"`
+		RSSI       json.RawMessage `json:"RSSI"`
 	}
-	if err := json.Unmarshal(raw, &envelope); err != nil || envelope.Raw == "" {
+	err := json.Unmarshal(raw, &envelope)
+	if err != nil {
 		log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex)
 		return
 	}
-	hexBytes, err := hex.DecodeString(envelope.Raw)
+	if envelope.Raw == "" {
+		if envelope.Len == "0" || envelope.Len == "" {
+			return // observer keepalive with no packet data
+		}
+		log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex)
+		return
+	}
+	hexBytes, err := hex.DecodeString(strings.ReplaceAll(envelope.Raw, " ", ""))
 	if err != nil {
 		log.Printf("ingest[%s]: invalid hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		return
 	}
 
-	// ── 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
 	}
+
+	pubkeyBytes, err := hex.DecodeString(pubkeyHex)
+	if err != nil {
+		log.Printf("ingest[%s]: invalid pubkey hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		return
+	}
+
+	id, err := w.db.UpsertObserver(ctx, pubkeyBytes)
+	if err != nil {
+		log.Printf("ingest[%s]: db: upsert observer failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		return
+	}
+	err = w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName)
+	if err != nil {
+		log.Printf("ingest[%s]: db: update observer broker failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		return
+	}
+	err = w.db.UpsertIATA(ctx, iata)
+	if err != nil {
+		log.Printf("ingest[%s]: db: upsert IATA failed with 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)
+	var transportCodes []byte
+	if packet.IsTransport() {
+		transportCodes = make([]byte, 4)
+		binary.LittleEndian.PutUint16(transportCodes[0:2], packet.TransportCode1)
+		binary.LittleEndian.PutUint16(transportCodes[2:4], packet.TransportCode2)
+	}
+	parsedPayload, err := json.Marshal(packet.Payload)
+	if err != nil {
+		log.Printf("ingest[%s]: failed to marshal payload from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		return
+	}
+	var channelHash []byte
+	if packet.PayloadType() == meshcore.PayloadTypeGrpTxt {
+		grpTxt, err := meshcore.GroupTextFromBytes(packet.Payload)
+		if err != nil {
+			log.Printf("ingest[%s]: error decoding group text payload from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+			return
+		}
+		channelHash = []byte{grpTxt.ChannelHash}
+	}
+	pParams := UpsertPacketParams{
+		PacketHash:     packetHash[:],
+		RouteType:      packet.RouteType(),
+		PayloadType:    packet.PayloadType(),
+		PayloadVersion: packet.PayloadVer(),
+		TransportCodes: transportCodes,
+		RawPayload:     hexBytes,
+		ParsedPayload:  parsedPayload,
+		OriginPubkey:   pubkeyBytes,
+		ChannelHash:    channelHash,
+	}
+	isNew, err := w.db.UpsertPacket(ctx, pParams)
+	if err != nil {
+		log.Printf("ingest[%s]: db: upsert packet failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		return
+	}
+	heardAt, err := time.Parse("2006-01-02T15:04:05.000000", envelope.Timestamp)
+	if err != nil {
+		log.Printf("ingest[%s]: error parsing time from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		return
+	}
 
-	// ── 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{...})
+	radio, err := w.db.GetObserverRadio(ctx, id)
+	if err != nil {
+		log.Printf("ingest[%s]: db: get observer radio failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+	}
+	oParams := InsertObservationParams{
+		PacketHash:        packetHash[:],
+		ObserverID:        id,
+		IATA:              iata,
+		HeardAt:           heardAt,
+		PathLengthByte:    packet.PathLength,
+		HashSize:          packet.PathHashSize(),
+		HopCount:          packet.PathHashCount(),
+		PathBytes:         packet.Path,
+		RSSI:              int16(parseNumber(envelope.RSSI)),
+		SNR:               float32(parseNumber(envelope.SNR)),
+		PropagationTimeMs: 0, // TODO: figure out how to calculate this
+		RadioFreqMHz:      radio.FreqMHz,
+		SpreadFactor:      radio.SF,
+		BandwidthKHz:      radio.BWKHz,
+		CodingRate:        radio.CR,
+		SourceBroker:      w.cfg.BrokerName,
+	}
+	inserted, err := w.db.InsertObservation(ctx, oParams)
+	if err != nil {
+		log.Printf("ingest[%s]: db: insert observation failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err)
+		return
+	}
 
-	// ── 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
+	resolvedIDs, err := w.db.ResolvePathHashes(ctx, iata, packet.PathHashes())
+	if err != nil {
+		log.Printf("ingest[%s]: db: resolve path hashes failed: %v", w.cfg.BrokerName, err)
+		resolvedIDs = []uuid.UUID{}
+	}
+	if inserted {
+		w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs)
+		w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:])
+		w.fanOut(packetHash[:], packet, iata, isNew, id, envelope.Origin, heardAt, oParams.RSSI, oParams.SNR, w.cfg.BrokerName)
+	}
 }
 
 // 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})
+	var envelope struct {
+		ObserverType    string `json:"source"`
+		SoftwareVersion string `json:"client_version"`
+		HardwareModel   string `json:"model"`
+		FirmwareVersion string `json:"firmware_version"`
+		DisplayName     string `json:"origin"`
+		RadioString     string `json:"radio"`
+		Stats           struct {
+			UptimeSeconds int64   `json:"uptime_secs"`
+			BatteryMV     int     `json:"battery_mv"`
+			NoiseFloor    float32 `json:"noise_floor"`
+			QueueLen      int     `json:"queue_len"`
+			DebugFlags    int     `json:"debug_flags"`
+			TxAirSecs     float64 `json:"tx_air_secs"`
+			RxAirSecs     float64 `json:"rx_air_secs"`
+			RecvErrors    int     `json:"recv_errors"`
+		} `json:"stats"`
+	}
+	if err := json.Unmarshal(raw, &envelope); err != nil {
+		log.Printf("ingest[%s]: malformed status envelope from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		return
+	}
+	pubkey, err := hex.DecodeString(pubkeyHex)
+	if err != nil {
+		log.Printf("ingest[%s]: invalid pubkey hex in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		return
+	}
+	_, err = w.db.UpsertObserver(ctx, pubkey)
+	if err != nil {
+		log.Printf("ingest[%s]: db: upsert observer failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		return
+	}
+	params := UpdateObserverStatusParams{
+		PublicKey:      pubkey,
+		StatusMetadata: raw,
+		LastStatusAt:   time.Now(),
+	}
+	params.UptimeSeconds = &envelope.Stats.UptimeSeconds
+	if envelope.Stats.BatteryMV != 0 {
+		batteryLevel := float32(envelope.Stats.BatteryMV) / 1000
+		params.BatteryLevel = &batteryLevel
+	}
+	if envelope.SoftwareVersion != "" {
+		params.SoftwareVersion = envelope.SoftwareVersion
+	}
+	if envelope.ObserverType != "" {
+		params.ObserverType = envelope.ObserverType
+	}
+	if envelope.DisplayName != "" {
+		params.DisplayName = envelope.DisplayName
+	}
+	if envelope.HardwareModel != "" {
+		params.HardwareModel = envelope.HardwareModel
+	}
+	if envelope.FirmwareVersion != "" {
+		params.FirmwareVersion = envelope.FirmwareVersion
+	}
 
-	log.Printf("ingest[%s]: status from %s (TODO)", w.cfg.BrokerName, pubkeyHex)
-	_ = ctx
+	radio := strings.Split(envelope.RadioString, ",")
+	if len(radio) != 4 {
+		log.Printf("ingest[%s]: missing or malformed radio params in status from %s, skipping radio fields", w.cfg.BrokerName, pubkeyHex)
+	} else {
+		freq, err := strconv.ParseFloat(radio[0], 32)
+		if err != nil {
+			log.Printf("ingest[%s]: error parsing radio freq in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		} else {
+			params.RadioFreqMHz = float32(freq)
+		}
+		bw, err := strconv.ParseFloat(radio[1], 32)
+		if err != nil {
+			log.Printf("ingest[%s]: error parsing radio bw in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		} else {
+			params.RadioBWKHz = float32(bw)
+		}
+		sf, err := strconv.ParseInt(radio[2], 10, 16)
+		if err != nil {
+			log.Printf("ingest[%s]: error parsing radio sf in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		} else {
+			params.RadioSF = int16(sf)
+		}
+		cr, err := strconv.ParseInt(radio[3], 10, 16)
+		if err != nil {
+			log.Printf("ingest[%s]: error parsing radio cr in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		} else {
+			params.RadioCR = int16(cr)
+		}
+	}
+
+	observerID, err := w.db.UpdateObserverStatus(ctx, params)
+	if err != nil {
+		log.Printf("ingest[%s]: db: update observer status failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		return
+	}
+	iata, err := w.db.GetObserverLastIATA(ctx, observerID)
+	if err != nil {
+		iata = "" // non-fatal, continue
+	}
+	evt := statusEvent{
+		ObserverID:    observerID.String(),
+		DisplayName:   envelope.DisplayName,
+		IATA:          iata,
+		Online:        true,
+		BatteryMV:     envelope.Stats.BatteryMV,
+		UptimeSeconds: envelope.Stats.UptimeSeconds,
+		LastStatusAt:  time.Now().UnixMilli(),
+	}
+	payload, err := json.Marshal(evt)
+	if err != nil {
+		log.Printf("ingest[%s]: failed to marshal status event payload for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
+		return
+	}
+	w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload, IATA: iata})
 }
 
 // runCapabilityDetection checks hash sizes and flips firmware capability flags.
@@ -317,9 +566,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 //   - 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) {
+func (w *Worker) runCapabilityDetection(ctx context.Context, payloadType uint8, hashSize uint8, resolvedNodeIDs []uuid.UUID) {
 	if hashSize < 2 {
 		return
 	}
@@ -333,13 +580,121 @@ func (w *Worker) runCapabilityDetection(ctx context.Context, payloadType uint8,
 	}
 }
 
+// handlePayloadTypeSideEffects runs payload-type-specific processing after a
+// new observation is confirmed inserted. Currently handles:
+//   - PayloadTypeAdvert (0x04): upsert node and node_iatas
+//   - PayloadTypeGrpTxt (0x05): decrypt and store channel message if key is known
+func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshcore.Packet, iata string, packetHash []byte) {
+	if packet.PayloadType() == meshcore.PayloadTypeAdvert {
+		advert, err := meshcore.AdvertFromBytes(packet.Payload)
+		if err != nil {
+			log.Printf("ingest[%s]: error decoding advert payload: %v", w.cfg.BrokerName, err)
+			return
+		}
+		var lat, lon *float64
+		if advert.AppData().Lat != 0 || advert.AppData().Lon != 0 {
+			la := float64(advert.AppData().Lat)
+			lo := float64(advert.AppData().Lon)
+			lat = &la
+			lon = &lo
+		}
+		params := UpsertNodeParams{
+			PublicKey: advert.PublicKey.PublicKeyBytes(),
+			Name:      advert.AppData().Name,
+			NodeType:  advert.Type(),
+			Latitude:  lat,
+			Longitude: lon,
+		}
+		nodeID, err := w.db.UpsertNode(ctx, params)
+		if err != nil {
+			log.Printf("ingest[%s]: db: upsert node failed: %v", w.cfg.BrokerName, err)
+			return
+		}
+		if err := w.db.UpsertNodeIATA(ctx, nodeID, iata); err != nil {
+			log.Printf("ingest[%s]: db: upsert node IATA failed: %v", w.cfg.BrokerName, err)
+		}
+		return
+	}
+	if packet.PayloadType() == meshcore.PayloadTypeGrpTxt {
+		grpTxt, err := meshcore.GroupTextFromBytes(packet.Payload)
+		if err != nil {
+			log.Printf("ingest[%s]: error decoding group text payload: %v", w.cfg.BrokerName, err)
+			return
+		}
+		key := w.keys.GetKey([]byte{grpTxt.ChannelHash})
+		if key == nil {
+			return // channel key unknown; message stored as encrypted blob only
+		}
+		channelID, err := w.db.UpsertChannel(ctx, []byte{grpTxt.ChannelHash})
+		if err != nil {
+			log.Printf("ingest[%s]: db: upsert channel failed: %v", w.cfg.BrokerName, err)
+			return
+		}
+		payload, err := grpTxt.DecryptStruct(key)
+		if err != nil {
+			log.Printf("ingest[%s]: error decrypting group text: %v", w.cfg.BrokerName, err)
+			return
+		}
+		params := InsertChannelMessageParams{
+			ChannelID:  channelID,
+			PacketHash: packetHash[:],
+			SenderName: payload.Sender,
+			SentAt:     time.Unix(int64(payload.Timestamp), 0),
+			Content:    payload.Text,
+		}
+		err = w.db.InsertChannelMessage(ctx, params)
+		if err != nil {
+			log.Printf("ingest[%s]: db: insert channel message failed: %v", w.cfg.BrokerName, err)
+		}
+		return
+	}
+}
+
 // 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{
+func (w *Worker) fanOut(packetHash []byte, p *meshcore.Packet, iata string, isFirst bool, observerID uuid.UUID, observerName string, heardAt time.Time, rssi int16, snr float32, sourceBroker string) {
+	evt := packetObservationEvent{}
+	evt.PacketHash = hex.EncodeToString(packetHash)
+	evt.Packet.PayloadType = p.PayloadType()
+	evt.Packet.PayloadTypeName = p.PayloadTypeString()
+	evt.Packet.RouteType = p.RouteType()
+	evt.Packet.IsFirstObservation = isFirst
+	evt.Observation.ObserverID = observerID.String()
+	evt.Observation.IATA = iata
+	evt.Observation.HeardAt = heardAt.UnixMilli()
+	evt.Observation.RSSI = rssi
+	evt.Observation.SNR = snr
+	evt.Observation.SourceBroker = sourceBroker
+
+	payload, err := json.Marshal(evt)
+	if err != nil {
+		log.Printf("ingest[%s]: failed to marshal packetObservation event: %v", w.cfg.BrokerName, err)
+		return
+	}
+	w.hub.Broadcast(hub.Event{
 		Type:        hub.EventPacketObservation,
 		Payload:     payload,
 		IATA:        iata,
-		PayloadType: payloadType,
-	}
-	w.hub.Broadcast(evt)
+		PayloadType: p.PayloadType(),
+	})
+}
+
+// parseNumber handles RSSI and SNR fields that different observer types send as
+// either a bare JSON number (e.g. -108) or a quoted string (e.g. "-108").
+// Returns 0 if the value is missing or unparseable.
+func parseNumber(raw json.RawMessage) float64 {
+	if len(raw) == 0 {
+		return 0
+	}
+	// try unquoted number first
+	var f float64
+	if err := json.Unmarshal(raw, &f); err == nil {
+		return f
+	}
+	// try quoted string
+	var s string
+	if err := json.Unmarshal(raw, &s); err == nil {
+		f, _ = strconv.ParseFloat(s, 64)
+		return f
+	}
+	return 0
 }

internal/keystore/keystore.go

@@ -0,0 +1,18 @@
+// Package keystore provides channel key lookup for the ingest pipeline.
+// Keys are loaded from config at startup and never written at runtime.
+// Future: add a DB-backed fallback that checks the channel_keys table.
+package keystore
+
+import "encoding/hex"
+
+type MapKeyStore struct {
+	keys map[string][]byte // channel hash hex string → key bytes
+}
+
+func NewMapKeyStore(keys map[string][]byte) *MapKeyStore {
+	return &MapKeyStore{keys}
+}
+
+func (s *MapKeyStore) GetKey(channelHash []byte) []byte {
+	return s.keys[hex.EncodeToString(channelHash)]
+}

internal/ws/handler.go

@@ -25,23 +25,26 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/coder/websocket"
 	"github.com/google/uuid"
 
 	"tower/internal/hub"
 )
 
 const (
-	pingTimeout  = 90 * time.Second
-	writeTimeout = 10 * time.Second
+	pingTimeout  = 90 * time.Second // server closes connection if no message received within this window
+	writeTimeout = 10 * time.Second // TODO: apply per-write deadline once nhooyr supports it cleanly
 )
 
 // 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.
+		conn, err := websocket.Accept(w, r, nil)
+		if err != nil {
+			log.Printf("ws: failed to accept connection: %v", err)
+			return
+		}
 
 		connID := uuid.NewString()
 		client := h.NewClient()
@@ -59,7 +62,11 @@ func Handler(h *hub.Hub) http.HandlerFunc {
 		}
 		helloBytes, _ := json.Marshal(hello)
 		log.Printf("ws[%s]: connected, hello: %s", connID, helloBytes)
-		// TODO: conn.Write(ctx, websocket.MessageText, helloBytes)
+		err = conn.Write(ctx, websocket.MessageText, helloBytes)
+		if err != nil {
+			log.Printf("ws[%s]: failed to send hello: %v", connID, err)
+			return
+		}
 
 		// Write pump: forward hub events to the WS connection.
 		go func() {
@@ -76,8 +83,11 @@ func Handler(h *hub.Hub) http.HandlerFunc {
 						"data":  json.RawMessage(evt.Payload),
 					}
 					msgBytes, _ := json.Marshal(msg)
-					_ = msgBytes
-					// TODO: conn.Write(ctx, websocket.MessageText, msgBytes)
+					err = conn.Write(ctx, websocket.MessageText, msgBytes)
+					if err != nil {
+						log.Printf("ws[%s]: failed to write hub event: %v", connID, err)
+						return
+					}
 				case <-ctx.Done():
 					return
 				}
@@ -87,18 +97,14 @@ func Handler(h *hub.Hub) http.HandlerFunc {
 		// 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)
+			readCtx, readCancel := context.WithTimeout(ctx, pingTimeout)
+			_, msgBytes, err := conn.Read(readCtx)
+			readCancel()
+			if err != nil {
+				log.Printf("ws[%s]: read error: %v", connID, err)
 				return
 			}
+			handleClientMessage(ctx, client, h, conn, connID, msgBytes)
 		}
 	}
 }
@@ -124,8 +130,7 @@ type subscribeScope struct {
 }
 
 // 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) {
+func handleClientMessage(ctx context.Context, client *hub.Client, h *hub.Hub, conn *websocket.Conn, connID string, raw []byte) {
 	var msg clientMessage
 	if err := json.Unmarshal(raw, &msg); err != nil {
 		log.Printf("ws[%s]: bad message: %v", connID, err)
@@ -142,7 +147,7 @@ func handleClientMessage(ctx context.Context, client *hub.Client, h *hub.Hub, co
 			PayloadTypes:  msg.Scope.PayloadTypes,
 			ChannelHashes: msg.Scope.ChannelHashes,
 			Events:        msg.Scope.Events,
-			// RegionIATAs: TODO expand msg.Scope.RegionIDs → IATA list via DB/config lookup
+			// RegionIATAs: TODO expand msg.Scope.RegionIDs → IATA slice via region_iatas DB lookup
 		}
 		h.AddScope(client, scope)
 		subID := uuid.NewString()
@@ -150,8 +155,10 @@ func handleClientMessage(ctx context.Context, client *hub.Client, h *hub.Hub, co
 			"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)
+		err := conn.Write(ctx, websocket.MessageText, reply)
+		if err != nil {
+			log.Printf("ws[%s]: failed to send subscribed reply: %v", connID, err)
+		}
 
 	case "unsubscribe":
 		// TODO: remove the specific subscriptionId from client.scope.
@@ -160,12 +167,12 @@ func handleClientMessage(ctx context.Context, client *hub.Client, h *hub.Hub, co
 
 	case "ping":
 		reply, _ := json.Marshal(map[string]any{"v": 1, "type": "pong", "id": msg.ID})
-		_ = reply
-		// TODO: conn.Write(ctx, websocket.MessageText, reply)
+		err := conn.Write(ctx, websocket.MessageText, reply)
+		if err != nil {
+			log.Printf("ws[%s]: failed to send pong: %v", connID, err)
+		}
 
 	default:
 		log.Printf("ws[%s]: unknown message type %q", connID, msg.Type)
 	}
-
-	_ = ctx
 }

sqlc.yaml

@@ -6,6 +6,7 @@ sql:
     gen:
       go:
         package: "db"
+        sql_package: "pgx/v5"
         out: "db/sqlc"
         emit_json_tags: true
         emit_pointers_for_null_types: true
@@ -15,5 +16,3 @@ sql:
             go_type: "github.com/google/uuid.UUID"
           - db_type: "jsonb"
             go_type: "encoding/json.RawMessage"
-          - db_type: "bytea"
-            go_type: "[]byte"