res_levis_backend/internal/pkg/bridge/handler.go

package bridge

import (
	"context"
	"encoding/json"
	"log/slog"
	"strings"
	"time"

	"github.com/AFASystems/presence/internal/pkg/common/appcontext"
	"github.com/AFASystems/presence/internal/pkg/kafkaclient"
	"github.com/AFASystems/presence/internal/pkg/model"
	"github.com/segmentio/kafka-go"
)

// BeaconLookup provides MAC->ID lookup (e.g. AppState).
type BeaconLookup interface {
	BeaconExists(mac string) (id string, ok bool)
}

// HandleMQTTMessage processes an MQTT message: parses JSON array of RawReading or CSV.
// For JSON, converts each reading to BeaconAdvertisement and writes to the writer if MAC is in lookup.
// Hostname is derived from topic (e.g. "publish_out/gateway1" -> "gateway1"). Safe if topic has no "/".
func HandleMQTTMessage(topic string, payload []byte, lookup BeaconLookup, writer *kafka.Writer) {
	parts := strings.SplitN(topic, "/", 2)
	hostname := ""
	if len(parts) >= 2 {
		hostname = parts[1]
	}

	msgStr := string(payload)
	if strings.HasPrefix(msgStr, "[") {
		var readings []model.RawReading
		if err := json.Unmarshal(payload, &readings); err != nil {
			slog.Error("parsing MQTT JSON", "err", err, "topic", topic)
			return
		}
		for _, reading := range readings {
			if reading.Type == "Gateway" {
				continue
			}
			id, ok := lookup.BeaconExists(reading.MAC)
			if !ok {
				continue
			}
			adv := appcontext.BeaconAdvertisement{
				ID:       id,
				Hostname: hostname,
				MAC:      reading.MAC,
				RSSI:     int64(reading.RSSI),
				Data:     reading.RawData,
			}
			encoded, err := json.Marshal(adv)
			if err != nil {
				slog.Error("marshaling beacon advertisement", "err", err)
				break
			}
			if err := kafkaclient.Write(context.Background(), writer, kafka.Message{Value: encoded}); err != nil {
				slog.Error("writing to Kafka", "err", err)
				time.Sleep(1 * time.Second)
				break
			}
		}
		return
	}
	// CSV format: validate minimum fields (e.g. 6 columns); full parsing can be added later
	s := strings.Split(msgStr, ",")
	if len(s) < 6 {
		slog.Error("invalid CSV MQTT message", "topic", topic, "message", msgStr)
		return
	}
	slog.Debug("CSV MQTT message received", "topic", topic, "fields", len(s))
}