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" log "github.com/sirupsen/logrus" "github.com/gorilla/mux" ) // SendMessageV1 — добавляет сообщение в очередь тенанта. // // Логическая схема: // 1. Разбор тела запроса (Query/JSON) → SendMessageRequest. // 2. Аутентифицированный тенант из контекста (SigV4 → AccessKey). // 3. Валидации: // - MessageBody обязателен (пустой → MissingParameter); // - DeduplicationId и GroupId ≤ 128 символов; // - MessageAttributes ≤ 10; // - FIFO-очередь БЕЗ MessageGroupId → MissingParameter (AWS требует GroupId // для каждой отправки в FIFO); // - размер тела ≤ MaximumMessageSize очереди → MessageTooBig; // - число сообщений < лимита очереди (OOM-защита) → OverLimit. // 4. Построение SqsMessage: MD5 тела и атрибутов, UUID, метки времени, задержка. // 5. Атомарно под Lock: // - FIFO: порядковый номер группы (NextSequenceNumber); // - проверка дубликата по deduplicationId (дубль НЕ добавляется); // - добавление в очередь + SaveMessage/SaveQueue в Redis. // Примечание: сетевые Redis-записи асинхронны (asyncWrite), под Lock // выполняется только json.Marshal — блокировка не удерживается на сетевом I/O. // 6. Ответ: MessageId, MD5OfMessageBody/MD5OfMessageAttributes, SequenceNumber (FIFO). func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { requestBody := models.NewSendMessageRequest() ok := utils.REQUEST_TRANSFORMER(requestBody, req, false) if !ok { log.Error("Invalid Request - SendMessageV1") return utils.CreateErrorResponseV1("InvalidParameterValue", true) } t := getTenantFromContext(req) if t == nil { return utils.CreateErrorResponseV1("InvalidClientTokenId", true) } messageBody := requestBody.MessageBody if messageBody == "" { return utils.CreateErrorResponseV1("MissingParameter", true) } messageGroupID := requestBody.MessageGroupId messageDeduplicationID := requestBody.MessageDeduplicationId // Валидация DeduplicationID и GroupID (макс 128 chars по AWS) if err := ValidateDeduplicationID(messageDeduplicationID); err != nil { return utils.CreateErrorResponseV1("InvalidParameterValue", true) } if err := ValidateGroupID(messageGroupID); err != nil { return utils.CreateErrorResponseV1("InvalidParameterValue", true) } // Валидация количества message attributes (макс 10) if len(requestBody.MessageAttributes) > MaxMessageAttributes { return utils.CreateErrorResponseV1("InvalidParameterValue", true) } queueUrl := getQueueFromPath(requestBody.QueueUrl, req.URL.String()) queueName := "" if queueUrl == "" { vars := mux.Vars(req) queueName = vars["queueName"] } else { // Ловушка #6: берём последний сегмент — это queueName, не tenantID uriSegments := strings.Split(queueUrl, "/") queueName = uriSegments[len(uriSegments)-1] } key := tenantQueueKey(t.AccessKey, queueName) queue := models.SyncQueues.GetLocked(key) if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } defer queue.UnlockMu() maxMessageSize := queue.MaximumMessageSize currentMsgCount := queue.LiveCount() queueIsFIFO := queue.IsFIFO contentBasedDedup := queue.ContentBasedDeduplication delaySecs := queue.DelaySeconds // FIFO-очередь БЕЗ MessageGroupId — ошибка: AWS требует GroupId для каждой // отправки в FIFO (иначе невозможно гарантировать порядок внутри группы). // Раньше сообщение просто попадало в группу с пустым именем. if queueIsFIFO && messageGroupID == "" { return utils.CreateErrorResponseV1("MissingParameter", true) } // ContentBasedDeduplication: если MessageDeduplicationId не передан, AWS сам // вычисляет его из тела сообщения (SHA-256). Без этого включённый флаг // ничего не делал бы и FIFO-дедупликация не работала. if queueIsFIFO && contentBasedDedup && messageDeduplicationID == "" { messageDeduplicationID = utils.GetSHA256Hash(messageBody) } if maxMessageSize > 0 && len(messageBody) > maxMessageSize { return utils.CreateErrorResponseV1("MessageTooBig", true) } // Fix #11: лимит сообщений в очереди — защита от OOM if currentMsgCount >= MaxMessagesForQueue(queueIsFIFO) { return utils.CreateErrorResponseV1("OverLimit", true) } if requestBody.DelaySeconds != 0 { delaySecs = requestBody.DelaySeconds } log.Debugf("Putting Message in Queue: [%s] tenant: [%s]", queueName, t.ID) msg := models.SqsMessage{MessageBody: messageBody} if len(requestBody.MessageAttributes) > 0 { msg.MessageAttributes = requestBody.MessageAttributes msg.MD5OfMessageAttributes = utils.HashAttributes(requestBody.MessageAttributes) } msg.MD5OfMessageBody = utils.GetMD5Hash(messageBody) msg.Uuid = uuid.NewString() msg.GroupID = messageGroupID msg.DeduplicationID = messageDeduplicationID msg.SentTime = time.Now() msg.DelaySecs = delaySecs // Повторная проверка лимита под per-queue Lock (между снимком выше и текущим // моментом другие горутины могли добавить сообщения — мягкий OOM-лимит). if queue.LiveCount() >= MaxMessagesForQueue(queue.IsFIFO) { return utils.CreateErrorResponseV1("OverLimit", true) } // Глобальный предохранитель памяти (HeapAlloc) — защита от OOM суммой тенантов. if heapOverLimit() { return utils.CreateErrorResponseV1("OverLimit", true) } fifoSeqNumber := "" persistErr := false if !queue.IsDuplicate(messageDeduplicationID) { idx := queue.AppendMessage(msg) // Персистим сообщение СИНХРОННО до ответа клиенту: краш пода в окне // между 200 OK и async-записью приводил бы к потере подтверждённого // сообщения. При ошибке — откатываем append (не оставляем в памяти // сообщение, которого нет в Redis). if err := persistence.SaveMessageSync(key, &msg); err != nil { log.Errorf("SendMessage: не удалось персистить сообщение %s в очередь %s: %v", msg.Uuid, key, err) queue.RemoveIndex(idx) persistErr = true } else { // Регистрируем dedupId и порядковый номер ТОЛЬКО при успешной записи. // Иначе ретрай с тем же dedupId получил бы 200 без сохранённого сообщения // (InitDuplicatation после отката append помечал бы id как обработанный). queue.InitDuplicatation(messageDeduplicationID) if queue.IsFIFO { fifoSeqNumber = queue.NextSequenceNumber(messageGroupID) } // Метаданные очереди (FIFO state, duplicates) — без Messages persistence.SaveQueue(key, queue) } } else { log.Debugf("Duplicate message deduplicationId [%s] in queue [%s]", messageDeduplicationID, queueName) } if persistErr { return utils.CreateErrorResponseV1("InternalError", true) } // Логируем только метаданные — тело сообщения не логируется (perf + security) log.Infof("Queue: %s, MessageId: %s, Size: %d bytes", queueName, msg.Uuid, len(messageBody)) respStruct := models.SendMessageResponse{ Xmlns: models.BaseXmlns, Result: models.SendMessageResult{ MD5OfMessageAttributes: msg.MD5OfMessageAttributes, MD5OfMessageBody: msg.MD5OfMessageBody, MessageId: msg.Uuid, SequenceNumber: fifoSeqNumber, }, Metadata: models.BaseResponseMetadata, } return http.StatusOK, respStruct }