res_levis_backend/internal/app/decoder/app.go

package decoder

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

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

// DecoderApp holds dependencies for the decoder service.
type DecoderApp struct {
	Cfg            *config.Config
	KafkaManager   *kafkaclient.KafkaManager
	AppState       *appcontext.AppState
	ParserRegistry *model.ParserRegistry
	ChRaw          chan appcontext.BeaconAdvertisement
	ChParser       chan model.KafkaParser
	Cleanup        func()
	wg             sync.WaitGroup
}

// New creates a DecoderApp with Kafka readers (rawbeacons, parser) and writer (alertbeacons).
func New(cfg *config.Config) (*DecoderApp, error) {
	appState := appcontext.NewAppState()
	kafkaManager := kafkaclient.InitKafkaManager()

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

	readerTopics := []string{"rawbeacons", "parser"}
	writerTopics := []string{"alertbeacons", "healthdecoder"}
	kafkaManager.PopulateKafkaManager(cfg.KafkaURL, "decoder", readerTopics)
	kafkaManager.PopulateKafkaManager(cfg.KafkaURL, "", writerTopics)
	slog.Info("decoder service initialized", "readers", readerTopics, "writers", writerTopics)

	registry := &model.ParserRegistry{
		ParserList: make(map[string]model.BeaconParser),
	}

	return &DecoderApp{
		Cfg:            cfg,
		KafkaManager:   kafkaManager,
		AppState:       appState,
		ParserRegistry: registry,
		ChRaw:          make(chan appcontext.BeaconAdvertisement, config.LARGE_CHANNEL_SIZE),
		ChParser:       make(chan model.KafkaParser, config.SMALL_CHANNEL_SIZE),
		Cleanup:        cleanup,
	}, nil
}

// Run starts Kafka consumers and the event loop until ctx is cancelled.
func (a *DecoderApp) Run(ctx context.Context) {
	a.wg.Add(2)
	go kafkaclient.Consume(a.KafkaManager.GetReader("rawbeacons"), a.ChRaw, ctx, &a.wg)
	go kafkaclient.Consume(a.KafkaManager.GetReader("parser"), a.ChParser, 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.GetDecoderHealth(a.KafkaManager)
			slog.Info("decoder health", "health", string(health))
			if err != nil {
				slog.Error("getting decoder health", "err", err)
				continue
			}
			m := kafka.Message{
				Value: health,
			}
			if err := kafkaclient.Write(ctx, a.KafkaManager.GetWriter("healthdecoder"), m); err != nil {
				slog.Error("writing decoder health", "err", err)
				continue
			}
		case msg := <-a.ChRaw:
			decoder.ProcessIncoming(msg, a.AppState, a.KafkaManager.GetWriter("alertbeacons"), a.ParserRegistry)
		case msg := <-a.ChParser:
			switch msg.ID {
			case "add":
				a.ParserRegistry.Register(msg.Config.Name, msg.Config)
			case "delete":
				a.ParserRegistry.Unregister(msg.Name)
			case "update":
				a.ParserRegistry.Register(msg.Config.Name, msg.Config)
			}
		}
	}
}

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