feat: add postgresql container, ORM packages, refactor bridge to use paho MQTT library not the unmaintained one, refactor code to filter BLE beacons at bridge to reduce network traffic in Kafka

build/docker-compose.yaml

@@ -1,5 +1,21 @@
 version: "2"
 services:
+  db:
+    image: postgres
+    container_name: db
+    restart: always
+    ports:
+     - "127.0.0.1:5432:5432"
+    environment:
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=postgres
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+      start_period: 30s
+
   kafdrop:
     image: obsidiandynamics/kafdrop
     restart: "no"
@@ -65,6 +81,8 @@ services:
     depends_on:
       kafka-init:
         condition: service_completed_successfully
+      db:
+        condition: service_healthy
     restart: always
   
   presense-server:
@@ -83,6 +101,8 @@ services:
         condition: service_started
       kafka-init:
         condition: service_completed_successfully
+      db:
+        condition: service_healthy
     restart: always
 
   presense-bridge:
@@ -99,6 +119,8 @@ services:
     depends_on:
       kafka-init:
         condition: service_completed_successfully
+      db:
+        condition: service_healthy
     restart: always
 
   presense-location:
@@ -112,6 +134,8 @@ services:
     depends_on:
       kafka-init:
         condition: service_completed_successfully
+      db:
+        condition: service_healthy
     restart: always
       
 

cmd/bridge/main.go

@@ -1,57 +1,210 @@
 package main
 
 import (
+	"context"
+	"encoding/json"
 	"fmt"
+	"io"
+	"log"
+	"log/slog"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
 
-	"github.com/AFASystems/presence/internal/pkg/bridge/mqtthandler"
+	"github.com/AFASystems/presence/internal/pkg/common/appcontext"
 	"github.com/AFASystems/presence/internal/pkg/config"
 	"github.com/AFASystems/presence/internal/pkg/kafkaclient"
-	"github.com/yosssi/gmq/mqtt"
-	"github.com/yosssi/gmq/mqtt/client"
+	"github.com/AFASystems/presence/internal/pkg/model"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/segmentio/kafka-go"
 )
 
+var wg sync.WaitGroup
+
+func mqtthandler(writer *kafka.Writer, topic string, message []byte, appState *appcontext.AppState) {
+	hostname := strings.Split(topic, "/")[1]
+	msgStr := string(message)
+
+	if strings.HasPrefix(msgStr, "[") {
+		var readings []model.RawReading
+		err := json.Unmarshal(message, &readings)
+		if err != nil {
+			log.Printf("Error parsing JSON: %v", err)
+			return
+		}
+
+		for _, reading := range readings {
+			if reading.Type == "Gateway" {
+				continue
+			}
+			// fmt.Printf("reading: %+v\n", reading)
+			if !appState.BeaconExists(reading.MAC) {
+				fmt.Printf("Not tracking beacon: %s\n", reading.MAC)
+				continue
+			}
+
+			fmt.Printf("Tracking beacon: %s\n", reading.MAC)
+
+			adv := model.BeaconAdvertisement{
+				Hostname: hostname,
+				MAC:      reading.MAC,
+				RSSI:     int64(reading.RSSI),
+				Data:     reading.RawData,
+			}
+
+			encodedMsg, err := json.Marshal(adv)
+			if err != nil {
+				fmt.Println("Error in marshaling: ", err)
+				break
+			}
+
+			msg := kafka.Message{
+				Value: encodedMsg,
+			}
+			err = writer.WriteMessages(context.Background(), msg)
+			if err != nil {
+				fmt.Println("Error in writing to Kafka: ", err)
+				time.Sleep(1 * time.Second)
+				break
+			}
+		}
+	}
+	// } else {
+	// 	s := strings.Split(string(message), ",")
+	// 	if len(s) < 6 {
+	// 		log.Printf("Messaggio CSV non valido: %s", msgStr)
+	// 		return
+	// 	}
+
+	// 	rawdata := s[4]
+	// 	buttonCounter := parseButtonState(rawdata)
+	// 	if buttonCounter > 0 {
+	// 		adv := model.BeaconAdvertisement{}
+	// 		i, _ := strconv.ParseInt(s[3], 10, 64)
+	// 		adv.Hostname = hostname
+	// 		adv.BeaconType = "hb_button"
+	// 		adv.MAC = s[1]
+	// 		adv.RSSI = i
+	// 		adv.Data = rawdata
+	// 		adv.HSButtonCounter = buttonCounter
+
+	// 		read_line := strings.TrimRight(string(s[5]), "\r\n")
+	// 		it, err33 := strconv.Atoi(read_line)
+	// 		if err33 != nil {
+	// 			fmt.Println(it)
+	// 			fmt.Println(err33)
+	// 			os.Exit(2)
+	// 		}
+	// 	}
+	// }
+}
+
+var messagePubHandler = func(msg mqtt.Message, writer *kafka.Writer, appState *appcontext.AppState) {
+	mqtthandler(writer, msg.Topic(), msg.Payload(), appState)
+}
+
+var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
+	fmt.Println("Connected")
+}
+
+var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
+	fmt.Printf("Connect lost: %v", err)
+}
+
 func main() {
+	// Load global context to init beacons and latest list
+	appState := appcontext.NewAppState()
 	cfg := config.Load()
 
-	cli := client.New(&client.Options{
-		ErrorHandler: func(err error) {
-			fmt.Println("Error in initiating MQTT client: ", err)
-		},
-	})
-	defer cli.Terminate()
-
-	err := cli.Connect(&client.ConnectOptions{
-		Network:  "tcp",
-		Address:  cfg.MQTTHost,
-		ClientID: []byte(cfg.MQTTClientID),
-		UserName: []byte(cfg.MQTTUser),
-		Password: []byte(cfg.MQTTPass),
-	})
-
+	// Create log file -> this section and below can be moved in a package, as it is always the same
+	logFile, err := os.OpenFile("server.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
 	if err != nil {
-		fmt.Printf("Could not connect to MQTT broker, addr: %s", cfg.MQTTHost)
-		panic(err)
+		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)
 
-	fmt.Printf("Successfuly connected to MQTT broker, addr: %s", cfg.MQTTHost)
-
-	writer := kafkaclient.KafkaWriter(cfg.KafkaURL, "rawbeacons")
-	defer writer.Close()
-
-	err = cli.Subscribe(&client.SubscribeOptions{
-		SubReqs: []*client.SubReq{
-			{
-				TopicFilter: []byte("publish_out/#"),
-				QoS:         mqtt.QoS0,
-				Handler: func(topicName, message []byte) {
-					mqtthandler.MqttHandler(writer, topicName, message)
-				},
-			},
-		},
-	})
-	if err != nil {
-		panic(err)
+	// define context
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+	defer stop()
+
+	// define kafka reader
+	apiReader := appState.AddKafkaReader(cfg.KafkaURL, "apibeacons", "bridge-api")
+	// define kafka writer
+	writer := appState.AddKafkaWriter(cfg.KafkaURL, "rawbeacons")
+
+	slog.Info("Bridge initialized, subscribed to kafka topics")
+
+	chApi := make(chan model.ApiUpdate, 200)
+
+	wg.Add(1)
+	go kafkaclient.Consume(apiReader, chApi, ctx, &wg)
+
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(fmt.Sprintf("tcp://%s:%d", cfg.MQTTHost, 1883))
+	opts.SetClientID("go_mqtt_client")
+	opts.SetUsername("emqx")
+	opts.SetPassword("public")
+	opts.SetDefaultPublishHandler(func(c mqtt.Client, m mqtt.Message) { messagePubHandler(m, writer, appState) })
+	opts.OnConnect = connectHandler
+	opts.OnConnectionLost = connectLostHandler
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		panic(token.Error())
 	}
 
-	select {}
+	sub(client)
+
+eventloop:
+	for {
+		select {
+		case <-ctx.Done():
+			break eventloop
+		case msg := <-chApi:
+			switch msg.Method {
+			case "POST":
+				id := msg.Beacon.ID
+				appState.AddBeaconToLookup(id)
+				lMsg := fmt.Sprintf("Beacon added to lookup: %s", id)
+				slog.Info(lMsg)
+			case "DELETE":
+				id := msg.Beacon.ID
+				appState.RemoveBeaconFromLookup(id)
+				lMsg := fmt.Sprintf("Beacon removed from lookup: %s", id)
+				slog.Info(lMsg)
+			}
+		}
+	}
+
+	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()
+
+	client.Disconnect(250)
+	slog.Info("Closing connection to MQTT broker")
+}
+
+func publish(client mqtt.Client) {
+	num := 10
+	for i := 0; i < num; i++ {
+		text := fmt.Sprintf("Message %d", i)
+		token := client.Publish("topic/test", 0, false, text)
+		token.Wait()
+		time.Sleep(time.Second)
+	}
+}
+
+func sub(client mqtt.Client) {
+	topic := "publish_out/#"
+	token := client.Subscribe(topic, 1, nil)
+	token.Wait()
+	fmt.Printf("Subscribed to topic: %s\n", topic)
 }

cmd/decoder/main.go

@@ -44,18 +44,15 @@ func main() {
 	defer stop()
 
 	rawReader := appState.AddKafkaReader(cfg.KafkaURL, "rawbeacons", "gid-raw")
-	apiReader := appState.AddKafkaReader(cfg.KafkaURL, "apibeacons", "gid-api")
 
 	alertWriter := appState.AddKafkaWriter(cfg.KafkaURL, "alertbeacons")
 
 	slog.Info("Decoder initialized, subscribed to Kafka topics")
 
 	chRaw := make(chan model.BeaconAdvertisement, 2000)
-	chApi := make(chan model.ApiUpdate, 200)
 
 	wg.Add(2)
 	go kafkaclient.Consume(rawReader, chRaw, ctx, &wg)
-	go kafkaclient.Consume(apiReader, chApi, ctx, &wg)
 
 eventloop:
 	for {
@@ -64,19 +61,6 @@ eventloop:
 			break eventloop
 		case msg := <-chRaw:
 			processIncoming(msg, appState, alertWriter)
-		case msg := <-chApi:
-			switch msg.Method {
-			case "POST":
-				id := msg.Beacon.ID
-				appState.AddBeaconToLookup(id)
-				lMsg := fmt.Sprintf("Beacon added to lookup: %s", id)
-				slog.Info(lMsg)
-			case "DELETE":
-				id := msg.Beacon.ID
-				appState.RemoveBeaconFromLookup(id)
-				lMsg := fmt.Sprintf("Beacon removed from lookup: %s", id)
-				slog.Info(lMsg)
-			}
 		}
 	}
 
@@ -89,12 +73,6 @@ eventloop:
 }
 
 func processIncoming(adv model.BeaconAdvertisement, appState *appcontext.AppState, writer *kafka.Writer) {
-	id := adv.MAC
-	ok := appState.BeaconExists(id)
-	if !ok {
-		return
-	}
-
 	err := decodeBeacon(adv, appState, writer)
 	if err != nil {
 		eMsg := fmt.Sprintf("Error in decoding: %v", err)

cmd/location/main.go

@@ -43,7 +43,6 @@ func main() {
 	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")
@@ -54,12 +53,10 @@ func main() {
 	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:
@@ -71,19 +68,6 @@ eventLoop:
 			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)
 		}
@@ -201,14 +185,8 @@ func getLikelyLocations(appState *appcontext.AppState, writer *kafka.Writer) {
 
 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) {

go.mod

@@ -17,8 +17,23 @@ require (
 require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/eclipse/paho.golang v0.23.0 // indirect
+	github.com/eclipse/paho.mqtt.golang v1.5.1 // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/pgx/v5 v5.6.0 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/joho/godotenv v1.5.1 // indirect
 	github.com/klauspost/compress v1.15.9 // indirect
 	github.com/pierrec/lz4/v4 v4.1.15 // indirect
+	golang.org/x/crypto v0.42.0 // indirect
+	golang.org/x/net v0.44.0 // indirect
+	golang.org/x/sync v0.17.0 // indirect
 	golang.org/x/sys v0.37.0 // indirect
+	golang.org/x/text v0.29.0 // indirect
+	gorm.io/driver/postgres v1.6.0 // indirect
+	gorm.io/gorm v1.31.1 // indirect
 )

go.sum

@@ -6,10 +6,15 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/eclipse/paho.golang v0.23.0 h1:KHgl2wz6EJo7cMBmkuhpt7C576vP+kpPv7jjvSyR6Mk=
+github.com/eclipse/paho.golang v0.23.0/go.mod h1:nQRhTkoZv8EAiNs5UU0/WdQIx2NrnWUpL9nsGJTQN04=
+github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
+github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
 github.com/felixge/httpsnoop v1.0.3 h1:s/nj+GCswXYzN5v2DpNMuMQYe+0DDwt5WVCU6CWBdXk=
 github.com/felixge/httpsnoop v1.0.3/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/gorilla/handlers v1.5.2 h1:cLTUSsNkgcwhgRqvCNmdbRWG0A3N4F+M2nWKdScwyEE=
@@ -18,6 +23,20 @@ github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
 github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
+github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
+github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
+github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
+github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
+github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
+github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
 github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
@@ -28,8 +47,12 @@ github.com/redis/go-redis/v9 v9.16.0 h1:OotgqgLSRCmzfqChbQyG1PHC3tLNR89DG4jdOERS
 github.com/redis/go-redis/v9 v9.16.0/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370=
 github.com/segmentio/kafka-go v0.4.49 h1:GJiNX1d/g+kG6ljyJEoi9++PUMdXGAxb7JGPiDCuNmk=
 github.com/segmentio/kafka-go v0.4.49/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
 github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
 github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
@@ -38,11 +61,27 @@ github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6
 github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
 github.com/yosssi/gmq v0.0.1 h1:GhlDVaAQoi3Mvjul/qJXXGfL4JBeE0GQwbWp3eIsja8=
 github.com/yosssi/gmq v0.0.1/go.mod h1:mReykazh0U1JabvuWh1PEbzzJftqOQWsjr0Lwg5jL1Y=
+golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI=
+golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8=
 golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8=
 golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
+golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE=
+golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg=
+golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I=
+golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY=
+golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
+golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
 golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
 golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
 golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
+golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
+golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gorm.io/driver/postgres v1.6.0 h1:2dxzU8xJ+ivvqTRph34QX+WrRaJlmfyPqXmoGVjMBa4=
+gorm.io/driver/postgres v1.6.0/go.mod h1:vUw0mrGgrTK+uPHEhAdV4sfFELrByKVGnaVRkXDhtWo=
+gorm.io/gorm v1.31.1 h1:7CA8FTFz/gRfgqgpeKIBcervUn3xSyPUmr6B2WXJ7kg=
+gorm.io/gorm v1.31.1/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs=

internal/pkg/config/config.go

@@ -27,7 +27,7 @@ func Load() *Config {
 	return &Config{
 		HTTPAddr:     getEnv("HTTP_HOST_PATH", "0.0.0.0:1902"),
 		WSAddr:       getEnv("HTTPWS_HOST_PATH", "0.0.0.0:8088"),
-		MQTTHost:     getEnv("MQTT_HOST", "192.168.1.101:1883"),
+		MQTTHost:     getEnv("MQTT_HOST", "192.168.1.101"),
 		MQTTUser:     getEnv("MQTT_USERNAME", "user"),
 		MQTTPass:     getEnv("MQTT_PASSWORD", "pass"),
 		MQTTClientID: getEnv("MQTT_CLIENT_ID", "presence-detector"),