Maksim Totmin 54259e51b9 feat: CID-based routing rules engine with flexible match conditions
Routing rules:
- RouteRule model: 3 match fields (caller_id, dest, ingress_trunk regex),
  3 action types (node/pool/auto), time-based (days of week, time range)
- Engine router.go: CompiledRule with pre-compiled regex,
  PickNodeForCall with cascade (dead node → next rule), pickNodeFromPool
- Separate sync.RWMutex for Router — zero contention admin vs traffic
- Validate-before-save pattern: engine validates before JSON save
- Graceful degradation: corrupted rules skipped, no rules → PickNode()

Persistence:
- data/routing.json with atomic save (tmp→rename), config.Manager
- Auto-created empty on first start

API (admin only):
- GET/POST/PUT/DELETE /api/routing/rules
- PUT /api/routing/rules/:id/toggle
- PUT /api/routing/rules/reorder

Integration:
- /api/route now uses PickNodeForCall with caller_id/dest/ingress_trunk
  query params (already sent by route.lua)
- RouteResponse includes matched_rule field for debugging
- Main.go loads routing rules on startup with graceful fallback

Web UI:
- /routing page: table with priority arrows, inline edit, create form,
  regex test tool, toggle, delete, action badges, hits counter

Tests:
- 28 tests: matchRule, time matching, PickNodeForCall cascade,
  pool scoring, validation, hit counting, compilation
2026-06-25 21:51:37 +07:00

235 lines
8.4 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package api
import (
"encoding/json"
"fmt"
"net/http"
"sync"
"github.com/pulse-lets-go/internal/config"
"github.com/pulse-lets-go/internal/engine"
"github.com/pulse-lets-go/internal/esl"
"github.com/pulse-lets-go/internal/models"
)
// API — главная структура HTTP API, агрегирует все зависимости.
type API struct {
engine *engine.Engine
configManager *config.Manager
jwtSecret string
monitoringAPIKey string
wsHub *wsHub
natsConnected func() bool // колбэк для проверки NATS-статуса (health)
eslClient *esl.Client // ESL-клиент (nil если не сконфигурирован)
logFormat string
routeLimiter *rateLimiter
apiLimiter *rateLimiter
// In-memory кэш gateway для /api/route (устраняет disk I/O в hot path)
trunkCacheMu sync.RWMutex
balanceGateways map[string]string // nodeID → gateway
fallbackGateway string
gatewayMu sync.RWMutex
gatewayStates map[string]string // trunkID → "up"/"down"
fsStats esl.FsStats
fsStatsMu sync.RWMutex
}
// NewAPI создаёт новый HTTP API с заданными зависимостями.
func NewAPI(eng *engine.Engine, cfgMgr *config.Manager, jwtSecret, monitoringAPIKey string, natsFn func() bool, cfg *config.Config) *API {
a := &API{
engine: eng,
configManager: cfgMgr,
jwtSecret: jwtSecret,
monitoringAPIKey: monitoringAPIKey,
wsHub: newWSHub(),
natsConnected: natsFn,
logFormat: cfg.Log.Format,
gatewayStates: make(map[string]string),
balanceGateways: make(map[string]string),
}
if cfg.RateLimit.Enabled {
a.routeLimiter = newRateLimiter(cfg.RateLimit.RoutePerSec, cfg.RateLimit.RoutePerSec)
a.apiLimiter = newRateLimiter(cfg.RateLimit.APIPerSec, cfg.RateLimit.APIPerSec)
}
return a
}
// SetESLClient устанавливает ESL-клиент (вызывается после инициализации ESL).
func (a *API) SetESLClient(client *esl.Client) {
a.eslClient = client
}
// Handler возвращает http.Handler со всеми зарегистрированными маршрутами.
func (a *API) Handler() http.Handler {
mux := http.NewServeMux()
// --- Health endpoints (без авторизации) ---
mux.HandleFunc("GET /api/health", a.handleHealth)
mux.HandleFunc("GET /api/health/live", a.handleLiveness)
mux.HandleFunc("GET /api/health/ready", a.handleReadiness)
// --- Публичные маршруты (без авторизации) ---
mux.HandleFunc("POST /api/auth/login", a.handleLogin)
mux.HandleFunc("POST /api/auth/refresh", a.handleRefresh)
mux.HandleFunc("GET /api/route", a.handleRoute)
mux.HandleFunc("GET /ws/metrics", a.handleWS)
// Monitoring с API-key (без JWT)
mux.Handle("GET /api/monitoring/zabbix", a.apiKeyMiddleware(http.HandlerFunc(a.handleZabbix)))
mux.Handle("GET /api/monitoring/prometheus", a.apiKeyMiddleware(http.HandlerFunc(a.handlePrometheus)))
// --- Админка: Nodes (admin + viewer) ---
mux.Handle("GET /api/nodes", a.authMiddleware(http.HandlerFunc(a.handleGetNodes)))
mux.Handle("GET /api/nodes/{id}/metrics", a.authMiddleware(http.HandlerFunc(a.handleGetNodeMetrics)))
mux.Handle("PUT /api/nodes/{id}/toggle", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleToggleNode))))
mux.Handle("PUT /api/route/mode", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleSetRouteMode))))
// --- Админка: Trunks (admin only) ---
mux.Handle("GET /api/trunks", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleGetTrunks))))
mux.Handle("POST /api/trunks", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleCreateTrunk))))
mux.Handle("PUT /api/trunks/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleUpdateTrunk))))
mux.Handle("DELETE /api/trunks/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleDeleteTrunk))))
// --- Админка: Users (admin only) ---
mux.Handle("GET /api/users", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleGetUsers))))
mux.Handle("POST /api/users", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleCreateUser))))
mux.Handle("PUT /api/users/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleUpdateUser))))
mux.Handle("DELETE /api/users/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleDeleteUser))))
// --- Админка: Routing Rules (admin + viewer for GET, admin only for write) ---
mux.Handle("GET /api/routing/rules", a.authMiddleware(http.HandlerFunc(a.handleGetRoutingRules)))
mux.Handle("POST /api/routing/rules", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleCreateRoutingRule))))
mux.Handle("PUT /api/routing/rules/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleUpdateRoutingRule))))
mux.Handle("DELETE /api/routing/rules/{id}", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleDeleteRoutingRule))))
mux.Handle("PUT /api/routing/rules/{id}/toggle", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleToggleRoutingRule))))
mux.Handle("PUT /api/routing/rules/reorder", a.authMiddleware(a.adminOnly(http.HandlerFunc(a.handleReorderRoutingRules))))
// Логирование + CORS (навешиваем после mux)
var wrapped http.Handler = mux
wrapped = requestLoggingMiddleware(a.logFormat)(wrapped)
// Rate limiting
if a.routeLimiter != nil && a.apiLimiter != nil {
handlerBeforeRL := wrapped
wrapped = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/api/route" {
rateLimitMiddleware(a.routeLimiter)(handlerBeforeRL).ServeHTTP(w, r)
return
}
if len(r.URL.Path) > 5 && r.URL.Path[:5] == "/api/" {
rateLimitMiddleware(a.apiLimiter)(handlerBeforeRL).ServeHTTP(w, r)
return
}
handlerBeforeRL.ServeHTTP(w, r)
})
}
return corsMiddleware(wrapped)
}
// corsMiddleware добавляет CORS-заголовки.
func corsMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-API-Key")
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
next.ServeHTTP(w, r)
})
}
// BroadcastGatewayEvent транслирует статус gateway всем WS-клиентам.
func (a *API) BroadcastGatewayEvent(gatewayName, trunkID, status string) {
a.gatewayMu.Lock()
a.gatewayStates[trunkID] = status
a.gatewayMu.Unlock()
msg := &models.WsMetricsMessage{
Type: "gateway_update",
Payload: map[string]string{
"gateway": gatewayName,
"trunk_id": trunkID,
"status": status,
},
}
a.wsHub.broadcast(msg)
}
func (a *API) gatewayStats() (total, up, down int) {
a.gatewayMu.RLock()
defer a.gatewayMu.RUnlock()
for _, s := range a.gatewayStates {
total++
switch s {
case "up":
up++
case "down":
down++
}
}
return
}
// RebuildTrunkCache перестраивает in-memory кэш gateway для /api/route.
// Вызывается при старте и после любого CRUD-оперирования с транками.
func (a *API) RebuildTrunkCache() {
trunks, err := a.configManager.ReadTrunks()
if err != nil {
return
}
a.trunkCacheMu.Lock()
defer a.trunkCacheMu.Unlock()
a.balanceGateways = make(map[string]string, len(trunks))
a.fallbackGateway = ""
for _, t := range trunks {
if t.Enabled {
switch t.Type {
case models.TrunkBalance:
if t.NodeID != "" {
a.balanceGateways[t.NodeID] = t.Gateway
}
case models.TrunkFallback:
if a.fallbackGateway == "" {
a.fallbackGateway = t.Gateway
}
}
}
}
}
// UpdateFsStats обновляет кэш метрик FreeSWITCH (вызывается из ticker).
func (a *API) UpdateFsStats(stats esl.FsStats) {
a.fsStatsMu.Lock()
a.fsStats = stats
a.fsStatsMu.Unlock()
}
// maxBodySize — максимальный размер тела запроса в байтах (1 MiB).
const maxBodySize = 1 << 20
// decodeJSON декодирует тело запроса в структуру v с ограничением размера.
func decodeJSON(r *http.Request, v interface{}) error {
defer r.Body.Close()
r.Body = http.MaxBytesReader(nil, r.Body, maxBodySize)
if err := json.NewDecoder(r.Body).Decode(v); err != nil {
return fmt.Errorf("декодирование JSON: %w", err)
}
return nil
}