From b51ced8655c7dc7176c8dcc58ea9f0e4a6c34c99 Mon Sep 17 00:00:00 2001 From: Kpa-clawbot <259247574+Kpa-clawbot@users.noreply.github.com> Date: Mon, 30 Mar 2026 20:57:45 -0700 Subject: [PATCH] Wire channel region filtering end-to-end Pass region through channel message routes, apply DB/store filtering, normalize IATA at read and write boundaries, and add regression coverage for routes/server/ingestor. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- cmd/ingestor/db.go | 6 ++-- cmd/ingestor/db_test.go | 2 +- cmd/server/db.go | 70 ++++++++++++++++++++++++++++++++++----- cmd/server/db_test.go | 55 ++++++++++++++++++++++++++++++ cmd/server/routes.go | 5 +-- cmd/server/routes_test.go | 47 ++++++++++++++++++++++++++ cmd/server/store.go | 22 ++++++++++-- 7 files changed, 191 insertions(+), 16 deletions(-) diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index 9dac68c8..d09747ab 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -7,6 +7,7 @@ import ( "log" "os" "path/filepath" + "strings" "sync/atomic" "time" @@ -501,6 +502,7 @@ type ObserverMeta struct { // UpsertObserver inserts or updates an observer with optional hardware metadata. func (s *Store) UpsertObserver(id, name, iata string, meta *ObserverMeta) error { now := time.Now().UTC().Format(time.RFC3339) + normalizedIATA := strings.TrimSpace(strings.ToUpper(iata)) var model, firmware, clientVersion, radio interface{} var batteryMv, uptimeSecs, noiseFloor interface{} @@ -529,8 +531,8 @@ func (s *Store) UpsertObserver(id, name, iata string, meta *ObserverMeta) error } _, err := s.stmtUpsertObserver.Exec( - id, name, iata, now, now, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, - name, iata, now, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, + id, name, normalizedIATA, now, now, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, + name, normalizedIATA, now, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, ) if err != nil { s.Stats.WriteErrors.Add(1) diff --git a/cmd/ingestor/db_test.go b/cmd/ingestor/db_test.go index 45c34a65..158ea820 100644 --- a/cmd/ingestor/db_test.go +++ b/cmd/ingestor/db_test.go @@ -219,7 +219,7 @@ func TestUpsertObserver(t *testing.T) { } defer s.Close() - if err := s.UpsertObserver("obs1", "Observer1", "SJC", nil); err != nil { + if err := s.UpsertObserver("obs1", "Observer1", " sjc ", nil); err != nil { t.Fatal(err) } diff --git a/cmd/server/db.go b/cmd/server/db.go index 0214e31a..11948103 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -951,17 +951,17 @@ func (db *DB) GetObserverByID(id string) (*Observer, error) { // GetObserverIdsForRegion returns observer IDs for given IATA codes. func (db *DB) GetObserverIdsForRegion(regionParam string) ([]string, error) { - if regionParam == "" { + codes := normalizeRegionCodes(regionParam) + if len(codes) == 0 { return nil, nil } - codes := strings.Split(regionParam, ",") placeholders := make([]string, len(codes)) args := make([]interface{}, len(codes)) for i, c := range codes { placeholders[i] = "?" - args[i] = strings.TrimSpace(c) + args[i] = c } - rows, err := db.conn.Query(fmt.Sprintf("SELECT id FROM observers WHERE iata IN (%s)", strings.Join(placeholders, ",")), args...) + rows, err := db.conn.Query(fmt.Sprintf("SELECT id FROM observers WHERE UPPER(TRIM(iata)) IN (%s)", strings.Join(placeholders, ",")), args...) if err != nil { return nil, err } @@ -969,12 +969,32 @@ func (db *DB) GetObserverIdsForRegion(regionParam string) ([]string, error) { var ids []string for rows.Next() { var id string - rows.Scan(&id) + if err := rows.Scan(&id); err != nil { + return nil, err + } ids = append(ids, id) } + if err := rows.Err(); err != nil { + return nil, err + } return ids, nil } +func normalizeRegionCodes(regionParam string) []string { + if regionParam == "" { + return nil + } + tokens := strings.Split(regionParam, ",") + codes := make([]string, 0, len(tokens)) + for _, token := range tokens { + code := strings.TrimSpace(strings.ToUpper(token)) + if code != "" { + codes = append(codes, code) + } + } + return codes +} + // GetDistinctIATAs returns all distinct IATA codes from observers. func (db *DB) GetDistinctIATAs() ([]string, error) { rows, err := db.conn.Query("SELECT DISTINCT iata FROM observers WHERE iata IS NOT NULL") @@ -1159,30 +1179,62 @@ func (db *DB) GetChannels() ([]map[string]interface{}, error) { // GetChannelMessages returns messages for a specific channel. // Uses transmission-level ordering (first_seen) to ensure correct message // sequence even when observations arrive out of order. -func (db *DB) GetChannelMessages(channelHash string, limit, offset int) ([]map[string]interface{}, int, error) { +func (db *DB) GetChannelMessages(channelHash string, limit, offset int, region ...string) ([]map[string]interface{}, int, error) { if limit <= 0 { limit = 100 } + regionParam := "" + if len(region) > 0 { + regionParam = region[0] + } + regionCodes := normalizeRegionCodes(regionParam) + regionArgs := make([]interface{}, 0, len(regionCodes)) + regionPlaceholders := "" + if len(regionCodes) > 0 { + placeholders := make([]string, len(regionCodes)) + for i, code := range regionCodes { + placeholders[i] = "?" + regionArgs = append(regionArgs, code) + } + regionPlaceholders = strings.Join(placeholders, ",") + } + var querySQL string + args := make([]interface{}, 0, len(regionArgs)) if db.isV3 { querySQL = `SELECT o.id, t.hash, t.decoded_json, t.first_seen, obs.id, obs.name, o.snr, o.path_json FROM observations o JOIN transmissions t ON t.id = o.transmission_id LEFT JOIN observers obs ON obs.rowid = o.observer_idx - WHERE t.payload_type = 5 + WHERE t.payload_type = 5` + if len(regionCodes) > 0 { + querySQL += fmt.Sprintf(" AND obs.rowid IS NOT NULL AND UPPER(TRIM(obs.iata)) IN (%s)", regionPlaceholders) + args = append(args, regionArgs...) + } + querySQL += ` ORDER BY t.first_seen ASC` } else { querySQL = `SELECT o.id, t.hash, t.decoded_json, t.first_seen, o.observer_id, o.observer_name, o.snr, o.path_json FROM observations o JOIN transmissions t ON t.id = o.transmission_id - WHERE t.payload_type = 5 + WHERE t.payload_type = 5` + if len(regionCodes) > 0 { + querySQL += fmt.Sprintf(` AND EXISTS ( + SELECT 1 + FROM observers obs + WHERE obs.id = o.observer_id + AND UPPER(TRIM(obs.iata)) IN (%s) + )`, regionPlaceholders) + args = append(args, regionArgs...) + } + querySQL += ` ORDER BY t.first_seen ASC` } - rows, err := db.conn.Query(querySQL) + rows, err := db.conn.Query(querySQL, args...) if err != nil { return nil, 0, err } diff --git a/cmd/server/db_test.go b/cmd/server/db_test.go index 3320ca7c..fef3f80b 100644 --- a/cmd/server/db_test.go +++ b/cmd/server/db_test.go @@ -631,6 +631,18 @@ func TestGetObserverIdsForRegion(t *testing.T) { } }) + t.Run("case and trim normalization", func(t *testing.T) { + db.conn.Exec(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) + VALUES ('obs3', 'Observer Three', ' sjc ', ?, '2026-01-01T00:00:00Z', 1)`, time.Now().UTC().Format(time.RFC3339)) + ids, err := db.GetObserverIdsForRegion(" sjc ") + if err != nil { + t.Fatal(err) + } + if len(ids) != 2 { + t.Errorf("expected 2 observers for normalized sjc, got %d", len(ids)) + } + }) + t.Run("empty param", func(t *testing.T) { ids, err := db.GetObserverIdsForRegion("") if err != nil { @@ -694,6 +706,49 @@ func TestGetChannelMessages(t *testing.T) { }) } +func TestGetChannelMessagesRegionFiltering(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + + now := time.Now().UTC() + ts1 := now.Add(-2 * time.Minute).Format(time.RFC3339) + ts2 := now.Add(-1 * time.Minute).Format(time.RFC3339) + epoch1 := now.Add(-2 * time.Minute).Unix() + epoch2 := now.Add(-1 * time.Minute).Unix() + + db.conn.Exec(`INSERT INTO observers (id, name, iata) VALUES ('obs1', 'Observer One', 'SJC')`) + db.conn.Exec(`INSERT INTO observers (id, name, iata) VALUES ('obs2', 'Observer Two', ' sfo ')`) + db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES ('AA', 'chanregion0001', ?, 1, 5, + '{"type":"CHAN","channel":"#region","text":"SjcUser: One","sender":"SjcUser"}')`, ts1) + db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES ('BB', 'chanregion0002', ?, 1, 5, + '{"type":"CHAN","channel":"#region","text":"SfoUser: Two","sender":"SfoUser"}')`, ts2) + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (1, 1, 10.0, -90, '[]', ?)`, epoch1) + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (2, 2, 9.0, -91, '[]', ?)`, epoch2) + + msgsSJC, totalSJC, err := db.GetChannelMessages("#region", 100, 0, " sjc ") + if err != nil { + t.Fatal(err) + } + if totalSJC != 1 || len(msgsSJC) != 1 { + t.Fatalf("expected 1 SJC message, total=%d len=%d", totalSJC, len(msgsSJC)) + } + if msgsSJC[0]["sender"] != "SjcUser" { + t.Fatalf("expected SJC sender SjcUser, got %v", msgsSJC[0]["sender"]) + } + + msgsMulti, totalMulti, err := db.GetChannelMessages("#region", 100, 0, "sjc, SFO") + if err != nil { + t.Fatal(err) + } + if totalMulti != 2 || len(msgsMulti) != 2 { + t.Fatalf("expected 2 multi-region messages, total=%d len=%d", totalMulti, len(msgsMulti)) + } +} + func TestBuildPacketWhereFilters(t *testing.T) { db := setupTestDB(t) defer db.Close() diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 61ca3194..a595a365 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -1308,12 +1308,13 @@ func (s *Server) handleChannelMessages(w http.ResponseWriter, r *http.Request) { hash := mux.Vars(r)["hash"] limit := queryInt(r, "limit", 100) offset := queryInt(r, "offset", 0) + region := r.URL.Query().Get("region") if s.store != nil { - messages, total := s.store.GetChannelMessages(hash, limit, offset) + messages, total := s.store.GetChannelMessages(hash, limit, offset, region) writeJSON(w, ChannelMessagesResponse{Messages: messages, Total: total}) return } - messages, total, err := s.db.GetChannelMessages(hash, limit, offset) + messages, total, err := s.db.GetChannelMessages(hash, limit, offset, region) if err != nil { writeError(w, 500, err.Error()) return diff --git a/cmd/server/routes_test.go b/cmd/server/routes_test.go index 9a584e93..7b8a474a 100644 --- a/cmd/server/routes_test.go +++ b/cmd/server/routes_test.go @@ -7,6 +7,7 @@ import ( "net/http/httptest" "strconv" "testing" + "time" "github.com/gorilla/mux" ) @@ -949,6 +950,52 @@ func TestChannelMessages(t *testing.T) { } } +func TestChannelMessagesWithRegion(t *testing.T) { + db := setupTestDB(t) + seedTestData(t, db) + db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES ('EEFF', 'chanextra000001', ?, 1, 5, '{"type":"CHAN","channel":"#test","text":"OtherUser: Cross region","sender":"OtherUser"}')`, + time.Now().UTC().Add(-30*time.Minute).Format(time.RFC3339)) + db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (4, 2, 11.0, -89, '[]', ?)`, time.Now().UTC().Add(-30*time.Minute).Unix()) + + cfg := &Config{Port: 3000} + hub := NewHub() + srv := NewServer(db, cfg, hub) + store := NewPacketStore(db, nil) + if err := store.Load(); err != nil { + t.Fatalf("store.Load failed: %v", err) + } + srv.store = store + router := mux.NewRouter() + srv.RegisterRoutes(router) + + req := httptest.NewRequest("GET", "/api/channels/%23test/messages?region=SJC", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + if w.Code != 200 { + t.Fatalf("expected 200, got %d", w.Code) + } + + var body map[string]interface{} + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("failed to parse response: %v", err) + } + msgs, ok := body["messages"].([]interface{}) + if !ok { + t.Fatalf("expected messages array, got %T", body["messages"]) + } + if len(msgs) == 0 { + t.Fatalf("expected at least one regional message") + } + for _, raw := range msgs { + msg, _ := raw.(map[string]interface{}) + if msg["sender"] == "OtherUser" { + t.Fatalf("cross-region message should be excluded") + } + } +} + func TestAnalyticsRFWithRegion(t *testing.T) { _, router := setupTestServer(t) req := httptest.NewRequest("GET", "/api/analytics/rf?region=SJC", nil) diff --git a/cmd/server/store.go b/cmd/server/store.go index 34d2333c..c9bab227 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -2144,7 +2144,7 @@ func (s *PacketStore) GetChannels(region string) []map[string]interface{} { } // GetChannelMessages returns deduplicated messages for a channel from in-memory packets. -func (s *PacketStore) GetChannelMessages(channelHash string, limit, offset int) ([]map[string]interface{}, int) { +func (s *PacketStore) GetChannelMessages(channelHash string, limit, offset int, region ...string) ([]map[string]interface{}, int) { s.mu.RLock() defer s.mu.RUnlock() @@ -2159,6 +2159,11 @@ func (s *PacketStore) GetChannelMessages(channelHash string, limit, offset int) } msgMap := map[string]*msgEntry{} var msgOrder []string + regionParam := "" + if len(region) > 0 { + regionParam = region[0] + } + regionObs := s.resolveRegionObservers(regionParam) // Iterate type-5 packets oldest-first (byPayloadType is ASC = oldest first) type decodedMsg struct { @@ -2172,6 +2177,19 @@ func (s *PacketStore) GetChannelMessages(channelHash string, limit, offset int) grpTxts := s.byPayloadType[5] for _, tx := range grpTxts { + if regionObs != nil { + match := false + for _, obs := range tx.Observations { + if regionObs[obs.ObserverID] { + match = true + break + } + } + if !match { + continue + } + } + if tx.DecodedJSON == "" { continue } @@ -3992,7 +4010,7 @@ func (s *PacketStore) computeAnalyticsHashSizes(region string) map[string]interf } return map[string]interface{}{ - "total": total, + "total": total, "distribution": distribution, "distributionByRepeaters": distributionByRepeaters, "hourly": hourly,