// Изменено: 2026-04-11 — убрано логирование тела сообщения (perf + security) // SendMessageBatchV1 — пакетная отправка сообщений в очередь тенанта. package gosqs import ( "net/http" "strings" "time" "github.com/google/uuid" "shared-sqs/app/interfaces" "shared-sqs/app/models" "shared-sqs/app/persistence" "shared-sqs/app/utils" "github.com/gorilla/mux" log "github.com/sirupsen/logrus" ) func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody) { requestBody := models.NewSendMessageBatchRequest() ok := utils.REQUEST_TRANSFORMER(requestBody, req, false) if !ok { log.Error("Invalid Request - SendMessageBatchV1") return utils.CreateErrorResponseV1("InvalidParameterValue", true) } t := getTenantFromContext(req) if t == nil { return utils.CreateErrorResponseV1("InvalidClientTokenId", true) } queueUrl := requestBody.QueueUrl queueName := "" if queueUrl == "" { vars := mux.Vars(req) queueName = vars["queueName"] } else { uriSegments := strings.Split(queueUrl, "/") queueName = uriSegments[len(uriSegments)-1] } key := tenantQueueKey(t.AccessKey, queueName) queue := models.SyncQueues.GetRLocked(key) if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } maxMsgSize := queue.MaximumMessageSize currentMsgCount := queue.LiveCount() queueIsFIFO := queue.IsFIFO queue.RUnlockMu() sendEntries := requestBody.Entries if len(sendEntries) == 0 { return utils.CreateErrorResponseV1("EmptyBatchRequest", true) } if len(sendEntries) > 10 { return utils.CreateErrorResponseV1("TooManyEntriesInBatchRequest", true) } ids := map[string]struct{}{} for _, v := range sendEntries { if _, ok := ids[v.Id]; ok { return utils.CreateErrorResponseV1("BatchEntryIdsNotDistinct", true) } // Валидация длины BatchEntryId (макс 80 chars) if err := ValidateBatchEntryID(v.Id); err != nil { return utils.CreateErrorResponseV1("InvalidParameterValue", true) } // Валидация DeduplicationID и GroupID (макс 128 chars) if err := ValidateDeduplicationID(v.MessageDeduplicationId); err != nil { return utils.CreateErrorResponseV1("InvalidParameterValue", true) } if err := ValidateGroupID(v.MessageGroupId); err != nil { return utils.CreateErrorResponseV1("InvalidParameterValue", true) } ids[v.Id] = struct{}{} } sentEntries := make([]models.SendMessageBatchResultEntry, 0) log.Debugf("Batch sending to Queue: %s (tenant: %s)", queueName, t.ID) // Fix #2: проверяем размер каждого сообщения в batch (Critical — batch size bypass) if maxMsgSize <= 0 { maxMsgSize = MaxMessageSizeDefault } for _, entry := range sendEntries { if len(entry.MessageBody) > maxMsgSize { return utils.CreateErrorResponseV1("MessageTooBig", true) } // Валидация количества message attributes (макс 10 по AWS) if len(entry.MessageAttributes) > MaxMessageAttributes { return utils.CreateErrorResponseV1("InvalidParameterValue", true) } } // Fix #11: лимит сообщений в очереди — защита от OOM if currentMsgCount+len(sendEntries) > MaxMessagesForQueue(queueIsFIFO) { return utils.CreateErrorResponseV1("OverLimit", true) } queue.LockMu() defer queue.UnlockMu() // Повторная проверка лимита под per-queue Lock: снимок currentMsgCount выше // был сделан под RLock, между ним и взятием Lock очередь могла измениться // (гонка TOCTOU). Здесь — обязательная проверка по свежему числу живых. if queue.LiveCount()+len(sendEntries) > MaxMessagesForQueue(queue.IsFIFO) { return utils.CreateErrorResponseV1("OverLimit", true) } // Глобальный предохранитель памяти (HeapAlloc) — защита от OOM суммой тенантов. if heapOverLimit() { return utils.CreateErrorResponseV1("OverLimit", true) } newMsgs := make([]models.SqsMessage, 0, len(sendEntries)) appendedIdx := make([]int, 0, len(sendEntries)) appendedDedupIDs := make([]string, 0, len(sendEntries)) for _, sendEntry := range sendEntries { msg := models.SqsMessage{MessageBody: sendEntry.MessageBody} if len(sendEntry.MessageAttributes) > 0 { msg.MessageAttributes = sendEntry.MessageAttributes msg.MD5OfMessageAttributes = utils.HashAttributes(sendEntry.MessageAttributes) } msg.MD5OfMessageBody = utils.GetMD5Hash(sendEntry.MessageBody) msg.GroupID = sendEntry.MessageGroupId // ContentBasedDeduplication: dedupId вычисляется из тела (SHA-256), // если не передан явно (AWS-семантика FIFO). dedupID := sendEntry.MessageDeduplicationId if queue.IsFIFO && queue.ContentBasedDeduplication && dedupID == "" { dedupID = utils.GetSHA256Hash(sendEntry.MessageBody) } msg.DeduplicationID = dedupID msg.Uuid = uuid.NewString() msg.SentTime = time.Now() fifoSeqNumber := "" if queue.IsFIFO { fifoSeqNumber = queue.NextSequenceNumber(sendEntry.MessageGroupId) } if !queue.IsDuplicate(dedupID) { appendedIdx = append(appendedIdx, queue.AppendMessage(msg)) newMsgs = append(newMsgs, msg) appendedDedupIDs = append(appendedDedupIDs, dedupID) } else { log.Debugf("Duplicate deduplicationId [%s] in queue [%s]", dedupID, queueName) } sentEntries = append(sentEntries, models.SendMessageBatchResultEntry{ Id: sendEntry.Id, MessageId: msg.Uuid, MD5OfMessageBody: msg.MD5OfMessageBody, MD5OfMessageAttributes: msg.MD5OfMessageAttributes, SequenceNumber: fifoSeqNumber, }) // Логируем только метаданные — тело сообщения не логируется (perf + security) log.Debugf("Queue: %s, MessageId: %s, Size: %d bytes", queueName, msg.Uuid, len(sendEntry.MessageBody)) } // Персистим СИНХРОННО до ответа (как одиночный 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) } // Регистрируем dedupId ТОЛЬКО после успешной записи — иначе ретрай с теми же // dedupId получил бы 200 без сохранённых сообщений. for _, id := range appendedDedupIDs { queue.InitDuplicatation(id) } // Метаданные очереди (FIFO state, duplicates) — без Messages persistence.SaveQueue(key, queue) respStruct := models.SendMessageBatchResponse{ Xmlns: models.BaseXmlns, Result: models.SendMessageBatchResult{Entry: sentEntries}, Metadata: models.BaseResponseMetadata, } return http.StatusOK, respStruct }