bridge: MessageID=хэш(дедуп), классификация permanent/transient SQS-ошибок, бесконечный ретрай подписки, валидация ns/deviceID

internal/service/bridge/bridge.go

@@ -104,28 +104,29 @@ func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, hand
 	})
 	opts.SetOnConnectHandler(func(c mqtt.Client) {
 		log.Info("bridge: MQTT connected")
-		// Подписка обязательна. Ретраим ограниченно. НЕ вызываем c.Disconnect():
-		// в paho это ЯВНОЕ отключение, после которого автореконнект НЕ срабатывает
-		// — bridge оставался бы отключённым навсегда (ревью 2026-10-10, 🔴6).
-		// При сбое ждём следующего реконнекта (OnConnectHandler вызовется заново).
-		for attempt := 1; attempt <= 3; attempt++ {
+		// Подписка обязательна. Ретраим пока не удастся (или не оборвётся ctx/
+		// соединение). НЕ вызываем c.Disconnect(): в paho это ЯВНОЕ отключение,
+		// после которого автореконнект НЕ срабатывает (ревью 2026-10-10, 🔴6).
+		// Раньше было 3 попытки и ожидание следующего реконнекта — но при живом
+		// соединении OnConnectHandler больше не вызовется, и bridge оставался бы
+		// без подписки навсегда (ревью 2026-10-11).
+		for {
+			if ctx.Err() != nil || !c.IsConnected() {
+				return
+			}
 			token := c.Subscribe(telemetryTopicFilter, 1, handler)
 			if token.WaitTimeout(10*time.Second) && token.Error() == nil {
-				log.Info("bridge: subscribed", "filter", telemetryTopicFilter, "attempt", attempt)
+				log.Info("bridge: subscribed", "filter", telemetryTopicFilter)
 				setSubscribed(onSubscribed, true)
 				return
 			}
-			log.Error("bridge: subscribe failed", "filter", telemetryTopicFilter, "attempt", attempt, "err", token.Error())
+			log.Error("bridge: subscribe failed, retrying", "filter", telemetryTopicFilter, "err", token.Error())
 			select {
 			case <-ctx.Done():
 				return
 			case <-time.After(subscribeRetryInterval):
 			}
 		}
-		// Подписка не удалась — bridge не готов (/ready=503) до следующего
-		// реконнекта.
-		setSubscribed(onSubscribed, false)
-		log.Error("bridge: subscribe failed after retries, will retry on next reconnect", "filter", telemetryTopicFilter)
 	})
 
 	client := mqtt.NewClient(opts)

internal/service/bridge/handler.go

@@ -5,13 +5,15 @@
 package bridge
 
 import (
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"log/slog"
+	"regexp"
 	"strings"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
-	"github.com/google/uuid"
 )
 
 // TelemetryEnvelope — сообщение, перекладываемое в SQS (разбирается consumer).
@@ -42,6 +44,13 @@ func newMessageHandler(dispatcher *sqsDispatcher, log *slog.Logger) mqtt.Message
 		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) {
@@ -55,7 +64,20 @@ func newMessageHandler(dispatcher *sqsDispatcher, log *slog.Logger) mqtt.Message
 			Topic:      topic,
 			Payload:    rawPayload,
 			ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano),
-			MessageID:  uuid.NewString(),
+			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[:])
+}

internal/service/bridge/sender.go

@@ -12,6 +12,7 @@ package bridge
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"log/slog"
 	"sync"
 	"sync/atomic"
@@ -19,6 +20,7 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/smithy-go"
 )
 
 // Константы диспатчера.
@@ -176,6 +178,12 @@ func (d *sqsDispatcher) sendWithRetry(ctx context.Context, body string) error {
 		if err == nil {
 			return nil
 		}
+		// Permanent ошибки (неверный параметр/размер/нет очереди/доступ) — ретрай
+		// бессмыслен: вернуть сразу, иначе воркер залип бы на poison-сообщении и
+		// заполнил буфер (дроп остальных).
+		if isPermanentSQSErr(err) {
+			return err
+		}
 		if ctx.Err() != nil {
 			// shutdown: ctx отменён, ретраить бессмысленно.
 			return err
@@ -190,3 +198,19 @@ func (d *sqsDispatcher) sendWithRetry(ctx context.Context, body string) error {
 		}
 	}
 }
+
+// isPermanentSQSErr — true для ошибок SQS, которые не пройдут ни при каком
+// ретрае (валидация, размер, отсутствие очереди, доступ). Остальное (throttling,
+// 5xx, сеть) — retryable.
+func isPermanentSQSErr(err error) bool {
+	var apiErr smithy.APIError
+	if errors.As(err, &apiErr) {
+		switch apiErr.ErrorCode() {
+		case "InvalidParameterValue", "InvalidMessageContents", "MessageTooBig",
+			"QueueDoesNotExist", "InvalidAddress", "InvalidSecurity",
+			"AccessDenied", "InvalidBatchEntryId":
+			return true
+		}
+	}
+	return false
+}