res_levis_backend/cmd/decoder/main.go

package main

import (
	"bytes"
	"context"
	"encoding/hex"
	"fmt"
	"io"
	"log"
	"log/slog"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"

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

var wg sync.WaitGroup

func main() {
	// Load global context to init beacons and latest list
	appState := appcontext.NewAppState()
	cfg := config.Load()

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

	// Create log file
	logFile, err := os.OpenFile("server.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
	if err != nil {
		log.Fatalf("Failed to open log file: %v\n", err)
	}
	// shell and log file multiwriter
	w := io.MultiWriter(os.Stderr, logFile)
	logger := slog.New(slog.NewJSONHandler(w, nil))
	slog.SetDefault(logger)

	// define context
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
	defer stop()

	rawReader := appState.AddKafkaReader(cfg.KafkaURL, "rawbeacons", "gid-raw")
	parserReader := appState.AddKafkaReader(cfg.KafkaURL, "parser", "gid-parser")

	alertWriter := appState.AddKafkaWriter(cfg.KafkaURL, "alertbeacons")

	slog.Info("Decoder initialized, subscribed to Kafka topics")

	chRaw := make(chan model.BeaconAdvertisement, 2000)
	chParser := make(chan model.KafkaParser, 200)

	wg.Add(3)
	go kafkaclient.Consume(rawReader, chRaw, ctx, &wg)
	go kafkaclient.Consume(parserReader, chParser, ctx, &wg)

eventloop:
	for {
		select {
		case <-ctx.Done():
			break eventloop
		case msg := <-chRaw:
			processIncoming(msg, appState, alertWriter, &parserRegistry)
		case msg := <-chParser:
			switch msg.ID {
			case "add":
				config := msg.Config
				parserRegistry.Register(config.Name, config)
			case "delete":
				parserRegistry.Unregister(msg.Name)
			case "update":
				config := msg.Config
				parserRegistry.Register(config.Name, config)
			}
		}
	}

	slog.Info("broken out of the main event loop")
	wg.Wait()

	slog.Info("All go routines have stopped, Beggining to close Kafka connections")
	appState.CleanKafkaReaders()
	appState.CleanKafkaWriters()
}

func processIncoming(adv model.BeaconAdvertisement, appState *appcontext.AppState, writer *kafka.Writer, parserRegistry *model.ParserRegistry) {
	err := decodeBeacon(adv, appState, writer, parserRegistry)
	if err != nil {
		eMsg := fmt.Sprintf("Error in decoding: %v", err)
		fmt.Println(eMsg)
		return
	}
}

func decodeBeacon(adv model.BeaconAdvertisement, appState *appcontext.AppState, writer *kafka.Writer, parserRegistry *model.ParserRegistry) error {
	beacon := strings.TrimSpace(adv.Data)
	id := adv.ID
	if beacon == "" {
		return nil
	}

	b, err := hex.DecodeString(beacon)
	if err != nil {
		return err
	}

	b = utils.RemoveFlagBytes(b)

	indeces := utils.ParseADFast(b)
	event := utils.LoopADStructures(b, indeces, id, parserRegistry)

	if event.ID == "" {
		return nil
	}

	prevEvent, ok := appState.GetBeaconEvent(id)
	appState.UpdateBeaconEvent(id, event)
	if ok && bytes.Equal(prevEvent.Hash(), event.Hash()) {
		return nil
	}

	eMsg, err := event.ToJSON()
	if err != nil {
		return err
	}

	if err := writer.WriteMessages(context.Background(), kafka.Message{Value: eMsg}); err != nil {
		return err
	}

	return nil
}