add iata ingest filter

generated code, reviewed and tested

this adds an optional lookup table for allowed IATA based on either country,
conintent or both. if the IATA is in either it is allowed.

gives operators the option to consume from global mqtt brokers but filter
to their area of interest

README.md

@@ -78,6 +78,7 @@ tower-server/
 │   │   ├── status.go       status message handling
 │   │   ├── side_effects.go payload-type side effects (node upsert, channel messages)
 │   │   └── capability.go   firmware capability detection
+│   ├── iatadb/             static IATA → country/continent map (generated)
 │   ├── keystore/           channel key store
 │   ├── scopestore/         transport scope key store
 │   └── ws/                 WebSocket handler and IP limiter
@@ -203,6 +204,15 @@ packets:
 # WebSocket settings.
 websocket:
   max_connections_per_ip: 5 # default: 5
+
+# Geographic ingest filter (optional).
+# Drop packets from observers outside the specified area.
+# Country codes are ISO 3166-1 alpha-2. Continent codes: AF AN AS EU NA OC SA.
+# If both are set an IATA passes if it matches either (OR semantics).
+# Omit entirely to accept all IATAs (default).
+ingest:
+  allow_countries: [CA, US] # only store packets from these countries
+  allow_continents: [NA] # or: accept all of North America
 ```
 
 IATAs are auto-created on first packet arrival. The config file adds display
@@ -399,6 +409,28 @@ For paginated responses use the generic page wrapper:
 //	@Success	200	{object}	api.Page[api.MyType]
 ```
 
+### Updating the IATA database
+
+Tower includes a static IATA → country/continent mapping compiled into the
+binary, generated from the [OurAirports](https://ourairports.com/data/) public
+dataset.
+
+To refresh it with the latest airport data:
+
+```bash
+rm internal/iatadb/gen/airports.csv
+go generate ./internal/iatadb/
+```
+
+This fetches a fresh `airports.csv` from OurAirports, saves it locally, and
+regenerates `internal/iatadb/db.go`. Commit both files.
+
+To use a local CSV instead (e.g. in a restricted network environment):
+
+```bash
+AIRPORTS_CSV=/path/to/airports.csv go run ./internal/iatadb/gen
+```
+
 ---
 
 ## Road Map

cmd/tower/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/MeshCore-Tower/tower-server/internal/api/router"
 	"github.com/MeshCore-Tower/tower-server/internal/config"
 	"github.com/MeshCore-Tower/tower-server/internal/hub"
+	"github.com/MeshCore-Tower/tower-server/internal/iatadb"
 	"github.com/MeshCore-Tower/tower-server/internal/ingest"
 	"github.com/MeshCore-Tower/tower-server/internal/keystore"
 	"github.com/MeshCore-Tower/tower-server/internal/scopestore"
@@ -165,6 +166,13 @@ func main() {
 
 	keys := keystore.NewMapKeyStore(entries)
 
+	// ── Build geographic ingest filter ───────────────────────────────────────────────────────────
+	allowedIATAs := iatadb.BuildAllowedSet(cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents)
+	if allowedIATAs != nil {
+		log.Printf("config: ingest filter active — %d allowed IATAs (countries=%v continents=%v)",
+			len(allowedIATAs), cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents)
+	}
+
 	broker1 := ingest.New(
 		ingest.Config{
 			BrokerName:          "mqtt1",
@@ -172,6 +180,7 @@ func main() {
 			Username:            mustEnv("MQTT_BROKER_1_USERNAME"),
 			Password:            mustEnv("MQTT_BROKER_1_PASSWORD"),
 			TelemetryResolution: telemetryResolution,
+			AllowedIATAs:        allowedIATAs,
 		},
 		store,
 		h,
@@ -186,6 +195,7 @@ func main() {
 			Username:            mustEnv("MQTT_BROKER_2_USERNAME"),
 			Password:            mustEnv("MQTT_BROKER_2_PASSWORD"),
 			TelemetryResolution: telemetryResolution,
+			AllowedIATAs:        allowedIATAs,
 		},
 		store,
 		h,

config.yaml.example

@@ -75,3 +75,14 @@ packets:
 websocket:
   max_connections_per_ip: 5 # default: 5
 
+# Geographic ingest filter (optional).
+# Drop packets from observers outside the specified area at ingest time.
+# Country codes are ISO 3166-1 alpha-2 (e.g. CA, US, GB).
+# Continent codes: AF (Africa), AN (Antarctica), AS (Asia),
+#                  EU (Europe), NA (North America), OC (Oceania), SA (South America).
+# If both are set an IATA passes if it matches either (OR semantics).
+# Omit this section entirely to accept packets from all IATAs (default).
+#ingest:
+#  allow_countries: [CA, US]
+#  allow_continents: [NA]
+

internal/config/config.go

@@ -17,6 +17,7 @@ type Config struct {
 	Telemetry   TelemetryConfig       `yaml:"telemetry"`
 	WebSocket   WebSocketConfig       `yaml:"websocket"`
 	Packets     PacketsConfig         `yaml:"packets"`
+	Ingest      IngestFilterConfig    `yaml:"ingest"`
 	Scopes      []ScopeConfig         `yaml:"scopes"`
 }
 
@@ -111,6 +112,24 @@ type RegionConfig struct {
 	IATAs        []string `yaml:"iatas"`
 }
 
+
+// IngestFilterConfig restricts which packets Tower stores based on the
+// observer's IATA geographic location. Both filters are optional — if neither
+// is set all IATAs are accepted. If both are set an IATA passes if it matches
+// either (OR semantics).
+//
+// Country codes are ISO 3166-1 alpha-2 (e.g. "CA", "US").
+// Continent codes are two-letter OurAirports codes: AF, AN, AS, EU, NA, OC, SA.
+type IngestFilterConfig struct {
+	// AllowCountries is a list of ISO 3166-1 alpha-2 country codes to accept.
+	// Packets from observers in other countries are dropped at ingest.
+	AllowCountries []string `yaml:"allow_countries"`
+
+	// AllowContinents is a list of continent codes to accept.
+	// Packets from observers in other continents are dropped at ingest.
+	AllowContinents []string `yaml:"allow_continents"`
+}
+
 // Load reads and parses the config file at path.
 // Returns an empty Config (not an error) if the file does not exist,
 // so Tower starts cleanly without a config file.

internal/iatadb/gen/main.go

@@ -0,0 +1,189 @@
+// gen generates internal/iatadb/db.go from the OurAirports airports.csv dataset.
+//
+// Usage (from repo root):
+//
+//	go run ./internal/iatadb/gen
+//
+// The source CSV is read from internal/iatadb/gen/airports.csv by default.
+// To use a different file, set the AIRPORTS_CSV environment variable.
+// To fetch a fresh copy from OurAirports, delete airports.csv first:
+//
+//	rm internal/iatadb/gen/airports.csv && go run ./internal/iatadb/gen
+//
+// The upstream dataset is published at:
+//
+//	https://raw.githubusercontent.com/davidmegginson/ourairports-data/main/airports.csv
+package main
+
+import (
+	"encoding/csv"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"sort"
+	"strings"
+	"text/template"
+	"time"
+)
+
+const (
+	sourceURL  = "https://raw.githubusercontent.com/davidmegginson/ourairports-data/main/airports.csv"
+	localCSV   = "internal/iatadb/gen/airports.csv"
+	outputPath = "internal/iatadb/db.go"
+
+	colContinent = 7
+	colCountry   = 8
+	colIATACode  = 13
+)
+
+type entry struct {
+	IATA      string
+	Country   string
+	Continent string
+}
+
+func main() {
+	// ensure we run from the repo root
+	if _, err := os.Stat("go.mod"); err != nil {
+		fmt.Fprintf(os.Stderr, "error: run from repo root (go.mod not found)\n")
+		os.Exit(1)
+	}
+
+	var r io.Reader
+
+	csvPath := os.Getenv("AIRPORTS_CSV")
+	if csvPath == "" {
+		csvPath = localCSV
+	}
+
+	if _, err := os.Stat(csvPath); err == nil {
+		f, err := os.Open(csvPath)
+		if err != nil {
+			log.Fatalf("open %s: %v", csvPath, err)
+		}
+		defer f.Close()
+		r = f
+		log.Printf("gen: reading from %s", csvPath)
+	} else {
+		log.Printf("gen: %s not found, fetching from %s", csvPath, sourceURL)
+		client := &http.Client{Timeout: 30 * time.Second}
+		resp, err := client.Get(sourceURL)
+		if err != nil {
+			log.Fatalf("fetch: %v", err)
+		}
+		defer resp.Body.Close()
+		if resp.StatusCode != http.StatusOK {
+			log.Fatalf("fetch: HTTP %d", resp.StatusCode)
+		}
+		// save locally for future runs
+		if err := saveLocal(resp.Body, csvPath); err != nil {
+			log.Printf("gen: warning: could not save local copy: %v", err)
+		}
+		f, err := os.Open(csvPath)
+		if err != nil {
+			log.Fatalf("re-open %s: %v", csvPath, err)
+		}
+		defer f.Close()
+		r = f
+	}
+
+	cr := csv.NewReader(r)
+	cr.LazyQuotes = true
+
+	if _, err := cr.Read(); err != nil { // skip header
+		log.Fatalf("read header: %v", err)
+	}
+
+	var entries []entry
+	for {
+		row, err := cr.Read()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			log.Printf("gen: skipping malformed row: %v", err)
+			continue
+		}
+		if len(row) <= colIATACode {
+			continue
+		}
+		iata := strings.TrimSpace(row[colIATACode])
+		if len(iata) != 3 {
+			continue
+		}
+		country := strings.TrimSpace(row[colCountry])
+		continent := strings.TrimSpace(row[colContinent])
+		if country == "" || continent == "" {
+			continue
+		}
+		entries = append(entries, entry{IATA: iata, Country: country, Continent: continent})
+	}
+
+	sort.Slice(entries, func(i, j int) bool { return entries[i].IATA < entries[j].IATA })
+
+	// deduplicate
+	seen := make(map[string]struct{})
+	unique := entries[:0]
+	for _, e := range entries {
+		if _, ok := seen[e.IATA]; !ok {
+			seen[e.IATA] = struct{}{}
+			unique = append(unique, e)
+		}
+	}
+	entries = unique
+
+	log.Printf("gen: %d valid IATA codes", len(entries))
+
+	out, err := os.Create(outputPath)
+	if err != nil {
+		log.Fatalf("create %s: %v", outputPath, err)
+	}
+	defer out.Close()
+
+	if err := tmpl.Execute(out, struct {
+		Source  string
+		Entries []entry
+	}{Source: sourceURL, Entries: entries}); err != nil {
+		log.Fatalf("render: %v", err)
+	}
+	log.Printf("gen: wrote %s", outputPath)
+}
+
+func saveLocal(r io.Reader, path string) error {
+	data, err := io.ReadAll(r)
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(path, data, 0644)
+}
+
+var tmpl = template.Must(template.New("").Parse(`// Code generated by internal/iatadb/gen; DO NOT EDIT.
+// Source: {{ .Source }}
+// To regenerate: go run ./internal/iatadb/gen
+
+package iatadb
+
+// Entry holds geographic metadata for a single IATA airport code.
+type Entry struct {
+	Country   string // ISO 3166-1 alpha-2 country code e.g. "CA"
+	Continent string // two-letter continent code: AF, AN, AS, EU, NA, OC, SA
+}
+
+// DB maps 3-letter IATA codes to geographic metadata.
+// Generated from the OurAirports dataset; run go generate to refresh.
+var DB = map[string]Entry{
+{{ range .Entries }}	"{{ .IATA }}": {Country: "{{ .Country }}", Continent: "{{ .Continent }}"},
+{{ end }}}
+
+// CountryContinent maps ISO 3166-1 alpha-2 country codes to continent codes.
+// Derived from the same OurAirports dataset as DB.
+var CountryContinent = func() map[string]string {
+	m := make(map[string]string, 250)
+	for _, e := range DB {
+		m[e.Country] = e.Continent
+	}
+	return m
+}()
+`))

internal/iatadb/iatadb.go

@@ -0,0 +1,58 @@
+// Package iatadb provides a static mapping from IATA airport codes to
+// geographic metadata (country and continent).
+//
+// The data is generated from the OurAirports public dataset and compiled
+// into the binary — no external calls at runtime.
+//
+// To refresh the dataset:
+//
+//	go generate ./internal/iatadb/
+//
+//go:generate go run ./gen
+package iatadb
+
+// Lookup returns the Entry for the given IATA code, and whether it was found.
+func Lookup(iata string) (Entry, bool) {
+	e, ok := DB[iata]
+	return e, ok
+}
+
+// CountryFor returns the ISO 3166-1 alpha-2 country code for the given IATA,
+// or empty string if not found.
+func CountryFor(iata string) string {
+	return DB[iata].Country
+}
+
+// ContinentFor returns the two-letter continent code for the given IATA,
+// or empty string if not found.
+func ContinentFor(iata string) string {
+	return DB[iata].Continent
+}
+
+// BuildAllowedSet returns a set of IATA codes permitted by the given country
+// and continent allowlists. If both slices are empty, returns nil (no filter).
+// An IATA passes if it matches any entry in either list (OR semantics).
+func BuildAllowedSet(allowCountries, allowContinents []string) map[string]struct{} {
+	if len(allowCountries) == 0 && len(allowContinents) == 0 {
+		return nil
+	}
+	countrySet := make(map[string]struct{}, len(allowCountries))
+	for _, c := range allowCountries {
+		countrySet[c] = struct{}{}
+	}
+	continentSet := make(map[string]struct{}, len(allowContinents))
+	for _, c := range allowContinents {
+		continentSet[c] = struct{}{}
+	}
+	allowed := make(map[string]struct{})
+	for iata, entry := range DB {
+		if _, ok := countrySet[entry.Country]; ok {
+			allowed[iata] = struct{}{}
+			continue
+		}
+		if _, ok := continentSet[entry.Continent]; ok {
+			allowed[iata] = struct{}{}
+		}
+	}
+	return allowed
+}

internal/ingest/ingest.go

@@ -51,6 +51,11 @@ type Config struct {
 	Username string
 	Password string
 
+	// AllowedIATAs is a pre-computed set of IATA codes derived from the ingest
+	// filter config. If non-nil, packets from IATAs not in this set are dropped.
+	// Build this set at startup from IngestFilterConfig using iatadb.
+	AllowedIATAs map[string]struct{}
+
 	// TelemetryResolution controls how frequently telemetry snapshots are stored.
 	// Status messages within the same window are deduplicated via ON CONFLICT.
 	// Defaults to 1 hour if zero.
@@ -236,6 +241,14 @@ func (w *Worker) handleMessage(msg mqtt.Message) {
 	}
 	iata, pubkeyHex, subtopic := parts[1], parts[2], parts[3]
 
+	// Drop packets from IATAs outside the configured geographic filter.
+	if w.cfg.AllowedIATAs != nil {
+		if _, ok := w.cfg.AllowedIATAs[iata]; !ok {
+			log.Printf("ingest[%s]: dropped packet from %s (not in allowed IATAs)", w.cfg.BrokerName, iata)
+			return
+		}
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()