Add MQTT presence to Redis integration

mqtt/Dockerfile

@@ -0,0 +1,26 @@
+# syntax=docker/dockerfile:1
+
+# Build stage
+FROM golang:1.22-alpine AS builder
+WORKDIR /app
+
+# Copy go mod files and download dependencies
+COPY go.mod go.sum ./
+RUN go mod download
+
+# Copy the rest of the application
+COPY . .
+
+# Build the application
+RUN go build -o mqtt_presence_redis_go mqtt_presence_redis_go.go
+
+# Runtime stage
+FROM alpine:3.20
+WORKDIR /app
+COPY --from=builder /app/mqtt_presence_redis_go .
+
+# Expose default MQTT port if needed
+EXPOSE 1883
+
+# Command to run the binary
+CMD ["./mqtt_presence_redis_go"]

mqtt/go.mod

@@ -0,0 +1,16 @@
+module git.proit.id/dsutanto/backone-manage-go
+
+go 1.24.3
+
+require (
+	github.com/eclipse/paho.mqtt.golang v1.5.1
+	github.com/go-redis/redis/v8 v8.11.5
+	golang.org/x/net v0.49.0
+)
+
+require (
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
+	golang.org/x/sync v0.17.0 // indirect
+)

mqtt/go.sum

@@ -0,0 +1,30 @@
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+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.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
+github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
+github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
+github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
+github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
+github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
+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/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
+github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
+github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
+github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
+github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
+github.com/onsi/gomega v1.18.1/go.mod h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5hitRs=
+golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o=
+golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8=
+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.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ=
+golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE=
+golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=

mqtt/mqtt_presence_redis_go.go

@@ -0,0 +1,123 @@
+package main
+
+import (
+    "encoding/json"
+    "fmt"
+    "log"
+    "os"
+    "strings"
+    "time"
+    "strconv"
+
+    mqtt "github.com/eclipse/paho.mqtt.golang"
+    "github.com/go-redis/redis/v8"
+    "golang.org/x/net/context"
+)
+
+// Configuration loaded from environment variables. These should match the
+// Django settings used in the original Python implementation.
+var (
+    mqttHost          = getEnv("MQTT_HOST", "localhost")
+    mqttPort          = getEnv("MQTT_PORT", "1883")
+    mqttUser          = getEnv("MQTT_USER", "")
+    mqttPass          = getEnv("MQTT_PASS", "")
+    mqttTopicPresence = getEnv("MQTT_TOPIC_PRESENCE", "presence")
+
+    redisHost = getEnv("MQTT_REDIS_HOST", "localhost")
+    redisPort = getEnv("MQTT_REDIS_PORT", "6379")
+    redisDB   = getEnv("MQTT_REDIS_DB", "0")
+
+    redisPrefix   = getEnv("MQTT_REDIS_PREFIX", "presence")
+    redisSetEX    = getEnv("MQTT_REDIS_SETEX", "86400") // seconds
+    redisPassword = getEnv("MQTT_REDIS_PASSWORD", "")
+)
+
+func main() {
+    ctx := context.Background()
+
+    // Connect to Redis
+    rdb := redis.NewClient(&redis.Options{
+        Addr:     fmt.Sprintf("%s:%s", redisHost, redisPort),
+        Password: redisPassword,
+        DB:       atoi(redisDB),
+    })
+
+    // Ping to verify connection
+    if err := rdb.Ping(ctx).Err(); err != nil {
+        log.Fatalf("Could not connect to Redis: %v", err)
+    }
+
+    // MQTT client options
+    opts := mqtt.NewClientOptions()
+    opts.AddBroker(fmt.Sprintf("tcp://%s:%s", mqttHost, mqttPort))
+    opts.SetUsername(mqttUser)
+    opts.SetPassword(mqttPass)
+    opts.SetClientID("mqtt_presence_redis_go")
+
+    // Handlers
+    opts.SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {
+        handleMessage(ctx, rdb, msg)
+    })
+
+    opts.OnConnect = func(c mqtt.Client) {
+        if token := c.Subscribe(mqttTopicPresence, 0, nil); token.Wait() && token.Error() != nil {
+            log.Fatalf("Failed to subscribe: %v", token.Error())
+        }
+        log.Printf("Connected to MQTT broker %s:%s, subscribed to %s", mqttHost, mqttPort, mqttTopicPresence)
+    }
+
+    // Create and start client
+    client := mqtt.NewClient(opts)
+    if token := client.Connect(); token.Wait() && token.Error() != nil {
+        log.Fatalf("Could not connect to MQTT broker: %v", token.Error())
+    }
+
+    // Block forever
+    select {}
+}
+
+func handleMessage(ctx context.Context, rdb *redis.Client, msg mqtt.Message) {
+    payload := string(msg.Payload())
+    log.Printf("Received message: %s", payload)
+
+    parts := strings.Split(payload, ";")
+    if len(parts) == 0 {
+        return
+    }
+    memberID := parts[0]
+    if len(memberID) > 50 {
+        memberID = memberID[:50]
+    }
+
+    key := fmt.Sprintf("%s:%s", redisPrefix, memberID)
+    timestamp := time.Now().Unix()
+
+    msgObj := map[string]interface{}{
+        "mqtt": payload,
+        "ts":   timestamp,
+    }
+    data, err := json.Marshal(msgObj)
+    if err != nil {
+        log.Printf("Failed to marshal message: %v", err)
+        return
+    }
+
+    if err := rdb.SetEX(ctx, key, data, time.Duration(atoi(redisSetEX))*time.Second).Err(); err != nil {
+        log.Printf("Failed to set Redis key: %v", err)
+    }
+}
+
+func getEnv(key, defaultVal string) string {
+    if v := os.Getenv(key); v != "" {
+        return v
+    }
+    return defaultVal
+}
+
+func atoi(s string) int {
+    i, err := strconv.Atoi(s)
+    if err != nil {
+        return 0
+    }
+    return i
+}