776 lines
29 KiB
Go
776 lines
29 KiB
Go
// app/admin/admin.go
|
||
// Admin API handlers for shared-sqs management
|
||
// Created: 2026-04-09
|
||
// Updated: 2026-04-12 10:12 MSK — пометки о временном demo showcase режиме
|
||
package admin
|
||
|
||
import (
|
||
"context"
|
||
"crypto/subtle"
|
||
"encoding/json"
|
||
"net/http"
|
||
"os"
|
||
"strings"
|
||
"time"
|
||
|
||
"shared-sqs/app/auth"
|
||
"shared-sqs/app/gosqs"
|
||
"shared-sqs/app/models"
|
||
"shared-sqs/app/persistence"
|
||
"shared-sqs/app/tenant"
|
||
|
||
"github.com/gorilla/mux"
|
||
log "github.com/sirupsen/logrus"
|
||
)
|
||
|
||
type uiContextKey string
|
||
|
||
const (
|
||
uiTenantContextKey uiContextKey = "ui-tenant"
|
||
)
|
||
|
||
// ─── вспомогательная функция: найти очередь тенанта по имени ───────────────
|
||
// findQueue — возвращает ключ и очередь тенанта по имени, или "",nil если не найдено.
|
||
func findQueue(tenantAccessKey, queueName string) (string, *models.Queue) {
|
||
key := tenantAccessKey + ":" + queueName
|
||
models.SyncQueues.RLock()
|
||
q, ok := models.SyncQueues.Queues[key]
|
||
models.SyncQueues.RUnlock()
|
||
if !ok {
|
||
return "", nil
|
||
}
|
||
return key, q
|
||
}
|
||
|
||
// Handler — admin API handler, holds TenantStore и admin token
|
||
type Handler struct {
|
||
store *tenant.TenantStore
|
||
adminToken string
|
||
nubesEndpoint string // URL nubes API для валидации JWT (напр. https://deck-api-test.ngcloud.ru/api/v1)
|
||
}
|
||
|
||
// NewHandler — создаёт admin handler
|
||
func NewHandler(store *tenant.TenantStore, adminToken string) *Handler {
|
||
nubesEndpoint := os.Getenv("NUBES_ENDPOINT")
|
||
// Если не задан — пустой: PingNubesAPI переберёт все стенды сам.
|
||
auth.StartTokenCacheCleanup() // фоновая очистка кэша валидации (идемпотентно)
|
||
return &Handler{store: store, adminToken: adminToken, nubesEndpoint: nubesEndpoint}
|
||
}
|
||
|
||
// currentUITenant — возвращает tenant, авторизованный через UI middleware.
|
||
func currentUITenant(r *http.Request) (*tenant.Tenant, bool) {
|
||
t, ok := r.Context().Value(uiTenantContextKey).(*tenant.Tenant)
|
||
return t, ok
|
||
}
|
||
|
||
// tenantListItemFromTenant — строит публичный JSON-ответ без SecretKey.
|
||
func tenantListItemFromTenant(t *tenant.Tenant) tenantListItem {
|
||
return tenantListItem{
|
||
ID: t.ID,
|
||
Name: t.Name,
|
||
AccessKey: t.AccessKey,
|
||
MaxQueues: t.MaxQueues,
|
||
CreatedAt: t.CreatedAt,
|
||
Active: t.Active,
|
||
}
|
||
}
|
||
|
||
// RegisterRoutes — регистрирует admin маршруты на переданном router (с bearer auth)
|
||
func (h *Handler) RegisterRoutes(r *mux.Router) {
|
||
adminRouter := r.PathPrefix("/admin").Subrouter()
|
||
adminRouter.Use(h.bearerAuthMiddleware)
|
||
adminRouter.HandleFunc("/tenants", h.createTenant).Methods("POST")
|
||
adminRouter.HandleFunc("/tenants", h.listTenants).Methods("GET")
|
||
adminRouter.HandleFunc("/tenants/{id}", h.getTenant).Methods("GET")
|
||
adminRouter.HandleFunc("/tenants/{id}", h.deleteTenant).Methods("DELETE")
|
||
adminRouter.HandleFunc("/tenants/{id}/queues", h.listTenantQueues).Methods("GET")
|
||
adminRouter.HandleFunc("/tenants/{id}/queues", h.createTenantQueue).Methods("POST")
|
||
adminRouter.HandleFunc("/tenants/{id}/queues/{queue}", h.deleteTenantQueue).Methods("DELETE")
|
||
adminRouter.HandleFunc("/tenants/{id}/queues/{queue}/messages", h.peekQueueMessages).Methods("GET")
|
||
adminRouter.HandleFunc("/tenants/{id}/queues/{queue}/messages", h.sendMessageToQueue).Methods("POST")
|
||
adminRouter.HandleFunc("/tenants/{id}/queues/{queue}/messages", h.purgeQueue).Methods("DELETE")
|
||
adminRouter.HandleFunc("/health", h.detailedHealth).Methods("GET")
|
||
}
|
||
|
||
// RegisterPublicRoutes — маршруты для UI console.
|
||
// POST /ui/api/auth — единственный публичный endpoint (принимает JWT, возвращает session).
|
||
// Остальные /ui/api/* — защищены JWT middleware (токен в Authorization header).
|
||
// ВАЖНО: все UI API routes на ОДНОМ subrouter PathPrefix("/ui/api") — иначе через nginx ingress
|
||
// root-router HandleFunc конфликтует с PathPrefix subrouter (404 на POST).
|
||
func (h *Handler) RegisterPublicRoutes(r *mux.Router) {
|
||
ui := r.PathPrefix("/ui/api").Subrouter()
|
||
// jwtMiddleware пропускает /ui/api/auth — единственный публичный endpoint
|
||
ui.Use(h.jwtMiddleware)
|
||
ui.HandleFunc("/auth", h.jwtAuth).Methods("POST")
|
||
ui.HandleFunc("/credentials", h.uiCredentials).Methods("GET")
|
||
ui.HandleFunc("/health", h.detailedHealth).Methods("GET")
|
||
ui.HandleFunc("/tenants", h.listTenants).Methods("GET")
|
||
ui.HandleFunc("/tenants", h.createTenant).Methods("POST")
|
||
ui.HandleFunc("/tenants/{id}", h.getTenant).Methods("GET")
|
||
ui.HandleFunc("/tenants/{id}", h.deleteTenant).Methods("DELETE")
|
||
ui.HandleFunc("/tenants/{id}/queues", h.listTenantQueues).Methods("GET")
|
||
ui.HandleFunc("/tenants/{id}/queues", h.createTenantQueue).Methods("POST")
|
||
ui.HandleFunc("/tenants/{id}/queues/{queue}", h.deleteTenantQueue).Methods("DELETE")
|
||
ui.HandleFunc("/tenants/{id}/queues/{queue}/messages", h.peekQueueMessages).Methods("GET")
|
||
ui.HandleFunc("/tenants/{id}/queues/{queue}/messages", h.sendMessageToQueue).Methods("POST")
|
||
ui.HandleFunc("/tenants/{id}/queues/{queue}/messages", h.purgeQueue).Methods("DELETE")
|
||
}
|
||
|
||
// bearerAuthMiddleware — проверяет Bearer token для admin API (Trap #12)
|
||
func (h *Handler) bearerAuthMiddleware(next http.Handler) http.Handler {
|
||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
authHeader := r.Header.Get("Authorization")
|
||
// Timing-safe сравнение: длина префикса и сам токен проверяются отдельно,
|
||
// само сравнение значения — через subtle.ConstantTimeCompare, чтобы не
|
||
// давать timing side-channel при подборе admin-токена.
|
||
expected := "Bearer " + h.adminToken
|
||
if len(authHeader) != len(expected) ||
|
||
subtle.ConstantTimeCompare([]byte(authHeader), []byte(expected)) != 1 {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusUnauthorized)
|
||
json.NewEncoder(w).Encode(map[string]string{"error": "unauthorized"})
|
||
return
|
||
}
|
||
next.ServeHTTP(w, r)
|
||
})
|
||
}
|
||
|
||
// jwtAuth — POST /ui/api/auth — принимает JWT токен, валидирует через nubes API,
|
||
// auto-provision тенанта если не существует, возвращает email + tenant info.
|
||
// Это единственный публичный endpoint UI API.
|
||
func (h *Handler) jwtAuth(w http.ResponseWriter, r *http.Request) {
|
||
var req struct {
|
||
Token string `json:"token"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Token == "" {
|
||
jsonErr(w, http.StatusBadRequest, "token is required")
|
||
return
|
||
}
|
||
|
||
// Парсим JWT claims
|
||
claims, err := auth.ParseJWTClaims(req.Token)
|
||
if err != nil {
|
||
log.Warnf("jwt auth: parse error: %v", err)
|
||
jsonErr(w, http.StatusUnauthorized, "invalid token: "+err.Error())
|
||
return
|
||
}
|
||
|
||
// Email — единственный ключ идентичности. Без него не создаём тенанта
|
||
// (иначе hash("") даст один тенант на всех пользователей без email).
|
||
if claims.Email == "" {
|
||
log.Warnf("jwt auth: token for sub=%s has no email claim", claims.Sub)
|
||
jsonErr(w, http.StatusBadRequest, "token has no email claim")
|
||
return
|
||
}
|
||
|
||
// Валидируем через nubes API
|
||
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||
defer cancel()
|
||
if err := auth.PingNubesAPI(ctx, h.nubesEndpoint, req.Token); err != nil {
|
||
log.Warnf("jwt auth: nubes rejected token for sub=%s: %v", claims.Sub, err)
|
||
jsonErr(w, http.StatusForbidden, "token rejected by cloud API: "+err.Error())
|
||
return
|
||
}
|
||
// Логин — всегда свежая проверка; после успеха прогреваем кэш,
|
||
// чтобы дальнейшие запросы консоли не дёргали нестабильный шлюз.
|
||
auth.WarmTokenCache(req.Token)
|
||
|
||
// Auto-provisioning: тенант и ключи детерминированы из email.
|
||
// Токен — только аутентификация; ключи в ответ не возвращаем (GET /ui/api/credentials).
|
||
t, err := h.store.CreateFromJWT(claims.Sub, claims.Email, tenant.DefaultTenantMaxQueues)
|
||
if err != nil {
|
||
log.Errorf("jwt auth: failed to create tenant: %v", err)
|
||
jsonErr(w, http.StatusInternalServerError, "failed to provision tenant")
|
||
return
|
||
}
|
||
|
||
log.Infof("jwt auth: authenticated sub=%s email=%s tenant=%s", claims.Sub, claims.Email, t.ID)
|
||
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(map[string]interface{}{
|
||
"email": claims.Email,
|
||
"tenant_id": t.ID,
|
||
"max_queues": t.MaxQueues,
|
||
"token": req.Token, // возвращаем для использования в последующих запросах
|
||
})
|
||
}
|
||
|
||
// uiCredentials — GET /ui/api/credentials: отдаёт AccessKey/SecretKey текущего тенанта.
|
||
// Ключи НЕ светятся в других ответах UI API — только по явному запросу под JWT-сессией.
|
||
func (h *Handler) uiCredentials(w http.ResponseWriter, r *http.Request) {
|
||
t, ok := currentUITenant(r)
|
||
if !ok {
|
||
jsonErr(w, http.StatusUnauthorized, "unauthorized")
|
||
return
|
||
}
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(map[string]string{
|
||
"tenant_id": t.ID,
|
||
"email": t.Email,
|
||
"access_key": t.AccessKey,
|
||
"secret_key": t.SecretKey,
|
||
})
|
||
}
|
||
|
||
// jwtMiddleware — middleware для /ui/api/* endpoints.
|
||
// Пропускает /ui/api/auth (публичный endpoint авторизации).
|
||
// Проверяет Authorization: Bearer <jwt> заголовок.
|
||
// Парсит JWT, проверяет что sub соответствует существующему тенанту.
|
||
// НЕ вызывает PingNubesAPI повторно — токен уже провалидирован при /ui/api/auth.
|
||
func (h *Handler) jwtMiddleware(next http.Handler) http.Handler {
|
||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
// /ui/api/auth — публичный endpoint, пропускаем без проверки JWT.
|
||
// Сравниваем ТОЧНЫЙ путь, а не суффикс "/auth": иначе, например,
|
||
// DELETE /ui/api/tenants/{id}/queues/auth пропускался бы без JWT и IDOR-проверки.
|
||
if r.URL.Path == "/ui/api/auth" {
|
||
next.ServeHTTP(w, r)
|
||
return
|
||
}
|
||
|
||
authHeader := r.Header.Get("Authorization")
|
||
if authHeader == "" {
|
||
jsonErr(w, http.StatusUnauthorized, "authorization required")
|
||
return
|
||
}
|
||
parts := strings.SplitN(authHeader, " ", 2)
|
||
if len(parts) != 2 || !strings.EqualFold(parts[0], "bearer") {
|
||
jsonErr(w, http.StatusUnauthorized, "invalid authorization format")
|
||
return
|
||
}
|
||
token := parts[1]
|
||
|
||
claims, err := auth.ParseJWTClaims(token)
|
||
if err != nil {
|
||
jsonErr(w, http.StatusUnauthorized, "invalid token: "+err.Error())
|
||
return
|
||
}
|
||
|
||
// Повторная валидация через кэш: в течение TTL шлюз не вызывается
|
||
// (шлюз нестабилен с IP платформы), отозванный токен умрёт через TTL,
|
||
// явный 401/403 от шлюза сбрасывает кэш.
|
||
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
|
||
defer cancel()
|
||
if err := auth.PingNubesAPICached(ctx, h.nubesEndpoint, token); err != nil {
|
||
jsonErr(w, http.StatusForbidden, "token rejected by cloud API: "+err.Error())
|
||
return
|
||
}
|
||
|
||
// Проверяем что тенант существует (был создан при /ui/api/auth)
|
||
jwtTenant, ok := h.store.GetByEmail(claims.Email)
|
||
if !ok {
|
||
jsonErr(w, http.StatusForbidden, "tenant not found — authenticate first via POST /ui/api/auth")
|
||
return
|
||
}
|
||
|
||
// IDOR защита: для /ui/api/tenants/{id}/... разрешаем доступ только к своему tenant ID.
|
||
if pathTenantID, exists := mux.Vars(r)["id"]; exists && pathTenantID != "" && pathTenantID != jwtTenant.ID {
|
||
jsonErr(w, http.StatusForbidden, "forbidden tenant access")
|
||
return
|
||
}
|
||
|
||
uiCtx := context.WithValue(r.Context(), uiTenantContextKey, jwtTenant)
|
||
next.ServeHTTP(w, r.WithContext(uiCtx))
|
||
})
|
||
}
|
||
|
||
// createTenantRequest — тело запроса POST /admin/tenants
|
||
type createTenantRequest struct {
|
||
Name string `json:"name"`
|
||
MaxQueues int `json:"max_queues"`
|
||
// BYOC: необязательные произвольные креды. Если задано одно поле —
|
||
// оба обязательны.
|
||
AccessKey string `json:"access_key,omitempty"`
|
||
SecretKey string `json:"secret_key,omitempty"`
|
||
}
|
||
|
||
// tenantCreateResponse — ответ с secret_key (показывается ТОЛЬКО при создании)
|
||
type tenantCreateResponse struct {
|
||
ID string `json:"id"`
|
||
Name string `json:"name"`
|
||
AccessKey string `json:"access_key"`
|
||
SecretKey string `json:"secret_key"`
|
||
MaxQueues int `json:"max_queues"`
|
||
CreatedAt time.Time `json:"created_at"`
|
||
Active bool `json:"active"`
|
||
}
|
||
|
||
// tenantListItem — данные тенанта без secret_key (для List/Get)
|
||
type tenantListItem struct {
|
||
ID string `json:"id"`
|
||
Name string `json:"name"`
|
||
AccessKey string `json:"access_key"`
|
||
MaxQueues int `json:"max_queues"`
|
||
CreatedAt time.Time `json:"created_at"`
|
||
Active bool `json:"active"`
|
||
}
|
||
|
||
// createTenant — POST /admin/tenants
|
||
func (h *Handler) createTenant(w http.ResponseWriter, r *http.Request) {
|
||
if _, ok := currentUITenant(r); ok {
|
||
jsonErr(w, http.StatusForbidden, "tenant creation via UI is disabled")
|
||
return
|
||
}
|
||
var req createTenantRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusBadRequest)
|
||
json.NewEncoder(w).Encode(map[string]string{"error": "invalid request body"})
|
||
return
|
||
}
|
||
if req.Name == "" {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusBadRequest)
|
||
json.NewEncoder(w).Encode(map[string]string{"error": "name is required"})
|
||
return
|
||
}
|
||
|
||
// BYOC: либо оба поля кредов заданы, либо ни одного.
|
||
var t *tenant.Tenant
|
||
var err error
|
||
if req.AccessKey != "" || req.SecretKey != "" {
|
||
if req.AccessKey == "" || req.SecretKey == "" {
|
||
jsonErr(w, http.StatusBadRequest, "both access_key and secret_key are required when specifying custom credentials")
|
||
return
|
||
}
|
||
if len(req.AccessKey) < 8 || len(req.SecretKey) < 16 {
|
||
jsonErr(w, http.StatusBadRequest, "access_key min 8 chars, secret_key min 16 chars")
|
||
return
|
||
}
|
||
t, err = h.store.CreateFixedAutoID(req.Name, req.MaxQueues, req.AccessKey, req.SecretKey)
|
||
} else {
|
||
t, err = h.store.Create(req.Name, req.MaxQueues)
|
||
}
|
||
if err != nil {
|
||
log.Errorf("admin: failed to create tenant: %v", err)
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusInternalServerError)
|
||
json.NewEncoder(w).Encode(map[string]string{"error": "failed to create tenant"})
|
||
return
|
||
}
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusCreated)
|
||
json.NewEncoder(w).Encode(tenantCreateResponse{
|
||
ID: t.ID,
|
||
Name: t.Name,
|
||
AccessKey: t.AccessKey,
|
||
SecretKey: t.SecretKey,
|
||
MaxQueues: t.MaxQueues,
|
||
CreatedAt: t.CreatedAt,
|
||
Active: t.Active,
|
||
})
|
||
}
|
||
|
||
// listTenants — GET /admin/tenants
|
||
func (h *Handler) listTenants(w http.ResponseWriter, r *http.Request) {
|
||
if uiTenant, ok := currentUITenant(r); ok {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode([]tenantListItem{tenantListItemFromTenant(uiTenant)})
|
||
return
|
||
}
|
||
tenants := h.store.List()
|
||
items := make([]tenantListItem, 0, len(tenants))
|
||
for _, t := range tenants {
|
||
items = append(items, tenantListItemFromTenant(t))
|
||
}
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(items)
|
||
}
|
||
|
||
// getTenant — GET /admin/tenants/{id}
|
||
func (h *Handler) getTenant(w http.ResponseWriter, r *http.Request) {
|
||
vars := mux.Vars(r)
|
||
id := vars["id"]
|
||
t, ok := h.store.GetByID(id)
|
||
if !ok {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusNotFound)
|
||
json.NewEncoder(w).Encode(map[string]string{"error": "tenant not found"})
|
||
return
|
||
}
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(tenantListItemFromTenant(t))
|
||
}
|
||
|
||
// deleteTenant — DELETE /admin/tenants/{id}
|
||
// Удаляет тенанта И ВСЕ его очереди из SyncQueues (Trap #11: иначе memory leak)
|
||
func (h *Handler) deleteTenant(w http.ResponseWriter, r *http.Request) {
|
||
if _, ok := currentUITenant(r); ok {
|
||
jsonErr(w, http.StatusForbidden, "tenant deletion via UI is disabled")
|
||
return
|
||
}
|
||
vars := mux.Vars(r)
|
||
id := vars["id"]
|
||
t, ok := h.store.GetByID(id)
|
||
if !ok {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusNotFound)
|
||
json.NewEncoder(w).Encode(map[string]string{"error": "tenant not found"})
|
||
return
|
||
}
|
||
// Удаляем все очереди тенанта из SyncQueues (через tenant-индекс, без скана всех очередей)
|
||
models.SyncQueues.Lock()
|
||
tenantQueueNames := models.SyncQueues.TenantQueueNames(t.AccessKey)
|
||
deletedQueueKeys := make([]string, 0, len(tenantQueueNames))
|
||
for _, name := range tenantQueueNames {
|
||
key := t.AccessKey + ":" + name
|
||
deletedQueueKeys = append(deletedQueueKeys, key)
|
||
models.SyncQueues.RemoveQueue(key)
|
||
}
|
||
models.SyncQueues.Unlock()
|
||
for _, queueKey := range deletedQueueKeys {
|
||
persistence.DeleteQueue(queueKey)
|
||
}
|
||
|
||
h.store.Delete(id)
|
||
w.WriteHeader(http.StatusNoContent)
|
||
}
|
||
|
||
// listTenantQueues — GET /admin/tenants/{id}/queues
|
||
// Возвращает список очередей тенанта с количеством сообщений.
|
||
func (h *Handler) listTenantQueues(w http.ResponseWriter, r *http.Request) {
|
||
vars := mux.Vars(r)
|
||
id := vars["id"]
|
||
t, ok := h.store.GetByID(id)
|
||
if !ok {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusNotFound)
|
||
json.NewEncoder(w).Encode(map[string]string{"error": "tenant not found"})
|
||
return
|
||
}
|
||
prefix := t.AccessKey + ":"
|
||
type queueInfo struct {
|
||
Name string `json:"name"`
|
||
Messages int `json:"messages"`
|
||
NotVisible int `json:"not_visible"`
|
||
VisibilityTimeout int `json:"visibility_timeout"`
|
||
MaxMessageSize int `json:"max_message_size"`
|
||
RetentionPeriod int `json:"retention_period"`
|
||
}
|
||
queues := make([]queueInfo, 0)
|
||
type nameQueue struct {
|
||
name string
|
||
q *models.Queue
|
||
}
|
||
models.SyncQueues.RLock()
|
||
pairs := make([]nameQueue, 0)
|
||
for _, name := range models.SyncQueues.TenantQueueNames(t.AccessKey) {
|
||
if q := models.SyncQueues.Queues[prefix+name]; q != nil {
|
||
pairs = append(pairs, nameQueue{name, q})
|
||
}
|
||
}
|
||
models.SyncQueues.RUnlock()
|
||
for _, p := range pairs {
|
||
p.q.RLockMu()
|
||
visible := 0
|
||
notVisible := 0
|
||
for i := range p.q.Messages {
|
||
if p.q.IsDead(i) {
|
||
continue
|
||
}
|
||
if p.q.Messages[i].ReceiptHandle != "" {
|
||
notVisible++
|
||
} else {
|
||
visible++
|
||
}
|
||
}
|
||
queues = append(queues, queueInfo{
|
||
Name: p.name,
|
||
Messages: visible,
|
||
NotVisible: notVisible,
|
||
VisibilityTimeout: p.q.VisibilityTimeout,
|
||
MaxMessageSize: p.q.MaximumMessageSize,
|
||
RetentionPeriod: p.q.MessageRetentionPeriod,
|
||
})
|
||
p.q.RUnlockMu()
|
||
}
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(queues)
|
||
}
|
||
|
||
// ─── QUEUE MANAGEMENT HANDLERS ────────────────────────────────────────────
|
||
|
||
// createQueueRequest — тело запроса POST .../queues
|
||
type createQueueRequest struct {
|
||
Name string `json:"name"`
|
||
}
|
||
|
||
// createTenantQueue — POST /admin/tenants/{id}/queues
|
||
// Создаёт новую очередь для тенанта прямо в SyncQueues (без SQS-протокола).
|
||
// Проверяет лимит MaxQueues тенанта и уникальность имени.
|
||
func (h *Handler) createTenantQueue(w http.ResponseWriter, r *http.Request) {
|
||
vars := mux.Vars(r)
|
||
tid := vars["id"]
|
||
t, ok := h.store.GetByID(tid)
|
||
if !ok {
|
||
jsonErr(w, http.StatusNotFound, "tenant not found")
|
||
return
|
||
}
|
||
var req createQueueRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Name == "" {
|
||
jsonErr(w, http.StatusBadRequest, "name is required")
|
||
return
|
||
}
|
||
// Та же валидация имени, что и в SQS CreateQueue (иначе имена с ':'/'/'
|
||
// ломали бы ключ "{accessKey}:{name}").
|
||
if err := gosqs.ValidateQueueName(req.Name); err != nil {
|
||
jsonErr(w, http.StatusBadRequest, "invalid queue name")
|
||
return
|
||
}
|
||
key := t.AccessKey + ":" + req.Name
|
||
models.SyncQueues.Lock()
|
||
if _, exists := models.SyncQueues.Queues[key]; exists {
|
||
models.SyncQueues.Unlock()
|
||
jsonErr(w, http.StatusConflict, "queue already exists")
|
||
return
|
||
}
|
||
// Проверяем лимит очередей тенанта (через tenant-индекс, O(1))
|
||
count := models.SyncQueues.CountTenantQueues(t.AccessKey)
|
||
if t.MaxQueues > 0 && count >= t.MaxQueues {
|
||
models.SyncQueues.Unlock()
|
||
jsonErr(w, http.StatusForbidden, "queue limit exceeded")
|
||
return
|
||
}
|
||
// Очередь создаётся через общий билдер — те же URL/ARN/дефолты и FIFO-детект,
|
||
// что и в SQS CreateQueue (единый путь, без расхождения поведения UI и API).
|
||
newQueue := gosqs.BuildQueueForTenant(t, req.Name)
|
||
models.SyncQueues.AddQueue(key, newQueue)
|
||
persistence.SaveQueue(key, newQueue)
|
||
models.SyncQueues.Unlock()
|
||
log.Infof("admin: created queue %s for tenant %s", req.Name, t.ID)
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusCreated)
|
||
json.NewEncoder(w).Encode(map[string]string{"name": req.Name, "status": "created"})
|
||
}
|
||
|
||
// deleteTenantQueue — DELETE /admin/tenants/{id}/queues/{queue}
|
||
// Удаляет очередь тенанта из SyncQueues вместе со всеми её сообщениями.
|
||
func (h *Handler) deleteTenantQueue(w http.ResponseWriter, r *http.Request) {
|
||
vars := mux.Vars(r)
|
||
tid, queueName := vars["id"], vars["queue"]
|
||
t, ok := h.store.GetByID(tid)
|
||
if !ok {
|
||
jsonErr(w, http.StatusNotFound, "tenant not found")
|
||
return
|
||
}
|
||
key := t.AccessKey + ":" + queueName
|
||
models.SyncQueues.Lock()
|
||
if _, exists := models.SyncQueues.Queues[key]; !exists {
|
||
models.SyncQueues.Unlock()
|
||
jsonErr(w, http.StatusNotFound, "queue not found")
|
||
return
|
||
}
|
||
models.SyncQueues.RemoveQueue(key)
|
||
models.SyncQueues.Unlock()
|
||
persistence.DeleteQueue(key)
|
||
log.Infof("admin: deleted queue %s for tenant %s", queueName, t.ID)
|
||
w.WriteHeader(http.StatusNoContent)
|
||
}
|
||
|
||
// peekMessageItem — одно сообщение в ответе peekQueueMessages (без receipt handle)
|
||
type peekMessageItem struct {
|
||
ID string `json:"id"`
|
||
Body string `json:"body"`
|
||
MD5 string `json:"md5"`
|
||
SentAt string `json:"sent_at"`
|
||
Receives int `json:"receives"`
|
||
InFlight bool `json:"in_flight"`
|
||
}
|
||
|
||
// peekQueueMessages — GET /admin/tenants/{id}/queues/{queue}/messages?limit=50
|
||
// Peek-просмотр сообщений: НЕ удаляет, НЕ выставляет ReceiptHandle — только чтение.
|
||
// Это принципиальное отличие от SQS ReceiveMessage (который скрывает сообщения).
|
||
func (h *Handler) peekQueueMessages(w http.ResponseWriter, r *http.Request) {
|
||
vars := mux.Vars(r)
|
||
tid, queueName := vars["id"], vars["queue"]
|
||
t, ok := h.store.GetByID(tid)
|
||
if !ok {
|
||
jsonErr(w, http.StatusNotFound, "tenant not found")
|
||
return
|
||
}
|
||
_, q := findQueue(t.AccessKey, queueName)
|
||
if q == nil {
|
||
jsonErr(w, http.StatusNotFound, "queue not found")
|
||
return
|
||
}
|
||
// Лимит по умолчанию 50, максимум 1000
|
||
limit := 50
|
||
if lv := r.URL.Query().Get("limit"); lv != "" {
|
||
if n := 0; len(lv) > 0 {
|
||
for _, c := range lv {
|
||
if c < '0' || c > '9' {
|
||
n = -1
|
||
break
|
||
}
|
||
n = n*10 + int(c-'0')
|
||
}
|
||
if n > 0 && n <= 1000 {
|
||
limit = n
|
||
}
|
||
}
|
||
}
|
||
q.RLockMu()
|
||
result := make([]peekMessageItem, 0, limit)
|
||
for i := range q.Messages {
|
||
if len(result) >= limit {
|
||
break
|
||
}
|
||
if q.IsDead(i) {
|
||
continue
|
||
}
|
||
msg := &q.Messages[i]
|
||
result = append(result, peekMessageItem{
|
||
ID: msg.Uuid,
|
||
Body: msg.MessageBody,
|
||
MD5: msg.MD5OfMessageBody,
|
||
SentAt: msg.SentTime.Format(time.RFC3339),
|
||
Receives: msg.NumberOfReceives,
|
||
InFlight: msg.ReceiptHandle != "",
|
||
})
|
||
}
|
||
q.RUnlockMu()
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(result)
|
||
}
|
||
|
||
// sendMessageRequest — тело запроса POST .../messages
|
||
type sendMessageRequest struct {
|
||
Body string `json:"body"`
|
||
}
|
||
|
||
// sendMessageToQueue — POST /admin/tenants/{id}/queues/{queue}/messages
|
||
// Отправляет сообщение напрямую в очередь минуя SQS-протокол.
|
||
// Используется только из UI console — для prod нужен нормальный SQS send.
|
||
func (h *Handler) sendMessageToQueue(w http.ResponseWriter, r *http.Request) {
|
||
vars := mux.Vars(r)
|
||
tid, queueName := vars["id"], vars["queue"]
|
||
t, ok := h.store.GetByID(tid)
|
||
if !ok {
|
||
jsonErr(w, http.StatusNotFound, "tenant not found")
|
||
return
|
||
}
|
||
key, q := findQueue(t.AccessKey, queueName)
|
||
if q == nil {
|
||
jsonErr(w, http.StatusNotFound, "queue not found")
|
||
return
|
||
}
|
||
var req sendMessageRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Body == "" {
|
||
jsonErr(w, http.StatusBadRequest, "body is required")
|
||
return
|
||
}
|
||
// Сообщение строится через общий билдер (UUID + MD5), + проверка размера —
|
||
// как в SQS SendMessage.
|
||
q.LockMu()
|
||
if q.MaximumMessageSize > 0 && len(req.Body) > q.MaximumMessageSize {
|
||
q.UnlockMu()
|
||
jsonErr(w, http.StatusBadRequest, "message too big")
|
||
return
|
||
}
|
||
msg := gosqs.BuildSqsMessage(req.Body)
|
||
q.AppendMessage(msg)
|
||
// Персистим одно сообщение отдельно
|
||
persistence.SaveMessage(key, &msg)
|
||
persistence.SaveQueue(key, q)
|
||
q.UnlockMu()
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusCreated)
|
||
json.NewEncoder(w).Encode(map[string]string{"id": msg.Uuid, "status": "sent"})
|
||
}
|
||
|
||
// purgeQueue — DELETE /admin/tenants/{id}/queues/{queue}/messages
|
||
// Удаляет все сообщения из очереди (purge). Сама очередь остаётся.
|
||
func (h *Handler) purgeQueue(w http.ResponseWriter, r *http.Request) {
|
||
vars := mux.Vars(r)
|
||
tid, queueName := vars["id"], vars["queue"]
|
||
t, ok := h.store.GetByID(tid)
|
||
if !ok {
|
||
jsonErr(w, http.StatusNotFound, "tenant not found")
|
||
return
|
||
}
|
||
key, q := findQueue(t.AccessKey, queueName)
|
||
if q == nil {
|
||
jsonErr(w, http.StatusNotFound, "queue not found")
|
||
return
|
||
}
|
||
q.LockMu()
|
||
q.Reset()
|
||
// Удаляем все сообщения из Redis одной командой
|
||
persistence.PurgeMessagesPersist(key)
|
||
persistence.SaveQueue(key, q)
|
||
q.UnlockMu()
|
||
log.Infof("admin: purged queue %s for tenant %s", queueName, t.ID)
|
||
w.WriteHeader(http.StatusNoContent)
|
||
}
|
||
|
||
// jsonErr — вспомогательная функция: ответ с ошибкой в JSON
|
||
func jsonErr(w http.ResponseWriter, code int, msg string) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(code)
|
||
json.NewEncoder(w).Encode(map[string]string{"error": msg})
|
||
}
|
||
|
||
// adminHealthDetail — ответ GET /admin/health
|
||
type adminHealthDetail struct {
|
||
Status string `json:"status"`
|
||
Version string `json:"version"`
|
||
TenantCount int `json:"tenant_count"`
|
||
QueueCount int `json:"queue_count"`
|
||
MessageCount int `json:"message_count"`
|
||
}
|
||
|
||
// detailedHealth — GET /admin/health
|
||
func (h *Handler) detailedHealth(w http.ResponseWriter, r *http.Request) {
|
||
if uiTenant, ok := currentUITenant(r); ok {
|
||
prefix := uiTenant.AccessKey + ":"
|
||
queueCount := 0
|
||
msgCount := 0
|
||
models.SyncQueues.RLock()
|
||
uiQueues := make([]*models.Queue, 0)
|
||
for key, q := range models.SyncQueues.Queues {
|
||
if strings.HasPrefix(key, prefix) {
|
||
queueCount++
|
||
uiQueues = append(uiQueues, q)
|
||
}
|
||
}
|
||
models.SyncQueues.RUnlock()
|
||
for _, q := range uiQueues {
|
||
q.RLockMu()
|
||
msgCount += q.LiveCount()
|
||
q.RUnlockMu()
|
||
}
|
||
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(adminHealthDetail{
|
||
Status: "ok",
|
||
Version: models.Version,
|
||
TenantCount: 1,
|
||
QueueCount: queueCount,
|
||
MessageCount: msgCount,
|
||
})
|
||
return
|
||
}
|
||
tenants := h.store.List()
|
||
models.SyncQueues.RLock()
|
||
queueCount := len(models.SyncQueues.Queues)
|
||
allQueues := make([]*models.Queue, 0, queueCount)
|
||
for _, q := range models.SyncQueues.Queues {
|
||
allQueues = append(allQueues, q)
|
||
}
|
||
models.SyncQueues.RUnlock()
|
||
msgCount := 0
|
||
for _, q := range allQueues {
|
||
q.RLockMu()
|
||
msgCount += q.LiveCount()
|
||
q.RUnlockMu()
|
||
}
|
||
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(adminHealthDetail{
|
||
Status: "ok",
|
||
Version: models.Version,
|
||
TenantCount: len(tenants),
|
||
QueueCount: queueCount,
|
||
MessageCount: msgCount,
|
||
})
|
||
}
|