112 lines
5.4 KiB
Go
112 lines
5.4 KiB
Go
package gosqs
|
||
|
||
import (
|
||
"net/http"
|
||
"time"
|
||
|
||
"shared-sqs/app/interfaces"
|
||
"shared-sqs/app/models"
|
||
"shared-sqs/app/persistence"
|
||
"shared-sqs/app/utils"
|
||
|
||
log "github.com/sirupsen/logrus"
|
||
)
|
||
|
||
// CreateQueueV1 — создаёт очередь для тенанта из request context.
|
||
//
|
||
// Логическая схема:
|
||
// 1. Разбор тела запроса (Query/JSON) → CreateQueueRequest.
|
||
// 2. Тенант из контекста (SigV4 → AccessKey).
|
||
// 3. Валидация имени очереди по AWS-правилам: до 80 символов, [a-zA-Z0-9_-],
|
||
// опциональный суффикс .fifo (ValidateQueueName).
|
||
// 4. Под Lock:
|
||
// - если очередь уже есть — просто возвращаем её URL (идемпотентность AWS);
|
||
// - проверка лимита очередей тенанта (MaxQueues) → LimitExceeded;
|
||
// - создание очереди с дефолтами и применение переданных атрибутов
|
||
// (setQueueAttributesV1 с валидацией диапазонов);
|
||
// - вставка в map с tenant-scoped ключом "{tenantAccessKey}:{queueName}".
|
||
// 5. Сохранение очереди в Redis (асинхронно, маршалинг под Lock).
|
||
// 6. Ответ: QueueUrl.
|
||
func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||
requestBody := models.NewCreateQueueRequest()
|
||
ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
|
||
if !ok {
|
||
log.Error("Invalid Request - CreateQueueV1")
|
||
return utils.CreateErrorResponseV1("InvalidParameterValue", true)
|
||
}
|
||
|
||
t := getTenantFromContext(req)
|
||
if t == nil {
|
||
return utils.CreateErrorResponseV1("InvalidClientTokenId", true)
|
||
}
|
||
|
||
// Валидация имени очереди по AWS правилам: макс 80 chars, [a-zA-Z0-9_-], опц. .fifo
|
||
queueName := requestBody.QueueName
|
||
if err := ValidateQueueName(queueName); err != nil {
|
||
return utils.CreateErrorResponseV1("InvalidParameterValue", true)
|
||
}
|
||
key := tenantQueueKey(t.AccessKey, queueName)
|
||
queueUrl := tenantQueueURL(t, queueName)
|
||
queueArn := tenantQueueARN(t, queueName)
|
||
|
||
// Готовим очередь и применяем атрибуты ВНЕ глобального лока: setQueueAttributesV1
|
||
// берёт SyncQueues.RLock (поиск DLQ), а RWMutex не реентрантен — вызов под
|
||
// уже захваченным SyncQueues.Lock дал бы дедлок.
|
||
queue := &models.Queue{
|
||
Name: queueName,
|
||
URL: queueUrl,
|
||
Arn: queueArn,
|
||
IsFIFO: utils.HasFIFOQueueName(queueName),
|
||
EnableDuplicates: models.CurrentEnvironment.EnableDuplicates,
|
||
Duplicates: make(map[string]time.Time),
|
||
// Дефолты (как в InitQueues): без них очередь из API получала бы
|
||
// VisibilityTimeout=0 и т.д. — сообщение сразу видимо, retention 0 = нет
|
||
// эвикции. Переданные атрибуты перезапишут дефолты в setQueueAttributesV1.
|
||
VisibilityTimeout: models.CurrentEnvironment.QueueAttributeDefaults.VisibilityTimeout,
|
||
MaximumMessageSize: models.CurrentEnvironment.QueueAttributeDefaults.MaximumMessageSize,
|
||
MessageRetentionPeriod: models.CurrentEnvironment.QueueAttributeDefaults.MessageRetentionPeriod,
|
||
ReceiveMessageWaitTimeSeconds: models.CurrentEnvironment.QueueAttributeDefaults.ReceiveMessageWaitTimeSeconds,
|
||
// CreatedTimestamp/LastModifiedTimestamp в GetQueueAttributes (AWS их отдаёт).
|
||
CreatedAt: time.Now().UTC(),
|
||
LastModifiedAt: time.Now().UTC(),
|
||
}
|
||
// Список имён переданных атрибутов — для whitelist-валидации
|
||
// (Query-протокол: Attribute.N.Name; JSON: nil — проверка пропускается).
|
||
var provided map[string]string
|
||
if req.Header.Get("Content-Type") != "application/x-amz-json-1.0" {
|
||
provided = utils.ExtractQueueAttributes(req.PostForm)
|
||
}
|
||
if err := setQueueAttributesV1(queue, requestBody.Attributes, provided, t.AccessKey); err != nil {
|
||
return utils.CreateErrorResponseV1(err.Error(), true)
|
||
}
|
||
|
||
models.SyncQueues.Lock()
|
||
existing, exists := models.SyncQueues.Queues[key]
|
||
if !exists {
|
||
// Проверка лимита очередей тенанта
|
||
if t.MaxQueues > 0 && countTenantQueues(t.AccessKey) >= t.MaxQueues {
|
||
models.SyncQueues.Unlock()
|
||
return utils.CreateErrorResponseV1("LimitExceeded", true)
|
||
}
|
||
log.Infof("Creating Queue: %s (tenant: %s)", queueName, t.ID)
|
||
models.SyncQueues.AddQueue(key, queue)
|
||
existing = queue
|
||
}
|
||
models.SyncQueues.Unlock()
|
||
|
||
// SaveQueue маршалит map'ы очереди (Duplicates/FIFO*/Tags) в JSON. Это нужно
|
||
// делать под per-queue блокировкой, а НЕ под глобальным SyncQueues.Lock:
|
||
// мутаторы (send/tag/CMV) пишут в эти map под queue.mu, и идемпотентный
|
||
// CreateQueue существующей очереди без queue.mu даёт fatal
|
||
// "concurrent map iteration and map write" (краш всего процесса).
|
||
existing.LockMu()
|
||
persistence.SaveQueue(key, existing)
|
||
existing.UnlockMu()
|
||
respStruct := models.CreateQueueResponse{
|
||
Xmlns: models.BaseXmlns,
|
||
Result: models.CreateQueueResult{QueueUrl: queueUrl},
|
||
Metadata: models.BaseResponseMetadata,
|
||
}
|
||
return http.StatusOK, respStruct
|
||
}
|