From 31756059314938d406d0e3c5e6df5afcdf172d43 Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Wed, 7 Oct 2026 07:23:25 +0300
Subject: [PATCH] =?UTF-8?q?P0-7:=20=D0=BF=D0=BE=D0=B4=D1=82=D0=B2=D0=B5?=
 =?UTF-8?q?=D1=80=D0=B6=D0=B4=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=B8=D0=B7=20SQS?=
 =?UTF-8?q?=20=D1=82=D0=BE=D0=BB=D1=8C=D0=BA=D0=BE=20=D0=BF=D0=BE=D1=81?=
 =?UTF-8?q?=D0=BB=D0=B5=20=D1=83=D1=81=D0=BF=D0=B5=D1=88=D0=BD=D0=BE=D0=B3?=
 =?UTF-8?q?=D0=BE=20flush=20=D1=82=D0=B5=D0=BB=D0=B5=D0=BC=D0=B5=D1=82?=
 =?UTF-8?q?=D1=80=D0=B8=D0=B8=20=D0=B2=20=D0=91=D0=94?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

InsertTelemetry лишь буферизует и возвращает nil до фактической записи. Consumer
удалял сообщение сразу после processTelemetry == nil — краш/flush-ошибка в этом
окне = безвозвратная потеря. Теперь: копим receipt handles, вызываем store.Flush,
и только при успехе DeleteMessage; при ошибке flush — не удаляем (сообщения
вернутся после visibility timeout). Добавлен iotpg.Flush().
---
 internal/service/consumer/consumer.go         | 34 ++++++++++++++-----
 internal/storage/iotpg/iot_telemetry_store.go |  8 +++++
 2 files changed, 34 insertions(+), 8 deletions(-)

diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go
index ba22a9b..5d46b4d 100644
--- a/internal/service/consumer/consumer.go
+++ b/internal/service/consumer/consumer.go
@@ -54,18 +54,24 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 		}
 
 		processBackoff := time.Second
+		// Копим подтверждения, но НЕ удаляем из SQS сразу: InsertTelemetry лишь
+		// буферизует и возвращает nil до фактической записи. Удаляем только после
+		// успешного Flush — иначе краш/flush-ошибка = потеря подтверждённых
+		// сообщений (P0-7).
+		type pendingAck struct {
+			handle *string
+			id     string
+		}
+		var pending []pendingAck
 		for _, msg := range resp.Messages {
 			if msg.Body == nil || *msg.Body == "" {
-				log.Warn("consumer: empty message body, deleting",
-					"message_id", aws.ToString(msg.MessageId))
-				if err := deleteMessage(ctx, sqsClient, queueURL, msg.ReceiptHandle, log); err != nil {
-					log.Error("consumer: DeleteMessage failed", "err", err, "message_id", aws.ToString(msg.MessageId))
-				}
+				log.Warn("consumer: empty message body", "message_id", aws.ToString(msg.MessageId))
+				pending = append(pending, pendingAck{msg.ReceiptHandle, aws.ToString(msg.MessageId)})
 				continue
 			}
 			if err := processTelemetry(ctx, *msg.Body, store, log); err != nil {
 				log.Error("consumer: process telemetry", "err", err, "message_id", aws.ToString(msg.MessageId))
-				// Сообщение вернётся в очередь после visibility timeout.
+				// Сообщение НЕ подтверждаем — вернётся после visibility timeout.
 				// Backoff с jitter — не долбить упавший PG в цикле
 				// (фикс HIGH из ревью 2026-08-16: лавина ретраев).
 				if !sleepCtx(ctx, processBackoff) {
@@ -78,8 +84,20 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 				continue
 			}
 			processBackoff = time.Second
-			if err := deleteMessage(ctx, sqsClient, queueURL, msg.ReceiptHandle, log); err != nil {
-				log.Error("consumer: DeleteMessage failed", "err", err, "message_id", aws.ToString(msg.MessageId))
+			pending = append(pending, pendingAck{msg.ReceiptHandle, aws.ToString(msg.MessageId)})
+		}
+
+		if len(pending) == 0 {
+			continue
+		}
+		// Сначала гарантируем запись в БД, только потом удаляем из SQS.
+		if err := store.Flush(ctx); err != nil {
+			log.Error("consumer: flush failed — сообщения НЕ удалены, вернутся после visibility timeout", "err", err)
+			continue
+		}
+		for _, p := range pending {
+			if err := deleteMessage(ctx, sqsClient, queueURL, p.handle, log); err != nil {
+				log.Error("consumer: DeleteMessage failed", "err", err, "message_id", p.id)
 			}
 		}
 	}
diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go
index dd71fa5..3ecf29d 100644
--- a/internal/storage/iotpg/iot_telemetry_store.go
+++ b/internal/storage/iotpg/iot_telemetry_store.go
@@ -141,6 +141,14 @@ func (s *IoTPostgresStore) flushAll(reportErr bool) error {
 	return firstErr
 }
 
+// Flush — принудительно синхронно пишет все накопленные батчи телеметрии в БД и
+// возвращает первую ошибку. Нужен consumer'у, чтобы подтверждать (удалять из SQS)
+// ТОЛЬКО фактически записанные данные: без этого краш/flush-ошибка после
+// InsertTelemetry (которая лишь буферизует и возвращает nil) = потеря сообщений.
+func (s *IoTPostgresStore) Flush(ctx context.Context) error {
+	return s.flushAll(true)
+}
+
 // flushTenant — INSERT батча строк одного тенанта (multi-VALUES).
 func (s *IoTPostgresStore) flushTenant(ns string, rows []telemetryInsert) error {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)