SQS-service/app/gosqs/change_message_visibility.go

// Изменено: 2026-04-11 — добавлена Redis persistence
// ChangeMessageVisibilityV1 — меняет visibility timeout сообщения в очереди тенанта.
package gosqs

import (
	"net/http"
	"strings"
	"time"

	"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 ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
	requestBody := models.NewChangeMessageVisibilityRequest()
	ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
	if !ok {
		log.Error("Invalid Request - ChangeMessageVisibilityV1")
		return utils.CreateErrorResponseV1("InvalidParameterValue", true)
	}

	t := getTenantFromContext(req)
	if t == nil {
		return utils.CreateErrorResponseV1("InvalidClientTokenId", true)
	}

	vars := mux.Vars(req)
	queueUrl := requestBody.QueueUrl
	queueName := ""
	if queueUrl == "" {
		queueName = vars["queueName"]
	} else {
		uriSegments := strings.Split(queueUrl, "/")
		queueName = uriSegments[len(uriSegments)-1]
	}

	key := tenantQueueKey(t.AccessKey, queueName)
	receiptHandle := requestBody.ReceiptHandle
	visibilityTimeout := requestBody.VisibilityTimeout

	if visibilityTimeout < 0 || visibilityTimeout > 43200 {
		return utils.CreateErrorResponseV1("InvalidParameterValue", true)
	}

	queue := models.SyncQueues.GetLocked(key)
	if queue == nil {
		return utils.CreateErrorResponseV1("QueueNotFound", true)
	}
	defer queue.UnlockMu()

	messageFound := false
	var changedMsgUuid string
	var msgRemoved bool
	for i := 0; i < len(queue.Messages); i++ {
		msgs := queue.Messages
		if msgs[i].ReceiptHandle == receiptHandle {
			timeout := queue.VisibilityTimeout
			if visibilityTimeout == 0 {
				msgs[i].ReceiptTime = time.Now().UTC()
				msgs[i].ReceiptHandle = ""
				msgs[i].VisibilityTimeout = time.Now().Add(time.Duration(timeout) * time.Second)
				msgs[i].Retry++
				if queue.MaxReceiveCount > 0 &&
					queue.DeadLetterQueue != nil &&
					msgs[i].Retry >= queue.MaxReceiveCount {
					changedMsgUuid = msgs[i].Uuid
					queue.DeadLetterQueue.LockMu()
					queue.DeadLetterQueue.Messages = append(queue.DeadLetterQueue.Messages, msgs[i])
					queue.DeadLetterQueue.UnlockMu()
					queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...)
					msgRemoved = true
				} else {
					changedMsgUuid = msgs[i].Uuid
				}
			} else {
				msgs[i].VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second)
				changedMsgUuid = msgs[i].Uuid
			}
			messageFound = true
			break
		}
	}
	// Персистим изменения в Redis под Lock (сетевые записи асинхронны).
	if messageFound {
		if msgRemoved {
			// Сообщение удалено (перемещено в DLQ) — удаляем из Redis
			persistence.DeleteMessagePersist(key, changedMsgUuid)
		} else {
			// Сообщение обновлено — перезаписываем в Redis
			for i := range queue.Messages {
				if queue.Messages[i].Uuid == changedMsgUuid {
					persistence.SaveMessage(key, &queue.Messages[i])
					break
				}
			}
		}
		persistence.SaveQueue(key, queue)
	}
	if !messageFound {
		return utils.CreateErrorResponseV1("MessageNotInFlight", true)
	}

	respStruct := models.ChangeMessageVisibilityResult{
		Xmlns:    models.BaseXmlns,
		Metadata: models.BaseResponseMetadata,
	}
	return http.StatusOK, &respStruct
}