meshcore-analyzer/cmd/ingestor/main.go

package main

import (
	"crypto/sha256"
	"crypto/tls"
	"encoding/hex"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"math"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"path/filepath"
	"strconv"
	"strings"
	"syscall"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
	// pprof profiling — off by default, enable with ENABLE_PPROF=true
	if os.Getenv("ENABLE_PPROF") == "true" {
		pprofPort := os.Getenv("PPROF_PORT")
		if pprofPort == "" {
			pprofPort = "6061"
		}
		go func() {
			log.Printf("[pprof] ingestor profiling at http://localhost:%s/debug/pprof/", pprofPort)
			if err := http.ListenAndServe(":"+pprofPort, nil); err != nil {
				log.Printf("[pprof] failed to start: %v (non-fatal)", err)
			}
		}()
	}

	configPath := flag.String("config", "config.json", "path to config file")
	flag.Parse()

	log.SetFlags(log.LstdFlags | log.Lmsgprefix)
	log.SetPrefix("[ingestor] ")

	cfg, err := LoadConfig(*configPath)
	if err != nil {
		log.Fatalf("config: %v", err)
	}

	sources := cfg.ResolvedSources()

	store, err := OpenStoreWithInterval(cfg.DBPath, cfg.MetricsSampleInterval())
	if err != nil {
		log.Fatalf("db: %v", err)
	}
	defer store.Close()
	log.Printf("SQLite opened: %s", cfg.DBPath)

	// Async backfill: path_json from raw_hex (#888) — must not block MQTT startup
	store.BackfillPathJSONAsync()

	// Check auto_vacuum mode and optionally migrate (#919)
	store.CheckAutoVacuum(cfg)

	// Node retention: move stale nodes to inactive_nodes on startup
	nodeDays := cfg.NodeDaysOrDefault()
	store.MoveStaleNodes(nodeDays)

	// Observer retention: remove stale observers on startup
	observerDays := cfg.ObserverDaysOrDefault()
	store.RemoveStaleObservers(observerDays)

	// Metrics retention: prune old metrics on startup
	metricsDays := cfg.MetricsRetentionDays()
	store.PruneOldMetrics(metricsDays)
	store.PruneDroppedPackets(metricsDays)
	vacuumPages := cfg.IncrementalVacuumPages()
	store.RunIncrementalVacuum(vacuumPages)

	// Daily ticker for node retention
	retentionTicker := time.NewTicker(1 * time.Hour)
	go func() {
		for range retentionTicker.C {
			store.MoveStaleNodes(nodeDays)
			store.RunIncrementalVacuum(vacuumPages)
		}
	}()

	// Daily ticker for observer retention (every 24h, staggered 90s after startup)
	observerRetentionTicker := time.NewTicker(24 * time.Hour)
	go func() {
		time.Sleep(90 * time.Second) // stagger after metrics prune
		store.RemoveStaleObservers(observerDays)
		store.RunIncrementalVacuum(vacuumPages)
		for range observerRetentionTicker.C {
			store.RemoveStaleObservers(observerDays)
			store.RunIncrementalVacuum(vacuumPages)
		}
	}()

	// Daily ticker for metrics retention (every 24h)
	metricsRetentionTicker := time.NewTicker(24 * time.Hour)
	go func() {
		for range metricsRetentionTicker.C {
			store.PruneOldMetrics(metricsDays)
			store.PruneDroppedPackets(metricsDays)
			store.RunIncrementalVacuum(vacuumPages)
		}
	}()

	// Periodic stats logging (every 5 minutes)
	statsTicker := time.NewTicker(5 * time.Minute)
	go func() {
		for range statsTicker.C {
			store.LogStats()
		}
	}()

	channelKeys := loadChannelKeys(cfg, *configPath)
	if len(channelKeys) > 0 {
		log.Printf("Loaded %d channel keys for GRP_TXT decryption", len(channelKeys))
	} else {
		log.Printf("No channel keys loaded — GRP_TXT packets will not be decrypted")
	}

	// Connect to each MQTT source
	var clients []mqtt.Client
	connectedCount := 0
	for _, source := range sources {
		tag := source.Name
		if tag == "" {
			tag = source.Broker
		}

		opts := buildMQTTOpts(source)
		connectTimeout := source.ConnectTimeoutOrDefault()
		log.Printf("MQTT [%s] connect timeout: %ds", tag, connectTimeout)

		opts.SetOnConnectHandler(func(c mqtt.Client) {
			log.Printf("MQTT [%s] connected to %s", tag, source.Broker)
			topics := source.Topics
			if len(topics) == 0 {
				topics = []string{"meshcore/#"}
			}
			for _, t := range topics {
				token := c.Subscribe(t, 0, nil)
				token.Wait()
				if token.Error() != nil {
					log.Printf("MQTT [%s] subscribe error for %s: %v", tag, t, token.Error())
				} else {
					log.Printf("MQTT [%s] subscribed to %s", tag, t)
				}
			}
		})

		opts.SetConnectionLostHandler(func(c mqtt.Client, err error) {
			log.Printf("MQTT [%s] disconnected from %s: %v", tag, source.Broker, err)
		})

		opts.SetReconnectingHandler(func(c mqtt.Client, options *mqtt.ClientOptions) {
			log.Printf("MQTT [%s] reconnecting to %s", tag, source.Broker)
		})

		// Capture source for closure
		src := source
		opts.SetDefaultPublishHandler(func(c mqtt.Client, m mqtt.Message) {
			handleMessage(store, tag, src, m, channelKeys, cfg)
		})

		client := mqtt.NewClient(opts)
		token := client.Connect()
		// With ConnectRetry=true, token.Wait() blocks forever for unreachable brokers.
		// WaitTimeout lets startup proceed; the client keeps retrying in the background
		// and OnConnect fires (subscribing) when it eventually connects (#910).
		if !token.WaitTimeout(time.Duration(connectTimeout) * time.Second) {
			log.Printf("MQTT [%s] initial connection timed out — retrying in background", tag)
			clients = append(clients, client)
			continue
		}
		if token.Error() != nil {
			log.Printf("MQTT [%s] connection failed (non-fatal): %v", tag, token.Error())
			// BL1 fix: Disconnect to stop Paho's internal retry goroutines.
			// With ConnectRetry=true, Connect() spawns background goroutines
			// that leak if the client is simply discarded.
			client.Disconnect(0)
			continue
		}
		connectedCount++
		clients = append(clients, client)
	}

	// BL2 fix: require at least one immediately-connected source. Timed-out
	// clients are retrying in background (tracked in clients) but don't count
	// as "connected" — a single unreachable broker must not silently run with
	// zero active connections.
	if connectedCount == 0 {
		// Clean up any timed-out clients still retrying
		for _, c := range clients {
			c.Disconnect(0)
		}
		log.Fatal("no MQTT sources connected — all timed out or failed. Check broker is running (default: mqtt://localhost:1883). Set MQTT_BROKER env var or configure mqttSources in config.json")
	}

	if connectedCount < len(clients) {
		log.Printf("Running — %d MQTT source(s) connected, %d retrying in background", connectedCount, len(clients)-connectedCount)
	} else {
		log.Printf("Running — %d MQTT source(s) connected", connectedCount)
	}

	// Wait for shutdown signal
	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
	<-sig

	log.Println("Shutting down...")
	retentionTicker.Stop()
	metricsRetentionTicker.Stop()
	statsTicker.Stop()
	store.LogStats() // final stats on shutdown
	for _, c := range clients {
		c.Disconnect(5000) // 5s to allow in-flight messages to drain
	}
	log.Println("Done.")
}

// buildMQTTOpts creates MQTT client options for a source with bounded reconnect
// backoff, connect timeout, and TLS/auth configuration.
func buildMQTTOpts(source MQTTSource) *mqtt.ClientOptions {
	opts := mqtt.NewClientOptions().
		AddBroker(source.Broker).
		SetAutoReconnect(true).
		SetConnectRetry(true).
		SetOrderMatters(true).
		SetMaxReconnectInterval(30 * time.Second).
		SetConnectTimeout(10 * time.Second).
		SetWriteTimeout(10 * time.Second)

	if source.Username != "" {
		opts.SetUsername(source.Username)
	}
	if source.Password != "" {
		opts.SetPassword(source.Password)
	}
	if source.RejectUnauthorized != nil && !*source.RejectUnauthorized {
		opts.SetTLSConfig(&tls.Config{InsecureSkipVerify: true})
	} else if strings.HasPrefix(source.Broker, "ssl://") {
		opts.SetTLSConfig(&tls.Config{})
	}
	return opts
}

func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, channelKeys map[string]string, cfg *Config) {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("MQTT [%s] panic in handler: %v", tag, r)
		}
	}()

	topic := m.Topic()
	parts := strings.Split(topic, "/")

	var msg map[string]interface{}
	if err := json.Unmarshal(m.Payload(), &msg); err != nil {
		return
	}

	// Skip status/connection topics
	if topic == "meshcore/status" || topic == "meshcore/events/connection" {
		return
	}

	// Observer blacklist: drop ALL messages from blacklisted observers before any
	// DB writes (status, metrics, packets). Trumps IATA filter.
	if len(parts) > 2 && cfg.IsObserverBlacklisted(parts[2]) {
		log.Printf("MQTT [%s] observer %.8s blacklisted, dropping", tag, parts[2])
		return
	}

	// Global observer IATA whitelist: if configured, drop messages from observers
	// in non-whitelisted IATA regions. Applies to ALL message types (status + packets).
	if len(parts) > 1 && !cfg.IsObserverIATAAllowed(parts[1]) {
		return
	}

	// Status topic: meshcore/<region>/<observer_id>/status
	// Per-source IATA filter does NOT apply here — observer metadata (noise_floor, battery, etc.)
	// is region-independent and should be accepted from all observers regardless of
	// which IATA regions are configured for packet ingestion.
	if len(parts) >= 4 && parts[3] == "status" {
		observerID := parts[2]
		name, _ := msg["origin"].(string)
		iata := parts[1]
		meta := extractObserverMeta(msg)
		if err := store.UpsertObserver(observerID, name, iata, meta); err != nil {
			log.Printf("MQTT [%s] observer status error: %v", tag, err)
		}
		// Insert metrics sample from status message
		if meta != nil {
			metricsData := &MetricsData{
				ObserverID:  observerID,
				NoiseFloor:  meta.NoiseFloor,
				TxAirSecs:   meta.TxAirSecs,
				RxAirSecs:   meta.RxAirSecs,
				RecvErrors:  meta.RecvErrors,
				BatteryMv:   meta.BatteryMv,
				PacketsSent: meta.PacketsSent,
				PacketsRecv: meta.PacketsRecv,
			}
			if err := store.InsertMetrics(metricsData); err != nil {
				log.Printf("MQTT [%s] metrics insert error: %v", tag, err)
			}
		}
		log.Printf("MQTT [%s] status: %s (%s)", tag, firstNonEmpty(name, observerID), iata)
		return
	}

	// IATA filter applies to packet messages only — not status messages above.
	if len(source.IATAFilter) > 0 && len(parts) > 1 {
		region := parts[1]
		matched := false
		for _, f := range source.IATAFilter {
			if f == region {
				matched = true
				break
			}
		}
		if !matched {
			return
		}
	}

	// Format 1: Raw packet (meshcoretomqtt / Cisien format)
	rawHex, _ := msg["raw"].(string)
	if rawHex != "" {
		validateSigs := cfg.ShouldValidateSignatures()
		decoded, err := DecodePacket(rawHex, channelKeys, validateSigs)
		if err != nil {
			log.Printf("MQTT [%s] decode error: %v", tag, err)
			return
		}

		observerID := ""
		region := ""
		if len(parts) > 2 {
			observerID = parts[2]
		}
		if len(parts) > 1 {
			region = parts[1]
		}
		// Fallback to source-level region config when topic has no region (#788)
		if region == "" && source.Region != "" {
			region = source.Region
		}

		mqttMsg := &MQTTPacketMessage{Raw: rawHex}
		// Parse optional region from JSON payload (#788)
		if v, ok := msg["region"].(string); ok && v != "" {
			mqttMsg.Region = v
		}
		if v, ok := msg["SNR"]; ok {
			if f, ok := toFloat64(v); ok {
				mqttMsg.SNR = &f
			}
		} else if v, ok := msg["snr"]; ok {
			if f, ok := toFloat64(v); ok {
				mqttMsg.SNR = &f
			}
		}
		if v, ok := msg["RSSI"]; ok {
			if f, ok := toFloat64(v); ok {
				mqttMsg.RSSI = &f
			}
		} else if v, ok := msg["rssi"]; ok {
			if f, ok := toFloat64(v); ok {
				mqttMsg.RSSI = &f
			}
		}
		if v, ok := msg["score"]; ok {
			if f, ok := toFloat64(v); ok {
				mqttMsg.Score = &f
			}
		} else if v, ok := msg["Score"]; ok {
			if f, ok := toFloat64(v); ok {
				mqttMsg.Score = &f
			}
		}
		if v, ok := msg["direction"].(string); ok {
			mqttMsg.Direction = &v
		} else if v, ok := msg["Direction"].(string); ok {
			mqttMsg.Direction = &v
		}
		if v, ok := msg["origin"].(string); ok {
			mqttMsg.Origin = v
		}

		// For ADVERT packets with known coordinates, enforce geo_filter before
		// storing anything — drop the entire message if outside the area.
		if decoded.Header.PayloadTypeName == "ADVERT" && decoded.Payload.PubKey != "" {
			ok, reason := ValidateAdvert(&decoded.Payload)
			if !ok {
				log.Printf("MQTT [%s] skipping corrupted ADVERT: %s", tag, reason)
				return
			}
			// Signature validation: drop adverts with invalid ed25519 signatures
			if validateSigs && decoded.Payload.SignatureValid != nil && !*decoded.Payload.SignatureValid {
				hash := ComputeContentHash(rawHex)
				truncPK := decoded.Payload.PubKey
				if len(truncPK) > 16 {
					truncPK = truncPK[:16]
				}
				log.Printf("MQTT [%s] DROPPED invalid signature: hash=%s name=%s observer=%s pubkey=%s",
					tag, hash, decoded.Payload.Name, firstNonEmpty(mqttMsg.Origin, observerID), truncPK)
				store.InsertDroppedPacket(&DroppedPacket{
					Hash:         hash,
					RawHex:       rawHex,
					Reason:       "invalid signature",
					ObserverID:   observerID,
					ObserverName: mqttMsg.Origin,
					NodePubKey:   decoded.Payload.PubKey,
					NodeName:     decoded.Payload.Name,
				})
				return
			}
			foreign := false
			if !NodePassesGeoFilter(decoded.Payload.Lat, decoded.Payload.Lon, cfg.GeoFilter) {
				if cfg.ForeignAdverts.IsDropMode() {
					return
				}
				foreign = true
				lat, lon := 0.0, 0.0
				if decoded.Payload.Lat != nil {
					lat = *decoded.Payload.Lat
				}
				if decoded.Payload.Lon != nil {
					lon = *decoded.Payload.Lon
				}
				truncPK := decoded.Payload.PubKey
				if len(truncPK) > 16 {
					truncPK = truncPK[:16]
				}
				log.Printf("MQTT [%s] foreign advert: node=%s name=%s lat=%.4f lon=%.4f observer=%s",
					tag, truncPK, decoded.Payload.Name, lat, lon, firstNonEmpty(mqttMsg.Origin, observerID))
			}
			pktData := BuildPacketData(mqttMsg, decoded, observerID, region)
			pktData.Foreign = foreign
			isNew, err := store.InsertTransmission(pktData)
			if err != nil {
				log.Printf("MQTT [%s] db insert error: %v", tag, err)
			}
			role := advertRole(decoded.Payload.Flags)
			if err := store.UpsertNode(decoded.Payload.PubKey, decoded.Payload.Name, role, decoded.Payload.Lat, decoded.Payload.Lon, pktData.Timestamp); err != nil {
				log.Printf("MQTT [%s] node upsert error: %v", tag, err)
			}
			if foreign {
				if err := store.MarkNodeForeign(decoded.Payload.PubKey); err != nil {
					log.Printf("MQTT [%s] mark foreign error: %v", tag, err)
				}
			}
			if isNew {
				if err := store.IncrementAdvertCount(decoded.Payload.PubKey); err != nil {
					log.Printf("MQTT [%s] advert count error: %v", tag, err)
				}
			}
			// Update telemetry if present in advert
			if decoded.Payload.BatteryMv != nil || decoded.Payload.TemperatureC != nil {
				if err := store.UpdateNodeTelemetry(decoded.Payload.PubKey, decoded.Payload.BatteryMv, decoded.Payload.TemperatureC); err != nil {
					log.Printf("MQTT [%s] node telemetry update error: %v", tag, err)
				}
			}
		} else {
			// Non-ADVERT packets: store normally (routing/channel messages from
			// in-area observers are relevant regardless of relay hop origin).
			pktData := BuildPacketData(mqttMsg, decoded, observerID, region)
			if _, err := store.InsertTransmission(pktData); err != nil {
				log.Printf("MQTT [%s] db insert error: %v", tag, err)
			}
		}

		// Upsert observer
		if observerID != "" {
			origin, _ := msg["origin"].(string)
			// Use effective region: payload > topic > source config (#788)
			effectiveRegion := region
			if mqttMsg.Region != "" {
				effectiveRegion = mqttMsg.Region
			}
			if err := store.UpsertObserver(observerID, origin, effectiveRegion, nil); err != nil {
				log.Printf("MQTT [%s] observer upsert error: %v", tag, err)
			}
		}

		return
	}

	// Format 2: Companion bridge channel message (meshcore/message/channel/<n>)
	if strings.HasPrefix(topic, "meshcore/message/channel/") {
		text, _ := msg["text"].(string)
		if text == "" {
			return
		}

		channelIdx := ""
		if len(parts) >= 4 {
			channelIdx = parts[3]
		}
		if ci, ok := msg["channel_idx"]; ok {
			channelIdx = fmt.Sprintf("%v", ci)
		}

		// Extract sender from "Name: message" format
		sender := ""
		if idx := strings.Index(text, ": "); idx > 0 && idx < 50 {
			sender = text[:idx]
		}

		channelName := fmt.Sprintf("ch%s", channelIdx)

		// Build decoded JSON matching Node.js CHAN format
		channelMsg := map[string]interface{}{
			"type":    "CHAN",
			"channel": channelName,
			"text":    text,
			"sender":  sender,
		}
		if st, ok := msg["sender_timestamp"]; ok {
			channelMsg["sender_timestamp"] = st
		}

		decodedJSON, _ := json.Marshal(channelMsg)

		now := time.Now().UTC().Format(time.RFC3339)
		hashInput := fmt.Sprintf("ch:%s:%s:%s", channelIdx, text, now)
		h := sha256.Sum256([]byte(hashInput))
		hash := hex.EncodeToString(h[:])[:16]

		var snr, rssi, score *float64
		var direction *string
		if v, ok := msg["SNR"]; ok {
			if f, ok := toFloat64(v); ok {
				snr = &f
			}
		} else if v, ok := msg["snr"]; ok {
			if f, ok := toFloat64(v); ok {
				snr = &f
			}
		}
		if v, ok := msg["RSSI"]; ok {
			if f, ok := toFloat64(v); ok {
				rssi = &f
			}
		} else if v, ok := msg["rssi"]; ok {
			if f, ok := toFloat64(v); ok {
				rssi = &f
			}
		}
		if v, ok := msg["score"]; ok {
			if f, ok := toFloat64(v); ok {
				score = &f
			}
		} else if v, ok := msg["Score"]; ok {
			if f, ok := toFloat64(v); ok {
				score = &f
			}
		}
		if v, ok := msg["direction"].(string); ok {
			direction = &v
		} else if v, ok := msg["Direction"].(string); ok {
			direction = &v
		}

		pktData := &PacketData{
			Timestamp:    now,
			ObserverID:   "companion",
			ObserverName: "L1 Pro (BLE)",
			SNR:          snr,
			RSSI:         rssi,
			Score:        score,
			Direction:    direction,
			Hash:         hash,
			RouteType:    1, // FLOOD
			PayloadType:  5, // GRP_TXT
			PathJSON:     "[]",
			DecodedJSON:  string(decodedJSON),
			ChannelHash:  channelName, // fast channel queries (#762)
		}

		if _, err := store.InsertTransmission(pktData); err != nil {
			log.Printf("MQTT [%s] channel insert error: %v", tag, err)
		}

		// Note: we intentionally do NOT create a node entry for channel message senders.
		// Channel messages don't carry the sender's real pubkey, so any entry we create
		// would use a synthetic key ("sender-<name>") that doesn't match the real pubkey
		// used for claiming/health lookups. The node will get a proper entry when it
		// sends an advert. See issue #665.

		log.Printf("MQTT [%s] channel message: ch%s from %s", tag, channelIdx, firstNonEmpty(sender, "unknown"))
		return
	}

	// Format 2b: Companion bridge direct message (meshcore/message/direct/<id>)
	if strings.HasPrefix(topic, "meshcore/message/direct/") {
		text, _ := msg["text"].(string)
		if text == "" {
			return
		}

		sender := ""
		if idx := strings.Index(text, ": "); idx > 0 && idx < 50 {
			sender = text[:idx]
		}

		dm := map[string]interface{}{
			"type":   "DM",
			"text":   text,
			"sender": sender,
		}
		if st, ok := msg["sender_timestamp"]; ok {
			dm["sender_timestamp"] = st
		}

		decodedJSON, _ := json.Marshal(dm)

		now := time.Now().UTC().Format(time.RFC3339)
		hashInput := fmt.Sprintf("dm:%s:%s", text, now)
		h := sha256.Sum256([]byte(hashInput))
		hash := hex.EncodeToString(h[:])[:16]

		var snr, rssi, score *float64
		var direction *string
		if v, ok := msg["SNR"]; ok {
			if f, ok := toFloat64(v); ok {
				snr = &f
			}
		} else if v, ok := msg["snr"]; ok {
			if f, ok := toFloat64(v); ok {
				snr = &f
			}
		}
		if v, ok := msg["RSSI"]; ok {
			if f, ok := toFloat64(v); ok {
				rssi = &f
			}
		} else if v, ok := msg["rssi"]; ok {
			if f, ok := toFloat64(v); ok {
				rssi = &f
			}
		}
		if v, ok := msg["score"]; ok {
			if f, ok := toFloat64(v); ok {
				score = &f
			}
		} else if v, ok := msg["Score"]; ok {
			if f, ok := toFloat64(v); ok {
				score = &f
			}
		}
		if v, ok := msg["direction"].(string); ok {
			direction = &v
		} else if v, ok := msg["Direction"].(string); ok {
			direction = &v
		}

		pktData := &PacketData{
			Timestamp:    now,
			ObserverID:   "companion",
			ObserverName: "L1 Pro (BLE)",
			SNR:          snr,
			RSSI:         rssi,
			Score:        score,
			Direction:    direction,
			Hash:         hash,
			RouteType:    1, // FLOOD
			PayloadType:  2, // TXT_MSG
			PathJSON:     "[]",
			DecodedJSON:  string(decodedJSON),
		}

		if _, err := store.InsertTransmission(pktData); err != nil {
			log.Printf("MQTT [%s] DM insert error: %v", tag, err)
		}

		log.Printf("MQTT [%s] direct message from %s", tag, firstNonEmpty(sender, "unknown"))
		return
	}
}

func toFloat64(v interface{}) (float64, bool) {
	switch n := v.(type) {
	case float64:
		return n, true
	case float32:
		return float64(n), true
	case int:
		return float64(n), true
	case int64:
		return float64(n), true
	case json.Number:
		f, err := n.Float64()
		return f, err == nil
	case string:
		s := strings.TrimSpace(n)
		s = stripUnitSuffix(s)
		f, err := strconv.ParseFloat(s, 64)
		return f, err == nil
	case uint:
		return float64(n), true
	case uint64:
		return float64(n), true
	default:
		return 0, false
	}
}

// unitSuffixes lists common RF/signal unit suffixes to strip before parsing.
var unitSuffixes = []string{"dBm", "dB", "mW", "km", "mi", "m"}

// stripUnitSuffix removes a trailing unit suffix (case-insensitive) from a
// numeric string so that values like "-110dBm" can be parsed as float64.
func stripUnitSuffix(s string) string {
	lower := strings.ToLower(s)
	for _, suffix := range unitSuffixes {
		if strings.HasSuffix(lower, strings.ToLower(suffix)) {
			return strings.TrimSpace(s[:len(s)-len(suffix)])
		}
	}
	return s
}

// extractObserverMeta extracts hardware metadata from an MQTT status message.
// Casts battery_mv and uptime_secs to integers (they're always whole numbers).
func extractObserverMeta(msg map[string]interface{}) *ObserverMeta {
	meta := &ObserverMeta{}
	hasData := false

	if v, ok := msg["model"].(string); ok && v != "" {
		meta.Model = &v
		hasData = true
	}
	if v, ok := msg["firmware"].(string); ok && v != "" {
		meta.Firmware = &v
		hasData = true
	}
	if v, ok := msg["firmware_version"].(string); ok && v != "" {
		meta.Firmware = &v
		hasData = true
	}
	if v, ok := msg["client_version"].(string); ok && v != "" {
		meta.ClientVersion = &v
		hasData = true
	}
	if v, ok := msg["clientVersion"].(string); ok && v != "" {
		meta.ClientVersion = &v
		hasData = true
	}
	if v, ok := msg["radio"].(string); ok && v != "" {
		meta.Radio = &v
		hasData = true
	}

	// Stats fields may be nested under a "stats" object or at top level.
	// Try nested first, fall back to top-level for backward compatibility.
	stats, _ := msg["stats"].(map[string]interface{})

	if v := nestedOrTopLevel(stats, msg, "battery_mv"); v != nil {
		if f, ok := toFloat64(v); ok {
			iv := int(math.Round(f))
			meta.BatteryMv = &iv
			hasData = true
		}
	}
	if v := nestedOrTopLevel(stats, msg, "uptime_secs"); v != nil {
		if f, ok := toFloat64(v); ok {
			iv := int64(math.Round(f))
			meta.UptimeSecs = &iv
			hasData = true
		}
	}
	if v := nestedOrTopLevel(stats, msg, "noise_floor"); v != nil {
		if f, ok := toFloat64(v); ok {
			meta.NoiseFloor = &f
			hasData = true
		}
	}
	if v := nestedOrTopLevel(stats, msg, "tx_air_secs"); v != nil {
		if f, ok := toFloat64(v); ok {
			iv := int(math.Round(f))
			meta.TxAirSecs = &iv
			hasData = true
		}
	}
	if v := nestedOrTopLevel(stats, msg, "rx_air_secs"); v != nil {
		if f, ok := toFloat64(v); ok {
			iv := int(math.Round(f))
			meta.RxAirSecs = &iv
			hasData = true
		}
	}
	if v := nestedOrTopLevel(stats, msg, "recv_errors"); v != nil {
		if f, ok := toFloat64(v); ok {
			iv := int(math.Round(f))
			meta.RecvErrors = &iv
			hasData = true
		}
	}
	if v := nestedOrTopLevel(stats, msg, "packets_sent"); v != nil {
		if f, ok := toFloat64(v); ok {
			iv := int(math.Round(f))
			meta.PacketsSent = &iv
			hasData = true
		}
	}
	if v := nestedOrTopLevel(stats, msg, "packets_recv"); v != nil {
		if f, ok := toFloat64(v); ok {
			iv := int(math.Round(f))
			meta.PacketsRecv = &iv
			hasData = true
		}
	}

	if !hasData {
		return nil
	}
	return meta
}

// nestedOrTopLevel looks up a key in the nested map first, then the top-level map.
func nestedOrTopLevel(nested, toplevel map[string]interface{}, key string) interface{} {
	if nested != nil {
		if v, ok := nested[key]; ok {
			return v
		}
	}
	if v, ok := toplevel[key]; ok {
		return v
	}
	return nil
}

func firstNonEmpty(vals ...string) string {
	for _, v := range vals {
		if v != "" {
			return v
		}
	}
	return ""
}

// deriveHashtagChannelKey derives an AES-128 key from a channel name.
// Same algorithm as Node.js: SHA-256(channelName) → first 32 hex chars (16 bytes).
func deriveHashtagChannelKey(channelName string) string {
	h := sha256.Sum256([]byte(channelName))
	return hex.EncodeToString(h[:16])
}

// loadChannelKeys loads channel decryption keys from config and/or a JSON file.
// Merge priority: rainbow (lowest) → derived from hashChannels → explicit config (highest).
func loadChannelKeys(cfg *Config, configPath string) map[string]string {
	keys := make(map[string]string)

	// 1. Rainbow table keys (lowest priority)
	keysPath := os.Getenv("CHANNEL_KEYS_PATH")
	if keysPath == "" {
		keysPath = cfg.ChannelKeysPath
	}
	if keysPath == "" {
		keysPath = filepath.Join(filepath.Dir(configPath), "channel-rainbow.json")
	}

	rainbowCount := 0
	if data, err := os.ReadFile(keysPath); err == nil {
		var fileKeys map[string]string
		if err := json.Unmarshal(data, &fileKeys); err == nil {
			for k, v := range fileKeys {
				keys[k] = v
			}
			rainbowCount = len(fileKeys)
			log.Printf("Loaded %d channel keys from %s", rainbowCount, keysPath)
		} else {
			log.Printf("Warning: failed to parse channel keys file %s: %v", keysPath, err)
		}
	}

	// 2. Derived keys from hashChannels (middle priority)
	derivedCount := 0
	for _, raw := range cfg.HashChannels {
		trimmed := strings.TrimSpace(raw)
		if trimmed == "" {
			continue
		}
		channelName := trimmed
		if !strings.HasPrefix(channelName, "#") {
			channelName = "#" + channelName
		}
		// Skip if explicit config already has this key
		if _, exists := cfg.ChannelKeys[channelName]; exists {
			continue
		}
		keys[channelName] = deriveHashtagChannelKey(channelName)
		derivedCount++
	}
	if derivedCount > 0 {
		log.Printf("[channels] %d derived from hashChannels", derivedCount)
	}

	// 3. Explicit config keys (highest priority — overrides rainbow + derived)
	for k, v := range cfg.ChannelKeys {
		keys[k] = v
	}

	return keys
}

// Version info (set via ldflags)
var version = "dev"

func init() {
	if len(os.Args) > 1 && os.Args[1] == "--version" {
		fmt.Println("corescope-ingestor", version)
		os.Exit(0)
	}
}