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") apiReader := appState.AddKafkaReader(cfg.KafkaURL, "apibeacons", "gid-api-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) chApi := make(chan model.ApiUpdate, 2000) chSettings := make(chan model.SettingsVal, 5) wg.Add(3) go kafkaclient.Consume(rawReader, chRaw, ctx, &wg) go kafkaclient.Consume(apiReader, chApi, 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 := <-chApi: switch msg.Method { case "POST": id := msg.Beacon.ID lMsg := fmt.Sprintf("Beacon added to lookup: %s", id) slog.Info(lMsg) appState.AddBeaconToLookup(id) case "DELETE": id := msg.Beacon.ID appState.RemoveBeaconFromLookup(id) lMsg := fmt.Sprintf("Beacon removed from lookup: %s", id) slog.Info(lMsg) } 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 { // 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 // Why do I need this if I am sending entire structure anyways? who knows 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.MAC ok := appState.BeaconExists(id) now := time.Now().Unix() if !ok { appState.UpdateLatestBeacon(id, model.Beacon{ID: id, BeaconType: adv.BeaconType, LastSeen: now, IncomingJSON: adv, BeaconLocation: adv.Hostname, Distance: utils.CalculateDistance(adv)}) return } 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) }