res_levis_backend/internal/app/bridge/app.go

package bridge

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

	"github.com/AFASystems/presence/internal/pkg/bridge"
	"github.com/AFASystems/presence/internal/pkg/common/appcontext"
	"github.com/AFASystems/presence/internal/pkg/config"
	"github.com/AFASystems/presence/internal/pkg/kafkaclient"
	"github.com/AFASystems/presence/internal/pkg/logger"
	"github.com/AFASystems/presence/internal/pkg/model"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/segmentio/kafka-go"
)

// BridgeApp holds dependencies for the bridge service (MQTT <-> Kafka).
type BridgeApp struct {
	Cfg          *config.Config
	KafkaManager *kafkaclient.KafkaManager
	AppState     *appcontext.AppState
	MQTT         *bridge.MQTTClient
	ChApi        chan model.ApiUpdate
	ChAlert      chan model.Alert
	ChMqtt       chan []model.Tracker
	Cleanup      func()
	wg           sync.WaitGroup
}

// New creates a BridgeApp with Kafka readers (apibeacons, alert, mqtt), writer (rawbeacons), and MQTT client.
func New(cfg *config.Config) (*BridgeApp, error) {
	appState := appcontext.NewAppState()
	kafkaManager := kafkaclient.InitKafkaManager()

	srvLogger, cleanup := logger.CreateLogger("bridge.log")
	slog.SetDefault(srvLogger)

	readerTopics := []string{"apibeacons", "alert", "mqtt"}
	writerTopics := []string{"rawbeacons", "healthbridge"}
	kafkaManager.PopulateKafkaManager(cfg.KafkaURL, "bridge", readerTopics)
	kafkaManager.PopulateKafkaManager(cfg.KafkaURL, "", writerTopics)
	slog.Info("bridge service initialized", "readers", readerTopics, "writers", writerTopics)

	writer := kafkaManager.GetWriter("rawbeacons")
	mqttClient, err := bridge.NewMQTTClient(cfg, func(m mqtt.Message) {
		bridge.HandleMQTTMessage(m.Topic(), m.Payload(), appState, writer)
	})
	if err != nil {
		cleanup()
		return nil, err
	}
	mqttClient.Subscribe()

	return &BridgeApp{
		Cfg:          cfg,
		KafkaManager: kafkaManager,
		AppState:     appState,
		MQTT:         mqttClient,
		ChApi:        make(chan model.ApiUpdate, config.SMALL_CHANNEL_SIZE),
		ChAlert:      make(chan model.Alert, config.SMALL_CHANNEL_SIZE),
		ChMqtt:       make(chan []model.Tracker, config.SMALL_CHANNEL_SIZE),
		Cleanup:      cleanup,
	}, nil
}

// Run starts Kafka consumers and the event loop until ctx is cancelled.
func (a *BridgeApp) Run(ctx context.Context) {
	a.wg.Add(3)
	go kafkaclient.Consume(a.KafkaManager.GetReader("apibeacons"), a.ChApi, ctx, &a.wg)
	go kafkaclient.Consume(a.KafkaManager.GetReader("alert"), a.ChAlert, ctx, &a.wg)
	go kafkaclient.Consume(a.KafkaManager.GetReader("mqtt"), a.ChMqtt, ctx, &a.wg)

	healthTicker := time.NewTicker(config.LARGE_TICKER_INTERVAL)
	defer healthTicker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-healthTicker.C:
			health, err := a.AppState.GetBridgeHealth(a.KafkaManager)
			if err != nil {
				slog.Error("getting bridge health", "err", err)
				continue
			}
			m := kafka.Message{
				Value: health,
			}
			if err := kafkaclient.Write(ctx, a.KafkaManager.GetWriter("healthbridge"), m); err != nil {
				slog.Error("writing bridge health", "err", err)
				continue
			}
		case msg := <-a.ChApi:
			switch msg.Method {
			case "POST":
				a.AppState.AddBeaconToLookup(msg.MAC, msg.ID)
				slog.Info("beacon added to lookup", "id", msg.ID)
			case "DELETE":
				if msg.MAC == "all" {
					a.AppState.CleanLookup()
					slog.Info("lookup cleared")
					continue
				}
				a.AppState.RemoveBeaconFromLookup(msg.MAC)
				slog.Info("beacon removed from lookup", "mac", msg.MAC)
			}
		case msg := <-a.ChAlert:
			p, err := json.Marshal(msg)
			if err != nil {
				slog.Error("marshaling alert", "err", err)
				continue
			}
			a.MQTT.Client.Publish("/alerts", 0, true, p)
		case msg := <-a.ChMqtt:
			p, err := json.Marshal(msg)
			if err != nil {
				slog.Error("marshaling trackers", "err", err)
				continue
			}
			a.MQTT.Client.Publish("/trackers", 0, true, p)
		}
	}
}

// Shutdown disconnects MQTT, waits for consumers, and cleans up.
func (a *BridgeApp) Shutdown() {
	a.wg.Wait()
	if a.MQTT != nil {
		a.MQTT.Disconnect()
	}
	a.KafkaManager.CleanKafkaReaders()
	a.KafkaManager.CleanKafkaWriters()
	if a.Cleanup != nil {
		a.Cleanup()
	}
	slog.Info("bridge service shutdown complete")
}