From 838681eec7eca7e58c467388e89822485e1e0fc2 Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Thu, 21 May 2026 14:56:42 -0700 Subject: [PATCH] ingestion + ws complete --- cmd/tower/main.go | 27 +- db/queries/queries.sql | 37 +- db/sqlc/db.go | 32 + db/sqlc/models.go | 221 ++++++ db/sqlc/queries.sql.go | 1250 +++++++++++++++++++++++++++++++++ db/store.go | 210 ++++++ go.mod | 10 +- go.sum | 30 +- internal/ingest/ingest.go | 507 +++++++++++-- internal/keystore/keystore.go | 18 + internal/ws/handler.go | 61 +- sqlc.yaml | 3 +- 12 files changed, 2285 insertions(+), 121 deletions(-) create mode 100644 db/sqlc/db.go create mode 100644 db/sqlc/models.go create mode 100644 db/sqlc/queries.sql.go create mode 100644 db/store.go create mode 100644 internal/keystore/keystore.go diff --git a/cmd/tower/main.go b/cmd/tower/main.go index 0800e03..78a1b7f 100644 --- a/cmd/tower/main.go +++ b/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 == "" { diff --git a/db/queries/queries.sql b/db/queries/queries.sql index d00a6df..bcf4c75 100644 --- a/db/queries/queries.sql +++ b/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; diff --git a/db/sqlc/db.go b/db/sqlc/db.go new file mode 100644 index 0000000..468d1fa --- /dev/null +++ b/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, + } +} diff --git a/db/sqlc/models.go b/db/sqlc/models.go new file mode 100644 index 0000000..34bec12 --- /dev/null +++ b/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"` +} diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go new file mode 100644 index 0000000..5e97818 --- /dev/null +++ b/db/sqlc/queries.sql.go @@ -0,0 +1,1250 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: queries.sql + +package db + +import ( + "context" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +const getChannel = `-- name: GetChannel :one +SELECT id, channel_hash, name, is_hashtag, is_public, key_known, first_seen, last_seen, message_count FROM channels WHERE channel_hash = $1 +` + +func (q *Queries) GetChannel(ctx context.Context, channelHash []byte) (Channel, error) { + row := q.db.QueryRow(ctx, getChannel, channelHash) + var i Channel + err := row.Scan( + &i.ID, + &i.ChannelHash, + &i.Name, + &i.IsHashtag, + &i.IsPublic, + &i.KeyKnown, + &i.FirstSeen, + &i.LastSeen, + &i.MessageCount, + ) + return i, err +} + +const getHourlyStats = `-- name: GetHourlyStats :many +SELECT iata, hour, observation_count, unique_packets, active_observers FROM mv_hourly_iata_stats +WHERE ($1::char(3) IS NULL OR iata = $1) + AND hour >= NOW() - $2::interval +ORDER BY iata, hour +` + +type GetHourlyStatsParams struct { + Column1 string `json:"column_1"` + Column2 pgtype.Interval `json:"column_2"` +} + +func (q *Queries) GetHourlyStats(ctx context.Context, arg GetHourlyStatsParams) ([]MvHourlyIataStat, error) { + rows, err := q.db.Query(ctx, getHourlyStats, arg.Column1, arg.Column2) + if err != nil { + return nil, err + } + defer rows.Close() + items := []MvHourlyIataStat{} + for rows.Next() { + var i MvHourlyIataStat + if err := rows.Scan( + &i.Iata, + &i.Hour, + &i.ObservationCount, + &i.UniquePackets, + &i.ActiveObservers, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getIATA = `-- name: GetIATA :one +SELECT iata, display_name, approx_lat, approx_lng, added_at FROM iata_codes WHERE iata = $1 +` + +func (q *Queries) GetIATA(ctx context.Context, iata string) (IataCode, error) { + row := q.db.QueryRow(ctx, getIATA, iata) + var i IataCode + err := row.Scan( + &i.Iata, + &i.DisplayName, + &i.ApproxLat, + &i.ApproxLng, + &i.AddedAt, + ) + return i, err +} + +const getNodeByPubkey = `-- name: GetNodeByPubkey :one +SELECT id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, metadata FROM nodes WHERE public_key = $1 +` + +func (q *Queries) GetNodeByPubkey(ctx context.Context, publicKey []byte) (Node, error) { + row := q.db.QueryRow(ctx, getNodeByPubkey, publicKey) + var i Node + err := row.Scan( + &i.ID, + &i.PublicKey, + &i.NodeType, + &i.Name, + &i.Latitude, + &i.Longitude, + &i.LocationSource, + &i.LastAdvertAt, + &i.SupportsMultibytePaths, + &i.SupportsMultibyteTraces, + &i.MinFirmwareVersion, + &i.FirstSeen, + &i.LastSeen, + &i.Metadata, + ) + return i, err +} + +const getObserverByPubkey = `-- name: GetObserverByPubkey :one +SELECT id, public_key, display_name, observer_type, software_version, hardware_model, firmware_version, firmware_build, radio_freq_mhz, radio_sf, radio_bw_khz, radio_cr, battery_level, uptime_seconds, status_metadata, last_status_at, first_seen, last_seen, observation_count, metadata FROM observers WHERE public_key = $1 +` + +func (q *Queries) GetObserverByPubkey(ctx context.Context, publicKey []byte) (Observer, error) { + row := q.db.QueryRow(ctx, getObserverByPubkey, publicKey) + var i Observer + err := row.Scan( + &i.ID, + &i.PublicKey, + &i.DisplayName, + &i.ObserverType, + &i.SoftwareVersion, + &i.HardwareModel, + &i.FirmwareVersion, + &i.FirmwareBuild, + &i.RadioFreqMhz, + &i.RadioSf, + &i.RadioBwKhz, + &i.RadioCr, + &i.BatteryLevel, + &i.UptimeSeconds, + &i.StatusMetadata, + &i.LastStatusAt, + &i.FirstSeen, + &i.LastSeen, + &i.ObservationCount, + &i.Metadata, + ) + return i, err +} + +const getObserverLastIATA = `-- name: GetObserverLastIATA :one +SELECT iata FROM packet_observations +WHERE observer_id = $1 +ORDER BY heard_at DESC +LIMIT 1 +` + +func (q *Queries) GetObserverLastIATA(ctx context.Context, observerID uuid.UUID) (string, error) { + row := q.db.QueryRow(ctx, getObserverLastIATA, observerID) + var iata string + err := row.Scan(&iata) + return iata, err +} + +const getObserverRadio = `-- name: GetObserverRadio :one +SELECT radio_freq_mhz, radio_bw_khz, radio_sf, radio_cr +FROM observers +WHERE id = $1 +` + +type GetObserverRadioRow struct { + RadioFreqMhz *float32 `json:"radio_freq_mhz"` + RadioBwKhz *float32 `json:"radio_bw_khz"` + RadioSf *int16 `json:"radio_sf"` + RadioCr *int16 `json:"radio_cr"` +} + +func (q *Queries) GetObserverRadio(ctx context.Context, id uuid.UUID) (GetObserverRadioRow, error) { + row := q.db.QueryRow(ctx, getObserverRadio, id) + var i GetObserverRadioRow + err := row.Scan( + &i.RadioFreqMhz, + &i.RadioBwKhz, + &i.RadioSf, + &i.RadioCr, + ) + return i, err +} + +const getPacket = `-- name: GetPacket :one +SELECT packet_hash, payload_type, payload_version, route_type, transport_codes_present, region_code, sub_region_code, origin_pubkey, raw_payload, parsed_payload, decrypted, channel_hash, first_heard_at, last_heard_at, observation_count FROM packets WHERE packet_hash = $1 +` + +func (q *Queries) GetPacket(ctx context.Context, packetHash []byte) (Packet, error) { + row := q.db.QueryRow(ctx, getPacket, packetHash) + var i Packet + err := row.Scan( + &i.PacketHash, + &i.PayloadType, + &i.PayloadVersion, + &i.RouteType, + &i.TransportCodesPresent, + &i.RegionCode, + &i.SubRegionCode, + &i.OriginPubkey, + &i.RawPayload, + &i.ParsedPayload, + &i.Decrypted, + &i.ChannelHash, + &i.FirstHeardAt, + &i.LastHeardAt, + &i.ObservationCount, + ) + return i, err +} + +const getStatsOverview = `-- 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) +` + +type GetStatsOverviewRow struct { + TotalPackets int64 `json:"total_packets"` + TotalObservations int64 `json:"total_observations"` + ActiveObservers int64 `json:"active_observers"` + ActiveIatas int64 `json:"active_iatas"` +} + +// ============================================================ +// STATS +// ============================================================ +func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 string) (GetStatsOverviewRow, error) { + row := q.db.QueryRow(ctx, getStatsOverview, dollar_1) + var i GetStatsOverviewRow + err := row.Scan( + &i.TotalPackets, + &i.TotalObservations, + &i.ActiveObservers, + &i.ActiveIatas, + ) + return i, err +} + +const getTopNodes = `-- name: GetTopNodes :many +SELECT iata, node_id, name, node_type, observation_count, last_heard FROM mv_top_nodes_by_iata +WHERE ($1::char(3) IS NULL OR iata = $1) +ORDER BY observation_count DESC +LIMIT $2 +` + +type GetTopNodesParams struct { + Column1 string `json:"column_1"` + Limit int32 `json:"limit"` +} + +func (q *Queries) GetTopNodes(ctx context.Context, arg GetTopNodesParams) ([]MvTopNodesByIatum, error) { + rows, err := q.db.Query(ctx, getTopNodes, arg.Column1, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + items := []MvTopNodesByIatum{} + for rows.Next() { + var i MvTopNodesByIatum + if err := rows.Scan( + &i.Iata, + &i.NodeID, + &i.Name, + &i.NodeType, + &i.ObservationCount, + &i.LastHeard, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const insertChannelMessage = `-- 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 +` + +type InsertChannelMessageParams struct { + 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"` +} + +// ============================================================ +// CHANNEL MESSAGES +// ============================================================ +func (q *Queries) InsertChannelMessage(ctx context.Context, arg InsertChannelMessageParams) error { + _, err := q.db.Exec(ctx, insertChannelMessage, + arg.ChannelID, + arg.PacketHash, + arg.SenderName, + arg.SenderPubkey, + arg.Content, + arg.SentAt, + ) + return err +} + +const insertObservation = `-- 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 id, 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 +` + +type InsertObservationParams struct { + 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"` +} + +// ============================================================ +// PACKET OBSERVATIONS +// ============================================================ +func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationParams) (PacketObservation, error) { + row := q.db.QueryRow(ctx, insertObservation, + arg.PacketHash, + arg.ObserverID, + arg.Iata, + arg.HeardAt, + arg.PathLengthByte, + arg.HashSize, + arg.HopCount, + arg.PathBytes, + arg.Rssi, + arg.Snr, + arg.PropagationTimeMs, + arg.RadioFreqMhz, + arg.SpreadFactor, + arg.BandwidthKhz, + arg.CodingRate, + arg.SourceBroker, + ) + var i PacketObservation + err := row.Scan( + &i.ID, + &i.PacketHash, + &i.ObserverID, + &i.Iata, + &i.HeardAt, + &i.PathLengthByte, + &i.HashSize, + &i.HopCount, + &i.PathBytes, + &i.Rssi, + &i.Snr, + &i.PropagationTimeMs, + &i.RadioFreqMhz, + &i.SpreadFactor, + &i.BandwidthKhz, + &i.CodingRate, + &i.SourceBroker, + ) + return i, err +} + +const listChannelMessages = `-- name: ListChannelMessages :many +SELECT id, channel_id, packet_hash, sender_name, sender_pubkey, content, sent_at FROM channel_messages +WHERE channel_id = $1 + AND ($2::timestamptz IS NULL OR sent_at >= $2) +ORDER BY sent_at DESC +LIMIT $3 +` + +type ListChannelMessagesParams struct { + ChannelID int32 `json:"channel_id"` + Column2 pgtype.Timestamptz `json:"column_2"` + Limit int32 `json:"limit"` +} + +func (q *Queries) ListChannelMessages(ctx context.Context, arg ListChannelMessagesParams) ([]ChannelMessage, error) { + rows, err := q.db.Query(ctx, listChannelMessages, arg.ChannelID, arg.Column2, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ChannelMessage{} + for rows.Next() { + var i ChannelMessage + if err := rows.Scan( + &i.ID, + &i.ChannelID, + &i.PacketHash, + &i.SenderName, + &i.SenderPubkey, + &i.Content, + &i.SentAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listChannels = `-- name: ListChannels :many +SELECT id, channel_hash, name, is_hashtag, is_public, key_known, first_seen, last_seen, message_count FROM channels ORDER BY last_seen DESC LIMIT $1 +` + +func (q *Queries) ListChannels(ctx context.Context, limit int32) ([]Channel, error) { + rows, err := q.db.Query(ctx, listChannels, limit) + if err != nil { + return nil, err + } + defer rows.Close() + items := []Channel{} + for rows.Next() { + var i Channel + if err := rows.Scan( + &i.ID, + &i.ChannelHash, + &i.Name, + &i.IsHashtag, + &i.IsPublic, + &i.KeyKnown, + &i.FirstSeen, + &i.LastSeen, + &i.MessageCount, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listIATAs = `-- name: ListIATAs :many +SELECT iata, display_name, approx_lat, approx_lng, added_at FROM iata_codes ORDER BY iata +` + +func (q *Queries) ListIATAs(ctx context.Context) ([]IataCode, error) { + rows, err := q.db.Query(ctx, listIATAs) + if err != nil { + return nil, err + } + defer rows.Close() + items := []IataCode{} + for rows.Next() { + var i IataCode + if err := rows.Scan( + &i.Iata, + &i.DisplayName, + &i.ApproxLat, + &i.ApproxLng, + &i.AddedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listNodes = `-- name: ListNodes :many +SELECT id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, metadata FROM nodes +WHERE + ($1::smallint IS NULL OR node_type = $1) +ORDER BY last_seen DESC +LIMIT $2 +` + +type ListNodesParams struct { + Column1 int16 `json:"column_1"` + Limit int32 `json:"limit"` +} + +func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]Node, error) { + rows, err := q.db.Query(ctx, listNodes, arg.Column1, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + items := []Node{} + for rows.Next() { + var i Node + if err := rows.Scan( + &i.ID, + &i.PublicKey, + &i.NodeType, + &i.Name, + &i.Latitude, + &i.Longitude, + &i.LocationSource, + &i.LastAdvertAt, + &i.SupportsMultibytePaths, + &i.SupportsMultibyteTraces, + &i.MinFirmwareVersion, + &i.FirstSeen, + &i.LastSeen, + &i.Metadata, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listObservationsForObserver = `-- name: ListObservationsForObserver :many +SELECT id, 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 FROM packet_observations +WHERE observer_id = $1 + AND ($2::timestamptz IS NULL OR heard_at >= $2) +ORDER BY heard_at DESC +LIMIT $3 +` + +type ListObservationsForObserverParams struct { + ObserverID uuid.UUID `json:"observer_id"` + Column2 pgtype.Timestamptz `json:"column_2"` + Limit int32 `json:"limit"` +} + +func (q *Queries) ListObservationsForObserver(ctx context.Context, arg ListObservationsForObserverParams) ([]PacketObservation, error) { + rows, err := q.db.Query(ctx, listObservationsForObserver, arg.ObserverID, arg.Column2, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + items := []PacketObservation{} + for rows.Next() { + var i PacketObservation + if err := rows.Scan( + &i.ID, + &i.PacketHash, + &i.ObserverID, + &i.Iata, + &i.HeardAt, + &i.PathLengthByte, + &i.HashSize, + &i.HopCount, + &i.PathBytes, + &i.Rssi, + &i.Snr, + &i.PropagationTimeMs, + &i.RadioFreqMhz, + &i.SpreadFactor, + &i.BandwidthKhz, + &i.CodingRate, + &i.SourceBroker, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listObservationsForPacket = `-- name: ListObservationsForPacket :many +SELECT id, 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 FROM packet_observations +WHERE packet_hash = $1 +ORDER BY heard_at ASC +` + +func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []byte) ([]PacketObservation, error) { + rows, err := q.db.Query(ctx, listObservationsForPacket, packetHash) + if err != nil { + return nil, err + } + defer rows.Close() + items := []PacketObservation{} + for rows.Next() { + var i PacketObservation + if err := rows.Scan( + &i.ID, + &i.PacketHash, + &i.ObserverID, + &i.Iata, + &i.HeardAt, + &i.PathLengthByte, + &i.HashSize, + &i.HopCount, + &i.PathBytes, + &i.Rssi, + &i.Snr, + &i.PropagationTimeMs, + &i.RadioFreqMhz, + &i.SpreadFactor, + &i.BandwidthKhz, + &i.CodingRate, + &i.SourceBroker, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listObservers = `-- name: ListObservers :many +SELECT id, public_key, display_name, observer_type, software_version, hardware_model, firmware_version, firmware_build, radio_freq_mhz, radio_sf, radio_bw_khz, radio_cr, battery_level, uptime_seconds, status_metadata, last_status_at, first_seen, last_seen, observation_count, metadata FROM observers ORDER BY last_seen DESC +` + +func (q *Queries) ListObservers(ctx context.Context) ([]Observer, error) { + rows, err := q.db.Query(ctx, listObservers) + if err != nil { + return nil, err + } + defer rows.Close() + items := []Observer{} + for rows.Next() { + var i Observer + if err := rows.Scan( + &i.ID, + &i.PublicKey, + &i.DisplayName, + &i.ObserverType, + &i.SoftwareVersion, + &i.HardwareModel, + &i.FirmwareVersion, + &i.FirmwareBuild, + &i.RadioFreqMhz, + &i.RadioSf, + &i.RadioBwKhz, + &i.RadioCr, + &i.BatteryLevel, + &i.UptimeSeconds, + &i.StatusMetadata, + &i.LastStatusAt, + &i.FirstSeen, + &i.LastSeen, + &i.ObservationCount, + &i.Metadata, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listPackets = `-- name: ListPackets :many +SELECT p.packet_hash, p.payload_type, p.payload_version, p.route_type, p.transport_codes_present, p.region_code, p.sub_region_code, p.origin_pubkey, p.raw_payload, p.parsed_payload, p.decrypted, p.channel_hash, p.first_heard_at, p.last_heard_at, p.observation_count +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 +` + +type ListPacketsParams struct { + Column1 int16 `json:"column_1"` + Column2 int16 `json:"column_2"` + Column3 pgtype.Timestamptz `json:"column_3"` + Column4 pgtype.Timestamptz `json:"column_4"` + Limit int32 `json:"limit"` +} + +func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Packet, error) { + rows, err := q.db.Query(ctx, listPackets, + arg.Column1, + arg.Column2, + arg.Column3, + arg.Column4, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []Packet{} + for rows.Next() { + var i Packet + if err := rows.Scan( + &i.PacketHash, + &i.PayloadType, + &i.PayloadVersion, + &i.RouteType, + &i.TransportCodesPresent, + &i.RegionCode, + &i.SubRegionCode, + &i.OriginPubkey, + &i.RawPayload, + &i.ParsedPayload, + &i.Decrypted, + &i.ChannelHash, + &i.FirstHeardAt, + &i.LastHeardAt, + &i.ObservationCount, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listPacketsAfterID = `-- name: ListPacketsAfterID :many +SELECT p.packet_hash, p.payload_type, p.payload_version, p.route_type, p.transport_codes_present, p.region_code, p.sub_region_code, p.origin_pubkey, p.raw_payload, p.parsed_payload, p.decrypted, p.channel_hash, p.first_heard_at, p.last_heard_at, p.observation_count +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 +` + +type ListPacketsAfterIDParams struct { + ID int64 `json:"id"` + Limit int32 `json:"limit"` +} + +func (q *Queries) ListPacketsAfterID(ctx context.Context, arg ListPacketsAfterIDParams) ([]Packet, error) { + rows, err := q.db.Query(ctx, listPacketsAfterID, arg.ID, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + items := []Packet{} + for rows.Next() { + var i Packet + if err := rows.Scan( + &i.PacketHash, + &i.PayloadType, + &i.PayloadVersion, + &i.RouteType, + &i.TransportCodesPresent, + &i.RegionCode, + &i.SubRegionCode, + &i.OriginPubkey, + &i.RawPayload, + &i.ParsedPayload, + &i.Decrypted, + &i.ChannelHash, + &i.FirstHeardAt, + &i.LastHeardAt, + &i.ObservationCount, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const resolvePathHashes = `-- 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 +` + +type ResolvePathHashesParams struct { + Iata string `json:"iata"` + Column2 [][]byte `json:"column_2"` +} + +// ============================================================ +// HELPERS +// ============================================================ +func (q *Queries) ResolvePathHashes(ctx context.Context, arg ResolvePathHashesParams) ([]uuid.UUID, error) { + rows, err := q.db.Query(ctx, resolvePathHashes, arg.Iata, arg.Column2) + if err != nil { + return nil, err + } + defer rows.Close() + items := []uuid.UUID{} + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + return nil, err + } + items = append(items, id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const setChannelKeyKnown = `-- name: SetChannelKeyKnown :exec +UPDATE channels SET key_known = TRUE WHERE channel_hash = $1 +` + +func (q *Queries) SetChannelKeyKnown(ctx context.Context, channelHash []byte) error { + _, err := q.db.Exec(ctx, setChannelKeyKnown, channelHash) + return err +} + +const setNodeMultibytePaths = `-- name: SetNodeMultibytePaths :exec +UPDATE nodes SET supports_multibyte_paths = TRUE +WHERE id = $1 AND supports_multibyte_paths = FALSE +` + +func (q *Queries) SetNodeMultibytePaths(ctx context.Context, id uuid.UUID) error { + _, err := q.db.Exec(ctx, setNodeMultibytePaths, id) + return err +} + +const setNodeMultibyteTraces = `-- name: SetNodeMultibyteTraces :exec +UPDATE nodes SET supports_multibyte_traces = TRUE +WHERE id = $1 AND supports_multibyte_traces = FALSE +` + +func (q *Queries) SetNodeMultibyteTraces(ctx context.Context, id uuid.UUID) error { + _, err := q.db.Exec(ctx, setNodeMultibyteTraces, id) + return err +} + +const updateObserverStatus = `-- name: UpdateObserverStatus :one +UPDATE observers SET + display_name = COALESCE(NULLIF($2, ''), display_name), + observer_type = COALESCE(NULLIF($3, ''), observer_type), + software_version = COALESCE($4, software_version), + hardware_model = COALESCE($5, hardware_model), + firmware_version = COALESCE($6, firmware_version), + firmware_build = COALESCE($7, firmware_build), + radio_freq_mhz = COALESCE($8, radio_freq_mhz), + radio_sf = COALESCE($9, radio_sf), + radio_bw_khz = COALESCE($10, radio_bw_khz), + radio_cr = COALESCE($11, radio_cr), + battery_level = COALESCE($12, battery_level), + uptime_seconds = COALESCE($13, uptime_seconds), + status_metadata = $14, + last_status_at = NOW(), + last_seen = NOW() +WHERE public_key = $1 +RETURNING id +` + +type UpdateObserverStatusParams struct { + PublicKey []byte `json:"public_key"` + Column2 interface{} `json:"column_2"` + Column3 interface{} `json:"column_3"` + 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"` +} + +func (q *Queries) UpdateObserverStatus(ctx context.Context, arg UpdateObserverStatusParams) (uuid.UUID, error) { + row := q.db.QueryRow(ctx, updateObserverStatus, + arg.PublicKey, + arg.Column2, + arg.Column3, + arg.SoftwareVersion, + arg.HardwareModel, + arg.FirmwareVersion, + arg.FirmwareBuild, + arg.RadioFreqMhz, + arg.RadioSf, + arg.RadioBwKhz, + arg.RadioCr, + arg.BatteryLevel, + arg.UptimeSeconds, + arg.StatusMetadata, + ) + var id uuid.UUID + err := row.Scan(&id) + return id, err +} + +const upsertChannel = `-- 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 id, channel_hash, name, is_hashtag, is_public, key_known, first_seen, last_seen, message_count +` + +type UpsertChannelParams struct { + ChannelHash []byte `json:"channel_hash"` + MessageCount *int64 `json:"message_count"` +} + +// ============================================================ +// CHANNELS +// ============================================================ +func (q *Queries) UpsertChannel(ctx context.Context, arg UpsertChannelParams) (Channel, error) { + row := q.db.QueryRow(ctx, upsertChannel, arg.ChannelHash, arg.MessageCount) + var i Channel + err := row.Scan( + &i.ID, + &i.ChannelHash, + &i.Name, + &i.IsHashtag, + &i.IsPublic, + &i.KeyKnown, + &i.FirstSeen, + &i.LastSeen, + &i.MessageCount, + ) + return i, err +} + +const upsertIATA = `-- name: UpsertIATA :exec + +INSERT INTO iata_codes (iata) +VALUES ($1) +ON CONFLICT (iata) DO NOTHING +` + +// ============================================================ +// IATA CODES +// ============================================================ +func (q *Queries) UpsertIATA(ctx context.Context, iata string) error { + _, err := q.db.Exec(ctx, upsertIATA, iata) + return err +} + +const upsertNode = `-- 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 id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, metadata +` + +type UpsertNodeParams struct { + PublicKey []byte `json:"public_key"` + NodeType int16 `json:"node_type"` + Name *string `json:"name"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` +} + +// ============================================================ +// NODES +// ============================================================ +func (q *Queries) UpsertNode(ctx context.Context, arg UpsertNodeParams) (Node, error) { + row := q.db.QueryRow(ctx, upsertNode, + arg.PublicKey, + arg.NodeType, + arg.Name, + arg.Latitude, + arg.Longitude, + ) + var i Node + err := row.Scan( + &i.ID, + &i.PublicKey, + &i.NodeType, + &i.Name, + &i.Latitude, + &i.Longitude, + &i.LocationSource, + &i.LastAdvertAt, + &i.SupportsMultibytePaths, + &i.SupportsMultibyteTraces, + &i.MinFirmwareVersion, + &i.FirstSeen, + &i.LastSeen, + &i.Metadata, + ) + return i, err +} + +const upsertNodeIATA = `-- 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 +` + +type UpsertNodeIATAParams struct { + NodeID uuid.UUID `json:"node_id"` + Iata string `json:"iata"` +} + +// ============================================================ +// NODE IATAS +// ============================================================ +func (q *Queries) UpsertNodeIATA(ctx context.Context, arg UpsertNodeIATAParams) error { + _, err := q.db.Exec(ctx, upsertNodeIATA, arg.NodeID, arg.Iata) + return err +} + +const upsertNodeShortID = `-- name: UpsertNodeShortID :exec +INSERT INTO node_short_ids (node_id, iata, prefix_4) +VALUES ($1, $2, $3) +ON CONFLICT (node_id, iata) DO NOTHING +` + +type UpsertNodeShortIDParams struct { + NodeID uuid.UUID `json:"node_id"` + Iata string `json:"iata"` + Prefix4 []byte `json:"prefix_4"` +} + +func (q *Queries) UpsertNodeShortID(ctx context.Context, arg UpsertNodeShortIDParams) error { + _, err := q.db.Exec(ctx, upsertNodeShortID, arg.NodeID, arg.Iata, arg.Prefix4) + return err +} + +const upsertObserver = `-- 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 id, public_key, display_name, observer_type, software_version, hardware_model, firmware_version, firmware_build, radio_freq_mhz, radio_sf, radio_bw_khz, radio_cr, battery_level, uptime_seconds, status_metadata, last_status_at, first_seen, last_seen, observation_count, metadata +` + +// ============================================================ +// OBSERVERS +// ============================================================ +func (q *Queries) UpsertObserver(ctx context.Context, publicKey []byte) (Observer, error) { + row := q.db.QueryRow(ctx, upsertObserver, publicKey) + var i Observer + err := row.Scan( + &i.ID, + &i.PublicKey, + &i.DisplayName, + &i.ObserverType, + &i.SoftwareVersion, + &i.HardwareModel, + &i.FirmwareVersion, + &i.FirmwareBuild, + &i.RadioFreqMhz, + &i.RadioSf, + &i.RadioBwKhz, + &i.RadioCr, + &i.BatteryLevel, + &i.UptimeSeconds, + &i.StatusMetadata, + &i.LastStatusAt, + &i.FirstSeen, + &i.LastSeen, + &i.ObservationCount, + &i.Metadata, + ) + return i, err +} + +const upsertObserverBroker = `-- 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() +` + +type UpsertObserverBrokerParams struct { + ObserverID uuid.UUID `json:"observer_id"` + BrokerName string `json:"broker_name"` +} + +// ============================================================ +// OBSERVER BROKERS +// ============================================================ +func (q *Queries) UpsertObserverBroker(ctx context.Context, arg UpsertObserverBrokerParams) error { + _, err := q.db.Exec(ctx, upsertObserverBroker, arg.ObserverID, arg.BrokerName) + return err +} + +const upsertPacket = `-- 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 packet_hash, payload_type, payload_version, route_type, transport_codes_present, region_code, sub_region_code, origin_pubkey, raw_payload, parsed_payload, decrypted, channel_hash, first_heard_at, last_heard_at, observation_count, (xmax = 0) AS inserted +` + +type UpsertPacketParams 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"` + ChannelHash []byte `json:"channel_hash"` +} + +type UpsertPacketRow 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"` + Inserted bool `json:"inserted"` +} + +// ============================================================ +// PACKETS +// ============================================================ +func (q *Queries) UpsertPacket(ctx context.Context, arg UpsertPacketParams) (UpsertPacketRow, error) { + row := q.db.QueryRow(ctx, upsertPacket, + arg.PacketHash, + arg.PayloadType, + arg.PayloadVersion, + arg.RouteType, + arg.TransportCodesPresent, + arg.RegionCode, + arg.SubRegionCode, + arg.OriginPubkey, + arg.RawPayload, + arg.ParsedPayload, + arg.ChannelHash, + ) + var i UpsertPacketRow + err := row.Scan( + &i.PacketHash, + &i.PayloadType, + &i.PayloadVersion, + &i.RouteType, + &i.TransportCodesPresent, + &i.RegionCode, + &i.SubRegionCode, + &i.OriginPubkey, + &i.RawPayload, + &i.ParsedPayload, + &i.Decrypted, + &i.ChannelHash, + &i.FirstHeardAt, + &i.LastHeardAt, + &i.ObservationCount, + &i.Inserted, + ) + return i, err +} diff --git a/db/store.go b/db/store.go new file mode 100644 index 0000000..ef7a0fb --- /dev/null +++ b/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 +} diff --git a/go.mod b/go.mod index 0944636..6f00a76 100644 --- a/go.mod +++ b/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 ) diff --git a/go.sum b/go.sum index cb95e96..899a080 100644 --- a/go.sum +++ b/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= diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index d38529e..664b7b3 100644 --- a/internal/ingest/ingest.go +++ b/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 } diff --git a/internal/keystore/keystore.go b/internal/keystore/keystore.go new file mode 100644 index 0000000..5e71dd7 --- /dev/null +++ b/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)] +} diff --git a/internal/ws/handler.go b/internal/ws/handler.go index 7f386b7..78bc33f 100644 --- a/internal/ws/handler.go +++ b/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 } diff --git a/sqlc.yaml b/sqlc.yaml index 6be8b1a..f405d3c 100644 --- a/sqlc.yaml +++ b/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"