84 lines
3.6 KiB
Go
84 lines
3.6 KiB
Go
// handler.go — обработчик MQTT-сообщений: envelope → канал диспатчера SQS.
|
|
//
|
|
// Колбэк НЕ делает сетевых вызовов: только парсинг топика и неблокирующий
|
|
// enqueue. Отправка в SQS — worker-пул sqsDispatcher (см. sender.go).
|
|
package bridge
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"log/slog"
|
|
"regexp"
|
|
"strings"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
)
|
|
|
|
// TelemetryEnvelope — сообщение, перекладываемое в SQS (разбирается consumer).
|
|
type TelemetryEnvelope struct {
|
|
Namespace string `json:"namespace"`
|
|
DeviceID string `json:"device_id"`
|
|
Topic string `json:"topic"`
|
|
Payload json.RawMessage `json:"payload"`
|
|
ReceivedAt string `json:"received_at"`
|
|
// MessageID — уникальный id сообщения. Нужен для идемпотентности: SQS
|
|
// at-least-once может доставить дубликат — consumer дедуплицирует по нему.
|
|
MessageID string `json:"message_id"`
|
|
}
|
|
|
|
// newMessageHandler возвращает обработчик MQTT-сообщений:
|
|
// валидация топика "{namespace}/telemetry/{deviceId}" → enqueue в диспатчер.
|
|
func newMessageHandler(dispatcher *sqsDispatcher, log *slog.Logger) mqtt.MessageHandler {
|
|
return func(_ mqtt.Client, msg mqtt.Message) {
|
|
topic := msg.Topic()
|
|
payload := msg.Payload()
|
|
|
|
// Топик: "{namespace}/telemetry/{deviceId}". Пустые части — мусор.
|
|
parts := strings.SplitN(topic, "/", 3)
|
|
if len(parts) != 3 || parts[0] == "" || parts[2] == "" || parts[1] != "telemetry" {
|
|
log.Warn("bridge: unexpected topic format, skipping", "topic", topic)
|
|
return
|
|
}
|
|
ns := parts[0]
|
|
deviceID := parts[2]
|
|
|
|
// Defense-in-depth: EMQX ACL — основной барьер, но при его ошибке/отключении
|
|
// бридж принял бы произвольный namespace. Дополнительно проверяем формат.
|
|
if !identRe.MatchString(ns) || !identRe.MatchString(deviceID) {
|
|
log.Warn("bridge: invalid namespace/device in topic, skipping", "topic", topic)
|
|
return
|
|
}
|
|
|
|
// Не-JSON payload оборачиваем в строку.
|
|
rawPayload := json.RawMessage(payload)
|
|
if !json.Valid(payload) {
|
|
quoted, _ := json.Marshal(string(payload))
|
|
rawPayload = json.RawMessage(quoted)
|
|
}
|
|
|
|
dispatcher.enqueue(TelemetryEnvelope{
|
|
Namespace: ns,
|
|
DeviceID: deviceID,
|
|
Topic: topic,
|
|
Payload: rawPayload,
|
|
ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano),
|
|
MessageID: messageIDFor(ns, deviceID, topic, payload),
|
|
})
|
|
}
|
|
}
|
|
|
|
// identRe — безопасный идентификатор (namespace/device_id): буквы/цифры/дефис,
|
|
// не с дефиса, ≤64 символов. Defense-in-depth поверх EMQX ACL.
|
|
var identRe = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9-]{0,63}$`)
|
|
|
|
// messageIDFor — детерминированный id сообщения для дедупа.
|
|
// Повторная QoS1-доставка того же (device, topic, payload) даёт тот же id,
|
|
// поэтому ON CONFLICT (device_id, message_id) в БД дедуплицирует дубликаты
|
|
// (раньше uuid на каждый колбэк — дедуп на повторе не срабатывал).
|
|
func messageIDFor(ns, deviceID, topic string, payload []byte) string {
|
|
h := sha256.Sum256([]byte(ns + "|" + deviceID + "|" + topic + "|" + string(payload)))
|
|
return hex.EncodeToString(h[:])
|
|
}
|