res_levis_backend/cmd/decoder/main.go

package main

import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strings"

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

func main() {
	// Load global context to init beacons and latest list
	appCtx := model.AppContext{
		Beacons: model.BeaconsList{
			Beacons: make(map[string]model.Beacon),
		},
		Settings: model.Settings{
			Settings: model.SettingsVal{
				LocationConfidence: 4,
				LastSeenThreshold:  15,
				BeaconMetricSize:   30,
				HASendInterval:     5,
				HASendChangesOnly:  false,
			},
		},
		BeaconEvents: model.BeaconEventList{
			Beacons: make(map[string]model.BeaconEvent),
		},
		BeaconsLookup: make(map[string]struct{}),
	}

	cfg := config.Load()

	// Kafka reader for Raw MQTT beacons
	rawReader := kafkaclient.KafkaReader(cfg.KafkaURL, "rawbeacons", "gid-raw")
	defer rawReader.Close()

	// Kafka reader for API server updates
	apiReader := kafkaclient.KafkaReader(cfg.KafkaURL, "apibeacons", "gid-api")
	defer apiReader.Close()

	alertWriter := kafkaclient.KafkaWriter(cfg.KafkaURL, "alertbeacons")
	defer alertWriter.Close()

	fmt.Println("Decoder initialized, subscribed to Kafka topics")
	chRaw := make(chan model.BeaconAdvertisement, 2000)
	chApi := make(chan model.ApiUpdate, 2000)

	go kafkaclient.Consume(rawReader, chRaw)
	go kafkaclient.Consume(apiReader, chApi)

	for {
		select {
		case msg := <-chRaw:
			processIncoming(msg, &appCtx, alertWriter)
		case msg := <-chApi:
			switch msg.Method {
			case "POST":
				id := msg.Beacon.ID
				appCtx.BeaconsLookup[id] = struct{}{}
			case "DELETE":
				fmt.Println("Incoming delete message")
			}
		}
	}
}

func processIncoming(adv model.BeaconAdvertisement, ctx *model.AppContext, writer *kafka.Writer) {
	id := adv.MAC
	_, ok := ctx.BeaconsLookup[id]
	if !ok {
		return
	}

	err := decodeBeacon(adv, ctx, writer)
	if err != nil {
		fmt.Println("error in decoding")
		return
	}
}

func decodeBeacon(adv model.BeaconAdvertisement, ctx *model.AppContext, writer *kafka.Writer) error {
	beacon := strings.TrimSpace(adv.Data)
	id := adv.MAC
	if beacon == "" {
		return nil // How to return error?, do I even need to return error
	}

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

	// check for flag byte, if first AD structure is flag bytes, remove it
	if len(b) > 1 && b[1] == 0x01 {
		l := int(b[0]) // length of AD structure
		if 1+l <= len(b) {
			b = b[1+l:]
		}
	}

	adStructureIndeces := ParseADFast(b)
	event := model.BeaconEvent{}
	for _, r := range adStructureIndeces {
		ad := b[r[0]:r[1]]
		if checkIngics(ad) {
			event = parseIngicsState(ad)
			event.ID = id
			event.Name = id
			break
		} else if checkEddystoneTLM(ad) {
			event = parseEddystoneState(ad)
			event.ID = id
			event.Name = id
			break
		} else if checkMinewB7(ad) {
			fmt.Println("Minew B7 vendor format")
			break
		}
	}

	if event.ID != "" {
		prevEvent, ok := ctx.BeaconEvents.Beacons[id]
		ctx.BeaconEvents.Beacons[id] = event
		if ok && bytes.Equal(prevEvent.Hash(), event.Hash()) {
			return nil
		}

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

		err = writer.WriteMessages(context.Background(), kafka.Message{
			Value: eMsg,
		})

		if err != nil {
			return err
		}

		fmt.Println("Message sent")
	}

	return nil
}

func checkIngics(ad []byte) bool {
	if len(ad) >= 6 &&
		ad[1] == 0xFF &&
		ad[2] == 0x59 &&
		ad[3] == 0x00 &&
		ad[4] == 0x80 &&
		ad[5] == 0xBC {
		return true
	}

	return false
}

func parseIngicsState(ad []byte) model.BeaconEvent {
	return model.BeaconEvent{
		Battery: uint32(binary.LittleEndian.Uint16(ad[6:8])),
		Event:   int(ad[8]),
		Type:    "Ingics",
	}
}

func checkEddystoneTLM(ad []byte) bool {
	if len(ad) >= 4 &&
		ad[1] == 0x16 &&
		ad[2] == 0xAA &&
		ad[3] == 0xFE &&
		ad[4] == 0x20 {
		return true
	}

	return false
}

func parseEddystoneState(ad []byte) model.BeaconEvent {
	return model.BeaconEvent{
		Battery: uint32(binary.BigEndian.Uint16(ad[6:8])),
		Type:    "Eddystone",
	}
}

// I dont think this is always true, but for testing is ok
func checkMinewB7(ad []byte) bool {
	if len(ad) >= 4 &&
		ad[1] == 0x16 &&
		ad[2] == 0xE1 &&
		ad[3] == 0xFF {
		return true
	}

	return false
}

func ParseADFast(b []byte) [][2]int {
	var res [][2]int
	i := 0

	for i < len(b) {
		l := int(b[i])
		if l == 0 || i+1+l > len(b) {
			break
		}

		res = append(res, [2]int{i, i + 1 + l})

		i += 1 + l
	}

	return res
}