808 lines
30 KiB
Go
808 lines
30 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"
|
||
"errors"
|
||
"net/http"
|
||
"os"
|
||
"strings"
|
||
"time"
|
||
|
||
"shared-sqs/app/auth"
|
||
"shared-sqs/app/models"
|
||
"shared-sqs/app/persistence"
|
||
"shared-sqs/app/tenant"
|
||
|
||
"github.com/google/uuid"
|
||
"github.com/gorilla/mux"
|
||
log "github.com/sirupsen/logrus"
|
||
)
|
||
|
||
type uiContextKey string
|
||
|
||
const (
|
||
uiTenantContextKey uiContextKey = "ui-tenant"
|
||
)
|
||
|
||
// ВАЖНО: значения ниже относятся только к временному demo/showcase режиму.
|
||
// Перед production rollout без публичного demo-режима этот блок должен быть удалён
|
||
// вместе с веткой authenticateUIDemoToken и связанными UI-подсказками.
|
||
const (
|
||
defaultUIDemoToken = "demo-ui-shared-sqs-ngcloud-2026"
|
||
defaultUIDemoTenantID = "t-demo-shared-sqs-ngcloud"
|
||
defaultUIDemoEmail = "demo@shared-sqs.ngcloud"
|
||
)
|
||
|
||
var errUIDemoTokenMismatch = errors.New("demo token mismatch")
|
||
|
||
// ─── вспомогательная функция: найти очередь тенанта по имени ───────────────
|
||
// 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}
|
||
}
|
||
|
||
// uiDemoToken — возвращает публичный demo token для UI.
|
||
// Временный showcase-only путь: в production без demo user этот метод нужно удалить.
|
||
func (h *Handler) uiDemoToken() string {
|
||
if token := os.Getenv("SHARED_SQS_UI_DEMO_TOKEN"); token != "" {
|
||
return token
|
||
}
|
||
return defaultUIDemoToken
|
||
}
|
||
|
||
// uiDemoTenantID — возвращает tenant ID, к которому привязан demo token.
|
||
// Существует только для showcase-режима, не для постоянной production auth-модели.
|
||
func (h *Handler) uiDemoTenantID() string {
|
||
if tenantID := os.Getenv("SHARED_SQS_UI_DEMO_TENANT_ID"); tenantID != "" {
|
||
return tenantID
|
||
}
|
||
return defaultUIDemoTenantID
|
||
}
|
||
|
||
// uiDemoEmail — возвращает отображаемый email для demo UI session.
|
||
// Нужен только для публичного демо-логина и должен уйти вместе с demo path.
|
||
func (h *Handler) uiDemoEmail() string {
|
||
if email := os.Getenv("SHARED_SQS_UI_DEMO_EMAIL"); email != "" {
|
||
return email
|
||
}
|
||
return defaultUIDemoEmail
|
||
}
|
||
|
||
// authenticateUIDemoToken — маппит публичный demo token на заранее сидированный demo tenant.
|
||
// Почему так: это быстрый showcase-вход для заказчика. Для production без demo user
|
||
// функция должна быть удалена, чтобы в коде не осталось публичного bypass-пути.
|
||
func (h *Handler) authenticateUIDemoToken(token string) (*tenant.Tenant, string, error) {
|
||
demoToken := h.uiDemoToken()
|
||
if demoToken == "" || subtle.ConstantTimeCompare([]byte(token), []byte(demoToken)) != 1 {
|
||
return nil, "", errUIDemoTokenMismatch
|
||
}
|
||
demoTenant, ok := h.store.GetByID(h.uiDemoTenantID())
|
||
if !ok {
|
||
return nil, "", errors.New("demo tenant unavailable — enable SHARED_SQS_SEED_DEMO=true")
|
||
}
|
||
return demoTenant, h.uiDemoEmail(), nil
|
||
}
|
||
|
||
// 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")
|
||
expected := "Bearer " + h.adminToken
|
||
if authHeader != expected {
|
||
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
|
||
}
|
||
|
||
if demoTenant, demoEmail, err := h.authenticateUIDemoToken(req.Token); err == nil {
|
||
log.Infof("ui auth: authenticated demo tenant=%s", demoTenant.ID)
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(map[string]interface{}{
|
||
"email": demoEmail,
|
||
"tenant_id": demoTenant.ID,
|
||
"access_key": demoTenant.AccessKey,
|
||
"secret_key": demoTenant.SecretKey,
|
||
"max_queues": demoTenant.MaxQueues,
|
||
"token": req.Token,
|
||
})
|
||
return
|
||
} else if !errors.Is(err, errUIDemoTokenMismatch) {
|
||
log.Warnf("ui auth: demo login unavailable: %v", err)
|
||
jsonErr(w, http.StatusForbidden, err.Error())
|
||
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, 10)
|
||
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
|
||
if strings.HasSuffix(r.URL.Path, "/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]
|
||
|
||
if demoTenant, _, err := h.authenticateUIDemoToken(token); err == nil {
|
||
if pathTenantID, exists := mux.Vars(r)["id"]; exists && pathTenantID != "" && pathTenantID != demoTenant.ID {
|
||
jsonErr(w, http.StatusForbidden, "forbidden tenant access")
|
||
return
|
||
}
|
||
ctx := context.WithValue(r.Context(), uiTenantContextKey, demoTenant)
|
||
next.ServeHTTP(w, r.WithContext(ctx))
|
||
return
|
||
} else if !errors.Is(err, errUIDemoTokenMismatch) {
|
||
jsonErr(w, http.StatusForbidden, err.Error())
|
||
return
|
||
}
|
||
|
||
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"`
|
||
}
|
||
|
||
// 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
|
||
}
|
||
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
|
||
prefix := t.AccessKey + ":"
|
||
deletedQueueKeys := make([]string, 0)
|
||
models.SyncQueues.Lock()
|
||
for key := range models.SyncQueues.Queues {
|
||
if strings.HasPrefix(key, prefix) {
|
||
deletedQueueKeys = append(deletedQueueKeys, key)
|
||
delete(models.SyncQueues.Queues, 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)
|
||
models.SyncQueues.RLock()
|
||
for key, q := range models.SyncQueues.Queues {
|
||
if strings.HasPrefix(key, prefix) {
|
||
name := strings.TrimPrefix(key, prefix)
|
||
visible := 0
|
||
notVisible := 0
|
||
for _, msg := range q.Messages {
|
||
if msg.ReceiptHandle != "" {
|
||
notVisible++
|
||
} else {
|
||
visible++
|
||
}
|
||
}
|
||
queues = append(queues, queueInfo{
|
||
Name: name,
|
||
Messages: visible,
|
||
NotVisible: notVisible,
|
||
VisibilityTimeout: q.VisibilityTimeout,
|
||
MaxMessageSize: q.MaximumMessageSize,
|
||
RetentionPeriod: q.MessageRetentionPeriod,
|
||
})
|
||
}
|
||
}
|
||
models.SyncQueues.RUnlock()
|
||
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
|
||
}
|
||
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
|
||
}
|
||
// Проверяем лимит очередей тенанта
|
||
count := 0
|
||
for k := range models.SyncQueues.Queues {
|
||
if strings.HasPrefix(k, t.AccessKey+":") {
|
||
count++
|
||
}
|
||
}
|
||
if t.MaxQueues > 0 && count >= t.MaxQueues {
|
||
models.SyncQueues.Unlock()
|
||
jsonErr(w, http.StatusForbidden, "queue limit exceeded")
|
||
return
|
||
}
|
||
models.SyncQueues.Queues[key] = &models.Queue{
|
||
Name: req.Name,
|
||
VisibilityTimeout: 30,
|
||
MaximumMessageSize: 262144,
|
||
MessageRetentionPeriod: 345600,
|
||
Messages: []models.SqsMessage{},
|
||
Duplicates: make(map[string]time.Time),
|
||
}
|
||
persistence.SaveQueue(key, models.SyncQueues.Queues[key])
|
||
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
|
||
}
|
||
delete(models.SyncQueues.Queues, 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
|
||
}
|
||
}
|
||
}
|
||
models.SyncQueues.RLock()
|
||
result := make([]peekMessageItem, 0, len(q.Messages))
|
||
for i, msg := range q.Messages {
|
||
if i >= limit {
|
||
break
|
||
}
|
||
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 != "",
|
||
})
|
||
}
|
||
models.SyncQueues.RUnlock()
|
||
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
|
||
}
|
||
msg := models.SqsMessage{
|
||
MessageBody: req.Body,
|
||
Uuid: uuid.NewString(),
|
||
SentTime: time.Now(),
|
||
}
|
||
models.SyncQueues.Lock()
|
||
models.SyncQueues.Queues[key].Messages = append(models.SyncQueues.Queues[key].Messages, msg)
|
||
// Персистим одно сообщение отдельно
|
||
persistence.SaveMessage(key, &msg)
|
||
persistence.SaveQueue(key, models.SyncQueues.Queues[key])
|
||
models.SyncQueues.Unlock()
|
||
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
|
||
}
|
||
models.SyncQueues.Lock()
|
||
models.SyncQueues.Queues[key].Messages = models.SyncQueues.Queues[key].Messages[:0]
|
||
// Удаляем все сообщения из Redis одной командой
|
||
persistence.PurgeMessagesPersist(key)
|
||
persistence.SaveQueue(key, models.SyncQueues.Queues[key])
|
||
models.SyncQueues.Unlock()
|
||
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()
|
||
for key, q := range models.SyncQueues.Queues {
|
||
if strings.HasPrefix(key, prefix) {
|
||
queueCount++
|
||
msgCount += len(q.Messages)
|
||
}
|
||
}
|
||
models.SyncQueues.RUnlock()
|
||
|
||
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)
|
||
msgCount := 0
|
||
for _, q := range models.SyncQueues.Queues {
|
||
msgCount += len(q.Messages)
|
||
}
|
||
models.SyncQueues.RUnlock()
|
||
|
||
w.Header().Set("Content-Type", "application/json")
|
||
json.NewEncoder(w).Encode(adminHealthDetail{
|
||
Status: "ok",
|
||
Version: models.Version,
|
||
TenantCount: len(tenants),
|
||
QueueCount: queueCount,
|
||
MessageCount: msgCount,
|
||
})
|
||
}
|