From 4031882caece59520a3967eff0b24b42741d4578 Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Wed, 7 Oct 2026 07:40:09 +0300
Subject: [PATCH] =?UTF-8?q?feat(consumer):=20=D0=B1=D0=B8=D1=82=D1=8B?=
 =?UTF-8?q?=D0=B5=20envelope=20=E2=86=92=20DLQ=20=D0=B2=D0=BC=D0=B5=D1=81?=
 =?UTF-8?q?=D1=82=D0=BE=20=D0=BF=D0=BE=D1=82=D0=B5=D1=80=D0=B8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

processTelemetry теперь возвращает sentinel errInvalidEnvelope при битом JSON
(раньше — молчаливый nil → mусор подтверждался и терялся). Consumer при этой
ошибке кладёт тело в DLQ (SQS_DLQ_NAME, дефолт iot-telemetry-dlq) и подтверждает
основное сообщение; без DLQ — только логирует, но НЕ подтверждает (оседает).
Добавлены SQSDLQName в config + DefaultSQSDLQName. Реальные ошибки PG (не invalid)
по-прежнему не подтверждаются и уходят в ретрай после visibility timeout.
---
 internal/service/config/config.go      |  2 ++
 internal/service/config/defaults.go    |  5 ++++
 internal/service/consumer/consumer.go  | 33 +++++++++++++++++++++++++-
 internal/service/consumer/processor.go | 12 +++++++---
 4 files changed, 48 insertions(+), 4 deletions(-)

diff --git a/internal/service/config/config.go b/internal/service/config/config.go
index 8d2b0f1..5b1aaee 100644
--- a/internal/service/config/config.go
+++ b/internal/service/config/config.go
@@ -26,6 +26,7 @@ type Config struct {
 	SQSAccessKey string
 	SQSSecretKey string
 	SQSQueueName string
+	SQSDLQName   string
 	SQSRegion    string
 
 	// SQS-потребитель
@@ -109,6 +110,7 @@ func Load() (*Config, error) {
 		SQSAccessKey:         os.Getenv("SQS_ACCESS_KEY"),
 		SQSSecretKey:         os.Getenv("SQS_SECRET_KEY"),
 		SQSQueueName:         getEnv("SQS_QUEUE_NAME", DefaultSQSQueueName),
+		SQSDLQName:           getEnv("SQS_DLQ_NAME", DefaultSQSDLQName),
 		SQSRegion:            getEnv("SQS_REGION", DefaultSQSRegion),
 		SQSLongPollSeconds:   getEnvInt("SQS_LONG_POLL_SECONDS", 20),
 		SQSVisibilityTimeout: getEnvInt("SQS_VISIBILITY_TIMEOUT", 120),
diff --git a/internal/service/config/defaults.go b/internal/service/config/defaults.go
index 0271db0..d3367dd 100644
--- a/internal/service/config/defaults.go
+++ b/internal/service/config/defaults.go
@@ -13,6 +13,11 @@ const (
 	// DefaultSQSQueueName — очередь телеметрии IoT.
 	DefaultSQSQueueName = "iot-telemetry"
 
+	// DefaultSQSDLQName — DLQ для «мёртвых» сообщений: битые envelope, которые
+	// нельзя разобрать (повторная обработка бессмысленна). Валидные данные
+	// идут в основную очередь; неразборчивые — сюда, а не отбрасываются.
+	DefaultSQSDLQName = "iot-telemetry-dlq"
+
 	// DefaultSQSRegion — регион для self-hosted SQS (не важен, оставлен для SDK).
 	DefaultSQSRegion = "us-east-1"
 
diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go
index 572552f..4a640d7 100644
--- a/internal/service/consumer/consumer.go
+++ b/internal/service/consumer/consumer.go
@@ -6,6 +6,7 @@ package consumer
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"time"
 
@@ -29,7 +30,14 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 	if err != nil {
 		return err
 	}
-	log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName)
+	// DLQ присутствует не всегда: резолвим лениво, отказ не фатален (битые
+	// сообщения тогда просто НЕ подтверждаются и оседают после ретраев).
+	dlqURL, dlqErr := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSDLQName)
+	if dlqErr != nil {
+		log.Warn("consumer: DLQ не найдена, битые сообщения будут только логироваться", "dlq", cfg.SQSDLQName, "err", dlqErr)
+		dlqURL = ""
+	}
+	log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName, "dlq", cfg.SQSDLQName)
 
 	for {
 		if ctx.Err() != nil {
@@ -70,6 +78,20 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 				continue
 			}
 			if err := processTelemetry(ctx, *msg.Body, store, log); err != nil {
+				// Битый envelope → DLQ (а не бесконечный ретрай и не потеря):
+				// кладём тело в DLQ, затем подтверждаем основное сообщение.
+				if errors.Is(err, errInvalidEnvelope) {
+					if dlqURL != "" {
+						if derr := sendToDLQ(ctx, sqsClient, dlqURL, *msg.Body, log); derr != nil {
+							log.Error("consumer: send to DLQ failed", "err", derr, "message_id", aws.ToString(msg.MessageId))
+						}
+					} else {
+						log.Warn("consumer: invalid envelope dropped (DLQ отсутствует)", "message_id", aws.ToString(msg.MessageId))
+					}
+					// Подтверждаем: DLQ-отправка (или логирование) — терминальная обработка.
+					pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
+					continue
+				}
 				log.Error("consumer: process telemetry", "err", err, "message_id", aws.ToString(msg.MessageId))
 				// Сообщение НЕ подтверждаем — вернётся после visibility timeout.
 				// Backoff с jitter — не долбить упавший PG в цикле
@@ -115,6 +137,15 @@ func deleteMessage(ctx context.Context, client *sqs.Client, queueURL string, rec
 	return err
 }
 
+// sendToDLQ — кладёт битое неразборчивое сообщение в DLQ для диагностики.
+func sendToDLQ(ctx context.Context, client *sqs.Client, dlqURL, body string, log *slog.Logger) error {
+	_, err := client.SendMessage(ctx, &sqs.SendMessageInput{
+		QueueUrl:    aws.String(dlqURL),
+		MessageBody: aws.String(body),
+	})
+	return err
+}
+
 // sleepCtx — ожидание с прерыванием по ctx. false = ctx завершён.
 func sleepCtx(ctx context.Context, d time.Duration) bool {
 	select {
diff --git a/internal/service/consumer/processor.go b/internal/service/consumer/processor.go
index 3e6f6af..32a0f39 100644
--- a/internal/service/consumer/processor.go
+++ b/internal/service/consumer/processor.go
@@ -4,19 +4,25 @@ package consumer
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"log/slog"
 
 	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/bridge"
 	"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
 )
 
+// errInvalidEnvelope — битый/неразборчивый envelope: повторная обработка
+// бессмысленна, сообщение следует отправить в DLQ, а не ретраить бесконечно.
+var errInvalidEnvelope = errors.New("invalid telemetry envelope")
+
 // processTelemetry десериализует envelope и пишет в tenant DB.
-// Битый JSON — пропускаем (не блокируем очередь).
+// Битый JSON — возвращает errInvalidEnvelope (consumer шлёт в DLQ),
+// а НЕ молчаливый nil: иначе мусор подтверждается и теряется безвозвратно.
 func processTelemetry(ctx context.Context, body string, store *iotpg.IoTPostgresStore, log *slog.Logger) error {
 	var envelope bridge.TelemetryEnvelope
 	if err := json.Unmarshal([]byte(body), &envelope); err != nil {
-		log.Warn("consumer: failed to unmarshal envelope, skipping", "err", err)
-		return nil
+		log.Warn("consumer: invalid envelope, routing to DLQ", "err", err)
+		return errInvalidEnvelope
 	}
 
 	// EnsureTenantDB идемпотентен, кэшируется после первого вызова.