SQS-service/app/gosqs/send_message.go

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 || requestBody.DelaySeconds > 900 {
		return utils.CreateErrorResponseV1("InvalidParameterValue", 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
}