res_levis_backend/cmd/location/main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"log/slog"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"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()

	// 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-loc")
	settingsReader := appState.AddKafkaReader(cfg.KafkaURL, "settings", "gid-settings-loc")

	writer := appState.AddKafkaWriter(cfg.KafkaURL, "locevents")

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

	locTicker := time.NewTicker(1 * time.Second)
	defer locTicker.Stop()

	chRaw := make(chan model.BeaconAdvertisement, 2000)
	chSettings := make(chan model.SettingsVal, 5)

	wg.Add(3)
	go kafkaclient.Consume(rawReader, chRaw, ctx, &wg)
	go kafkaclient.Consume(settingsReader, chSettings, ctx, &wg)

eventLoop:
	for {
		select {
		case <-ctx.Done():
			break eventLoop
		case <-locTicker.C:
			getLikelyLocations(appState, writer)
		case msg := <-chRaw:
			assignBeaconToList(msg, appState)
		case msg := <-chSettings:
			appState.UpdateSettings(msg)
		}
	}

	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 getLikelyLocations(appState *appcontext.AppState, writer *kafka.Writer) {
	beacons := appState.GetAllBeacons()
	settings := appState.GetSettingsValue()

	for _, beacon := range beacons {
		fmt.Println("id: ", beacon.ID)
		// Shrinking the model because other properties have nothing to do with the location
		r := model.HTTPLocation{
			Method:   "Standard",
			Distance: 999,
			ID:       beacon.ID,
			Location: "",
			LastSeen: 999,
		}

		mSize := len(beacon.BeaconMetrics)

		if (int64(time.Now().Unix()) - (beacon.BeaconMetrics[mSize-1].Timestamp)) > settings.LastSeenThreshold {
			slog.Warn("beacon is too old")
			continue
		}

		locList := make(map[string]float64)
		seenW := 1.5
		rssiW := 0.75
		for _, metric := range beacon.BeaconMetrics {
			res := seenW + (rssiW * (1.0 - (float64(metric.RSSI) / -100.0)))
			locList[metric.Location] += res
		}

		bestLocName := ""
		maxScore := 0.0
		for locName, score := range locList {
			if score > maxScore {
				maxScore = score
				bestLocName = locName
			}
		}

		if bestLocName == beacon.PreviousLocation {
			beacon.LocationConfidence++
		} else {
			beacon.LocationConfidence = 0
		}

		r.Distance = beacon.BeaconMetrics[mSize-1].Distance
		r.Location = bestLocName
		r.LastSeen = beacon.BeaconMetrics[mSize-1].Timestamp

		if beacon.LocationConfidence == settings.LocationConfidence && beacon.PreviousConfidentLocation != bestLocName {
			beacon.LocationConfidence = 0
			// js, err := json.Marshal(model.LocationChange{
			// 	Method:           "LocationChange",
			// 	BeaconRef:        beacon,
			// 	Name:             beacon.Name,
			// 	PreviousLocation: beacon.PreviousConfidentLocation,
			// 	NewLocation:      bestLocName,
			// 	Timestamp:        time.Now().Unix(),
			// })

			// if err != nil {
			// 	eMsg := fmt.Sprintf("Error in marshaling: %v", err)
			// 	slog.Error(eMsg)
			// 	beacon.PreviousConfidentLocation = bestLocName
			// 	beacon.PreviousLocation = bestLocName
			// 	appState.UpdateBeacon(beacon.ID, beacon)
			// 	continue
			// }

			// msg := kafka.Message{
			// 	Value: js,
			// }

			// err = writer.WriteMessages(context.Background(), msg)
			// if err != nil {
			// 	fmt.Println("Error in sending Kafka message")
			// }
		}

		beacon.PreviousLocation = bestLocName
		appState.UpdateBeacon(beacon.ID, beacon)

		js, err := json.Marshal(r)
		if err != nil {
			eMsg := fmt.Sprintf("Error in marshaling location: %v", err)
			slog.Error(eMsg)
			continue
		}

		msg := kafka.Message{
			Value: js,
		}

		err = writer.WriteMessages(context.Background(), msg)
		if err != nil {
			eMsg := fmt.Sprintf("Error in sending Kafka message: %v", err)
			slog.Error(eMsg)
		}
	}
}

func assignBeaconToList(adv model.BeaconAdvertisement, appState *appcontext.AppState) {
	id := adv.ID
	now := time.Now().Unix()

	settings := appState.GetSettingsValue()

	if settings.RSSIEnforceThreshold && (int64(adv.RSSI) < settings.RSSIMinThreshold) {
		slog.Info("Settings returns")
		return
	}

	beacon, ok := appState.GetBeacon(id)
	if !ok {
		beacon = model.Beacon{
			ID: id,
		}
	}

	beacon.IncomingJSON = adv
	beacon.LastSeen = now

	if beacon.BeaconMetrics == nil {
		beacon.BeaconMetrics = make([]model.BeaconMetric, 0, settings.BeaconMetricSize)
	}

	metric := model.BeaconMetric{
		Distance:  utils.CalculateDistance(adv),
		Timestamp: now,
		RSSI:      int64(adv.RSSI),
		Location:  adv.Hostname,
	}

	if len(beacon.BeaconMetrics) >= settings.BeaconMetricSize {
		copy(beacon.BeaconMetrics, beacon.BeaconMetrics[1:])
		beacon.BeaconMetrics[settings.BeaconMetricSize-1] = metric
	} else {
		beacon.BeaconMetrics = append(beacon.BeaconMetrics, metric)
	}

	appState.UpdateBeacon(id, beacon)
}