From a341508b6d73eab40d7e2a2aef99aa03fc980321 Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Sat, 10 Oct 2026 16:18:54 +0300
Subject: [PATCH] =?UTF-8?q?fix(batch):=20SendMessageBatch=20=D0=BF=D0=B8?=
 =?UTF-8?q?=D1=88=D0=B5=D1=82=20=D0=B2=20Redis=20=D1=81=D0=B8=D0=BD=D1=85?=
 =?UTF-8?q?=D1=80=D0=BE=D0=BD=D0=BD=D0=BE=20(=D0=BA=D0=B0=D0=BA=20SendMess?=
 =?UTF-8?q?age)=20+=20=D0=BD=D0=B5=20=D0=BF=D0=B5=D1=80=D1=81=D0=B8=D1=81?=
 =?UTF-8?q?=D1=82=D0=B8=D1=82=20=D0=B4=D1=83=D0=B1=D0=BB=D0=B8;=20SaveMess?=
 =?UTF-8?q?agesSync?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/gosqs/send_message_batch.go | 16 ++++++++++++----
 app/persistence/redis.go        | 31 +++++++++++++++++++++++++++++++
 2 files changed, 43 insertions(+), 4 deletions(-)

diff --git a/app/gosqs/send_message_batch.go b/app/gosqs/send_message_batch.go
index a635845..3fa6d20 100644
--- a/app/gosqs/send_message_batch.go
+++ b/app/gosqs/send_message_batch.go
@@ -113,6 +113,7 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
 	}
 
 	newMsgs := make([]models.SqsMessage, 0, len(sendEntries))
+	appendedIdx := make([]int, 0, len(sendEntries))
 	for _, sendEntry := range sendEntries {
 		msg := models.SqsMessage{MessageBody: sendEntry.MessageBody}
 		if len(sendEntry.MessageAttributes) > 0 {
@@ -130,7 +131,8 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
 			fifoSeqNumber = queue.NextSequenceNumber(sendEntry.MessageGroupId)
 		}
 		if !queue.IsDuplicate(sendEntry.MessageDeduplicationId) {
-			queue.AppendMessage(msg)
+			appendedIdx = append(appendedIdx, queue.AppendMessage(msg))
+			newMsgs = append(newMsgs, msg)
 		} else {
 			log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", sendEntry.MessageDeduplicationId, queueName)
 		}
@@ -143,12 +145,18 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody
 			MD5OfMessageAttributes: msg.MD5OfMessageAttributes,
 			SequenceNumber:         fifoSeqNumber,
 		})
-		newMsgs = append(newMsgs, msg)
 		// Логируем только метаданные — тело сообщения не логируется (perf + security)
 		log.Debugf("Queue: %s, MessageId: %s, Size: %d bytes", queueName, msg.Uuid, len(sendEntry.MessageBody))
 	}
-	// Персистим новые сообщения отдельно — O(batch_size*msg_size) вместо O(N*msg_size)
-	persistence.SaveMessages(key, newMsgs)
+	// Персистим СИНХРОННО до ответа (как одиночный SendMessage): иначе краш в окне
+	// между 200 OK и async-записью потеряет подтверждённые сообщения.
+	if err := persistence.SaveMessagesSync(key, newMsgs); err != nil {
+		log.Errorf("SendMessageBatch: не удалось персистить батч в %s: %v", key, err)
+		for _, idx := range appendedIdx {
+			queue.RemoveIndex(idx)
+		}
+		return utils.CreateErrorResponseV1("InternalError", true)
+	}
 	// Метаданные очереди (FIFO state, duplicates) — без Messages
 	persistence.SaveQueue(key, queue)
 
diff --git a/app/persistence/redis.go b/app/persistence/redis.go
index ab87379..8ae3c62 100644
--- a/app/persistence/redis.go
+++ b/app/persistence/redis.go
@@ -314,6 +314,37 @@ func SaveMessages(queueKey string, msgs []models.SqsMessage) {
 	})
 }
 
+// SaveMessagesSync — сохраняет несколько сообщений в Redis СИНХРОННО (pipeline).
+// Используется на пути SendMessageBatch: ответ 200 выдаётся только после записи
+// всех сообщений (как в одиночном SendMessage). Возвращает ошибку для отката.
+func SaveMessagesSync(queueKey string, msgs []models.SqsMessage) error {
+	if Client == nil || len(msgs) == 0 {
+		return nil
+	}
+	fields := make(map[string]string, len(msgs))
+	for i := range msgs {
+		data, err := json.Marshal(&msgs[i])
+		if err != nil {
+			return fmt.Errorf("marshal message %s/%s: %w", queueKey, msgs[i].Uuid, err)
+		}
+		fields[msgs[i].Uuid] = string(data)
+	}
+	if len(fields) == 0 {
+		return nil
+	}
+	hashKey := redisMsgHashPrefix + queueKey
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	args := make([]interface{}, 0, len(fields)*2)
+	for k, v := range fields {
+		args = append(args, k, v)
+	}
+	if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil {
+		return fmt.Errorf("HSet messages %s (%d msgs): %w", queueKey, len(fields), err)
+	}
+	return nil
+}
+
 // DeleteMessagePersist — удаляет одно сообщение из Redis СИНХРОННО.
 // Синхронность обязательна: если удаление потеряется при рестарте пода,
 // сообщение «воскреснет» из Redis (факт: удалённые очереди возвращались