209 lines
7.6 KiB
Go
209 lines
7.6 KiB
Go
// app/router/router.go
|
||
// HTTP router для shared-sqs
|
||
// Updated: 2026-04-09 — добавлены TenantStore, admin API, auth middleware
|
||
package router
|
||
|
||
import (
|
||
"encoding/json"
|
||
"encoding/xml"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"strings"
|
||
|
||
"time"
|
||
|
||
"shared-sqs/app/admin"
|
||
"shared-sqs/app/auth"
|
||
"shared-sqs/app/billing"
|
||
sqs "shared-sqs/app/gosqs"
|
||
"shared-sqs/app/interfaces"
|
||
"shared-sqs/app/metrics"
|
||
"shared-sqs/app/models"
|
||
"shared-sqs/app/tenant"
|
||
"shared-sqs/app/ui"
|
||
|
||
"github.com/gorilla/mux"
|
||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||
log "github.com/sirupsen/logrus"
|
||
)
|
||
|
||
// New — создаёт HTTP router с tenant auth и admin API
|
||
func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler {
|
||
r := mux.NewRouter()
|
||
|
||
// Fix #3.5: глобальный лимит размера тела HTTP запроса — 3MB
|
||
// Защита от OOM при отправке гигантских batch/message requests
|
||
r.Use(requestBodyLimitMiddleware)
|
||
|
||
// /health — публичный, без auth
|
||
r.HandleFunc("/health", health).Methods("GET")
|
||
|
||
// /metrics — Prometheus endpoint для Victoria Metrics scrape
|
||
r.Handle("/metrics", promhttp.Handler()).Methods("GET")
|
||
|
||
// Admin API — Bearer token auth, регистрируется через AdminHandler
|
||
adminHandler := admin.NewHandler(tenantStore, adminToken)
|
||
adminHandler.RegisterRoutes(r)
|
||
|
||
// UI public API — JWT auth, для встроенной console
|
||
// ВАЖНО: RegisterPublicRoutes ПЕРЕД static handler — иначе PathPrefix("/ui") перехватит /ui/api/*
|
||
adminHandler.RegisterPublicRoutes(r)
|
||
|
||
// UI console — встроенный SPA, публичный доступ
|
||
// НЕ ловит /ui/api/* — mux сначала проверит более специфичные маршруты выше
|
||
r.PathPrefix("/ui").Handler(http.StripPrefix("/ui", ui.Handler()))
|
||
|
||
// SQS API — tenant auth middleware оборачивает каждый handler отдельно.
|
||
// r.NewRoute().Subrouter() с Use() некорректно работает в gorilla/mux v1.8.0
|
||
// при пустом prefix — ответы теряются. Поэтому используем явную обёртку.
|
||
sqsAuth := auth.AuthMiddleware(tenantStore)
|
||
r.Handle("/", sqsAuth(http.HandlerFunc(actionHandler))).Methods("GET", "POST")
|
||
r.Handle("/{account}", sqsAuth(http.HandlerFunc(actionHandler))).Methods("GET", "POST")
|
||
r.Handle("/queue/{queueName}", sqsAuth(http.HandlerFunc(actionHandler))).Methods("GET", "POST")
|
||
r.Handle("/{account}/{queueName}", sqsAuth(http.HandlerFunc(actionHandler))).Methods("GET", "POST")
|
||
|
||
return r
|
||
}
|
||
|
||
// requestBodyLimitMiddleware — ограничивает размер тела HTTP запроса до 3MB.
|
||
// Защита от OOM при отправке гигантских messages/batch (Fix #3.5).
|
||
func requestBodyLimitMiddleware(next http.Handler) http.Handler {
|
||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
if r.Body != nil {
|
||
r.Body = http.MaxBytesReader(w, r.Body, 3*1024*1024) // 3MB
|
||
}
|
||
next.ServeHTTP(w, r)
|
||
})
|
||
}
|
||
|
||
func encodeResponse(w http.ResponseWriter, req *http.Request, statusCode int, body interfaces.AbstractResponseBody) {
|
||
protocol := resolveProtocol(req)
|
||
switch protocol {
|
||
case AwsJsonProtocol:
|
||
w.Header().Set("x-amzn-RequestId", body.GetRequestId())
|
||
w.Header().Set("Content-Type", "application/x-amz-json-1.0")
|
||
w.WriteHeader(statusCode)
|
||
if body.GetResult() == nil {
|
||
return
|
||
}
|
||
err := json.NewEncoder(w).Encode(body.GetResult())
|
||
if err != nil {
|
||
log.Errorf("Response Encoding Error: %v\nResponse: %+v", err, body)
|
||
http.Error(w, "General Error", http.StatusInternalServerError)
|
||
}
|
||
case AwsQueryProtocol:
|
||
w.Header().Set("Content-Type", "application/xml")
|
||
w.WriteHeader(statusCode)
|
||
result, err := xml.Marshal(body)
|
||
if err != nil {
|
||
log.Errorf("Response Encoding Error: %v\nResponse: %+v", err, body)
|
||
http.Error(w, "General Error", http.StatusInternalServerError)
|
||
}
|
||
_, _ = w.Write(result)
|
||
}
|
||
}
|
||
|
||
// routingTableV1 — только SQS actions (SNS удалён)
|
||
var routingTableV1 = map[string]func(r *http.Request) (int, interfaces.AbstractResponseBody){
|
||
"CreateQueue": sqs.CreateQueueV1,
|
||
"ListQueues": sqs.ListQueuesV1,
|
||
"GetQueueAttributes": sqs.GetQueueAttributesV1,
|
||
"SetQueueAttributes": sqs.SetQueueAttributesV1,
|
||
"SendMessage": sqs.SendMessageV1,
|
||
"ReceiveMessage": sqs.ReceiveMessageV1,
|
||
"ChangeMessageVisibility": sqs.ChangeMessageVisibilityV1,
|
||
"DeleteMessage": sqs.DeleteMessageV1,
|
||
"GetQueueUrl": sqs.GetQueueUrlV1,
|
||
"PurgeQueue": sqs.PurgeQueueV1,
|
||
"DeleteQueue": sqs.DeleteQueueV1,
|
||
"SendMessageBatch": sqs.SendMessageBatchV1,
|
||
"DeleteMessageBatch": sqs.DeleteMessageBatchV1,
|
||
"ChangeMessageVisibilityBatch": sqs.ChangeMessageVisibilityBatchV1,
|
||
"TagQueue": sqs.TagQueueV1,
|
||
"UntagQueue": sqs.UntagQueueV1,
|
||
"ListQueueTags": sqs.ListQueueTagsV1,
|
||
}
|
||
|
||
func health(w http.ResponseWriter, req *http.Request) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(200)
|
||
fmt.Fprintf(w, `{"status":"ok","version":%q}`, models.Version)
|
||
}
|
||
|
||
func actionHandler(w http.ResponseWriter, req *http.Request) {
|
||
action := extractAction(req)
|
||
start := time.Now()
|
||
log.WithFields(log.Fields{
|
||
"action": action,
|
||
"url": req.URL,
|
||
}).Debug("Handling URL request")
|
||
jsonFn, ok := routingTableV1[action]
|
||
if ok {
|
||
statusCode, responseBody := jsonFn(req)
|
||
encodeResponse(w, req, statusCode, responseBody)
|
||
|
||
// Metrics: latency гистограмма для каждой операции
|
||
duration := time.Since(start).Seconds()
|
||
metrics.RequestDuration.WithLabelValues(action).Observe(duration)
|
||
|
||
if statusCode < 400 {
|
||
if t, _ := req.Context().Value(auth.TenantContextKey).(*tenant.Tenant); t != nil {
|
||
msgBytes := req.ContentLength
|
||
if msgBytes < 0 {
|
||
msgBytes = 0
|
||
}
|
||
// Извлекаем имя очереди: из URL path vars или из form-параметра QueueName (CreateQueue)
|
||
queueName := mux.Vars(req)["queueName"]
|
||
if queueName == "" {
|
||
queueName = req.FormValue("QueueName")
|
||
}
|
||
// Billing: запись в PostgreSQL (async, no-op если billing выключен)
|
||
billing.RecordUsage(t.ID, action, queueName, 1, msgBytes)
|
||
// Prometheus counters: requests + bytes
|
||
metrics.RequestsTotal.WithLabelValues(t.ID, action).Inc()
|
||
metrics.RequestBytesTotal.WithLabelValues(t.ID, action).Add(float64(msgBytes))
|
||
}
|
||
} else {
|
||
// Prometheus: счётчик ошибок
|
||
metrics.ErrorsTotal.WithLabelValues(action).Inc()
|
||
}
|
||
return
|
||
}
|
||
log.Warnf("Bad Request - Action: %s", action)
|
||
w.WriteHeader(http.StatusBadRequest)
|
||
io.WriteString(w, "Bad Request")
|
||
}
|
||
|
||
type AwsProtocol int
|
||
|
||
const (
|
||
AwsJsonProtocol AwsProtocol = iota
|
||
AwsQueryProtocol AwsProtocol = iota
|
||
)
|
||
|
||
// extractAction — извлекает Action из запроса (Query Protocol или JSON Protocol)
|
||
func extractAction(req *http.Request) string {
|
||
protocol := resolveProtocol(req)
|
||
switch protocol {
|
||
case AwsJsonProtocol:
|
||
action := req.Header.Get("X-Amz-Target")
|
||
parts := strings.SplitN(action, ".", 2)
|
||
if len(parts) != 2 || parts[1] == "" {
|
||
return ""
|
||
}
|
||
return parts[1]
|
||
case AwsQueryProtocol:
|
||
return req.FormValue("Action")
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// resolveProtocol — определяет протокол по Content-Type
|
||
func resolveProtocol(req *http.Request) AwsProtocol {
|
||
if req.Header.Get("Content-Type") == "application/x-amz-json-1.0" {
|
||
return AwsJsonProtocol
|
||
}
|
||
return AwsQueryProtocol
|
||
}
|