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

@@ -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)

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)
 	}
 

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
 	}

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()

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

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)

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,