res_levis_backend/internal/pkg/service/beacon_service.go

package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"slices"
	"strings"
	"time"

	"github.com/AFASystems/presence/internal/pkg/model"
	"github.com/google/uuid"
	"github.com/segmentio/kafka-go"
	"gorm.io/gorm"
)

// KafkaWriter defines the interface for writing Kafka messages (allows mocking in tests)
type KafkaWriter interface {
	WriteMessages(ctx context.Context, msgs ...kafka.Message) error
}

func findTracker(msg model.HTTPLocation, db *gorm.DB) (model.Tracker, error) {
	var tracker model.Tracker
	if msg.MAC != "" {
		if err := db.Where("mac = ?", msg.MAC).Find(&tracker).Error; err != nil {
			return model.Tracker{}, err
		}

		return tracker, nil
	}

	if msg.ID != "" {
		if err := db.Where("id = ?", msg.ID).Find(&tracker).Error; err != nil {
			return model.Tracker{}, err
		}

		return tracker, nil
	}

	return model.Tracker{}, errors.New("both ID and MAC are not provided")
}

func findZones(trackerID string, db *gorm.DB) ([]string, error) {
	var zones []model.TrackerZones
	if err := db.Select("zoneList").Where("tracker = ?", trackerID).Find(&zones).Error; err != nil {
		return nil, err
	}

	var allowedZones []string
	for _, z := range zones {
		allowedZones = append(allowedZones, z.ZoneList...)
	}

	return allowedZones, nil
}

func LocationToBeaconService(msg model.HTTPLocation, db *gorm.DB, writer KafkaWriter, ctx context.Context) {
	tracker, err := findTracker(msg, db)
	if err != nil {
		msg := fmt.Sprintf("Error in finding tracker: %v", err)
		slog.Error(msg)
		return
	}

	allowedZones, err := findZones(tracker.ID, db)
	if err != nil {
		msg := fmt.Sprintf("Error in finding zones: %v", err)
		slog.Error(msg)
		return
	}

	var gw model.Gateway
	mac := formatMac(msg.Location)
	if err := db.Select("*").Where("mac = ?", mac).First(&gw).Error; err != nil {
		msg := fmt.Sprintf("Gateway not found for MAC: %s", mac)
		slog.Error(msg)
		return
	}

	if err := db.Create(&model.Tracks{UUID: msg.ID, Timestamp: time.Now(), Gateway: gw.ID, GatewayMac: gw.MAC, Tracker: msg.ID, Floor: gw.Floor, Building: gw.Building, TrackerMac: tracker.MAC, Signal: msg.RSSI}).Error; err != nil {
		msg := fmt.Sprintf("Error in saving distance for beacon: %v", err)
		slog.Error(msg)
		return
	}

	err = db.Where("id = ?", msg.ID).Updates(model.Tracker{Position: gw.ID, X: gw.X, Y: gw.Y}).Error
	if err != nil {
		msg := fmt.Sprintf("Error in updating tracker: %v", err)
		slog.Error(msg)
		return
	}

	sendAlert(gw.ID, msg.ID, writer, ctx, allowedZones, db)
}

func LocationToBeaconServiceAI(msg model.HTTPLocation, db *gorm.DB, writer KafkaWriter, ctx context.Context) {
	tracker, err := findTracker(msg, db)
	if err != nil {
		msg := fmt.Sprintf("Error in finding tracker: %v", err)
		slog.Error(msg)
		return
	}

	allowedZones, err := findZones(tracker.ID, db)
	if err != nil {
		msg := fmt.Sprintf("Error in finding zones: %v", err)
		slog.Error(msg)
		return
	}

	var gw model.Gateway
	if err := db.Order(fmt.Sprintf("POW(x - %f, 2) + POW(y - %f, 2)", msg.X, msg.Y)).First(&gw).Error; err != nil {
		msg := fmt.Sprintf("Error in finding gateway: %v", err)
		slog.Error(msg)
		return
	}

	if err := db.Create(&model.Tracks{UUID: tracker.ID, Timestamp: time.Now(), Gateway: gw.ID, GatewayMac: gw.MAC, Tracker: tracker.ID, Floor: gw.Floor, Building: gw.Building, TrackerMac: tracker.MAC}).Error; err != nil {
		msg := fmt.Sprintf("Error in saving distance for beacon: %v", err)
		slog.Error(msg)
		return
	}

	err = db.Where("id = ?", tracker.ID).Updates(model.Tracker{Position: gw.ID, X: msg.X, Y: msg.Y}).Error
	if err != nil {
		msg := fmt.Sprintf("Error in updating tracker: %v", err)
		slog.Error(msg)
		return
	}

	sendAlert(gw.ID, tracker.ID, writer, ctx, allowedZones, db)
}

func sendAlert(gwId, trackerId string, writer KafkaWriter, ctx context.Context, allowedZones []string, db *gorm.DB) {
	if len(allowedZones) != 0 && !slices.Contains(allowedZones, gwId) {
		alert := model.Alert{
			ID:        uuid.New().String(),
			TrackerID: trackerId,
			Type:      "Restricted zone",
			Value:     gwId,
		}

		if err := InsertAlert(alert, db, ctx); err != nil {
			msg := fmt.Sprintf("Error in inserting alert: %v", err)
			slog.Error(msg)
			return
		}

		eMsg, err := json.Marshal(alert)
		if err != nil {
			msg := "Error in marshaling"
			slog.Error(msg)
			return
		} else {
			msg := kafka.Message{
				Value: eMsg,
			}
			writer.WriteMessages(ctx, msg)
		}
	}
}

func formatMac(MAC string) string {
	var res strings.Builder
	for i := 0; i < len(MAC); i += 2 {
		if i > 0 {
			res.WriteByte(':')
		}

		end := min(i+2, len(MAC))

		res.WriteString(MAC[i:end])
	}

	return res.String()
}