res_levis/cmd/decoder/main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/AFASystems/presence/internal/pkg/model"
	"github.com/AFASystems/presence/internal/pkg/mqttclient"
	"github.com/redis/go-redis/v9"
	"github.com/segmentio/kafka-go"
)

func main() {
	// Load global context to init beacons and latest list
	appCtx := model.AppContext{
		Beacons: model.BeaconsList{
			Beacons: make(map[string]model.Beacon),
		},
		LatestList: model.LatestBeaconsList{
			LatestList: make(map[string]model.Beacon),
		},
	}

	// Kafka writer idk why yet
	writer := kafkaWriter("127.0.0.1:9092", "beacons")

	// Kafka reader for Raw MQTT beacons
	rawReader := kafkaReader("127.0.0.1:9092", "rawbeacons", "someID")
	defer rawReader.Close()

	// Kafka reader for API server updates
	apiReader := kafkaReader("127.0.0.1:9092", "apibeacons", "someID")
	defer apiReader.Close()

	// Kafka reader for latest list updates
	latestReader := kafkaReader("127.0.0.1:9092", "latestbeacons", "someID")
	defer latestReader.Close()

	defer writer.Close()

	ctx := context.Background()

	// Init Redis Client
	client := redis.NewClient(&redis.Options{
		Addr:     "127.0.0.1:6379",
		Password: "",
	})

	// Initialize list values from Redis
	beaconsList, err := client.Get(ctx, "beaconsList").Result()
	if err == redis.Nil {
		fmt.Println("no beacons list, starting empty")
	} else if err != nil {
		panic(err)
	} else {
		json.Unmarshal([]byte(beaconsList), &appCtx.Beacons.Beacons)
	}

	// Initialize list values from Redis
	latestList, err := client.Get(ctx, "latestList").Result()
	if err == redis.Nil {
		fmt.Println("no latest list, starting empty")
	} else if err != nil {
		panic(err)
	} else {
		json.Unmarshal([]byte(latestList), &appCtx.LatestList.LatestList)
	}

	// declare channel for collecting Kafka messages
	chRaw := make(chan model.Incoming_json, 2000)
	chApi := make(chan model.ApiUpdate, 2000)
	chLatest := make(chan model.Incoming_json, 2000)

	go consume(rawReader, chRaw)
	go consume(apiReader, chApi)
	go consume(latestReader, chLatest)

	go func() {
		ticker := time.NewTicker(1 * time.Second)
		defer ticker.Stop()

		for range ticker.C {
			appCtx.Beacons.Lock.Lock()
			data, _ := json.Marshal(appCtx.Beacons.Beacons)
			appCtx.Beacons.Lock.Unlock()

			err := client.Set(ctx, "beaconsList", data, 0).Err()
			if err != nil {
				fmt.Println("error saving to redis:", err)
			}

			appCtx.LatestList.Lock.Lock()
			ldata, _ := json.Marshal(appCtx.LatestList.LatestList)
			appCtx.LatestList.Lock.Unlock()

			err = client.Set(ctx, "latestList", ldata, 0).Err()
			if err != nil {
				fmt.Println("error saving latest list:", err)
			}
		}
	}()

	for {
		select {
		case msg := <-chRaw:
			processIncoming(msg, &appCtx)
		case msg := <-chApi:
			switch msg.Method {
			case "POST":
				fmt.Println("method POST")
				appCtx.Beacons.Lock.Lock()
				appCtx.Beacons.Beacons[msg.Beacon.Beacon_id] = msg.Beacon
			case "DELETE":
				_, exists := appCtx.Beacons.Beacons[msg.ID]
				if exists {
					appCtx.Beacons.Lock.Lock()
					delete(appCtx.Beacons.Beacons, msg.ID)
				}
				fmt.Println("method DELETE")
			default:
				fmt.Println("unknown method: ", msg.Method)
			}
			appCtx.Beacons.Lock.Unlock()
		case msg := <-chLatest:
			fmt.Println("latest msg: ", msg)
		}
	}
}

func kafkaWriter(kafkaURL, topic string) *kafka.Writer {
	return &kafka.Writer{
		Addr:         kafka.TCP(kafkaURL),
		Topic:        topic,
		Balancer:     &kafka.LeastBytes{},
		BatchSize:    100,
		BatchTimeout: 10 * time.Millisecond,
	}
}

func kafkaReader(kafkaURL, topic, groupID string) *kafka.Reader {
	brokers := strings.Split(kafkaURL, ",")
	return kafka.NewReader(kafka.ReaderConfig{
		Brokers:  brokers,
		GroupID:  groupID,
		Topic:    topic,
		MinBytes: 1,
		MaxBytes: 10e6,
	})
}

func consume[T any](r *kafka.Reader, ch chan<- T) {
	for {
		msg, err := r.ReadMessage(context.Background())
		if err != nil {
			fmt.Println("error reading message:", err)
			continue
		}

		var data T
		if err := json.Unmarshal(msg.Value, &data); err != nil {
			fmt.Println("error decoding:", err)
			continue
		}

		ch <- data
	}
}

func processIncoming(incoming model.Incoming_json, ctx *model.AppContext) {
	defer func() {
		if err := recover(); err != nil {
			fmt.Println("work failed:", err)
		}
	}()

	incoming = mqttclient.IncomingBeaconFilter(incoming)
	id := mqttclient.GetBeaconID(incoming)
	now := time.Now().Unix()

	beacons := &ctx.Beacons

	beacons.Lock.Lock()
	defer beacons.Lock.Unlock()

	latestList := &ctx.LatestList

	latestList.Lock.Lock()
	defer latestList.Lock.Unlock()

	beacon, exists := beacons.Beacons[id]
	if !exists {
		fmt.Println("beacon does not yet exist")
		fmt.Println("time now: ", now)
		return
	}

	fmt.Println("Beacon does exist: ", beacon)
}