package httpapi import ( "context" "encoding/json" "io" "net/http" "strings" "sync" "time" "evobgp/internal/jobs" "evobgp/internal/pgmonitor" ) var ( pgMaintRateMu sync.Mutex pgMaintLastByTK = map[string]time.Time{} ) func (s *Server) registerPostgresMaintenanceRoutes(m *http.ServeMux) { m.HandleFunc("POST /postgres/vacuum", s.handlePostgresVacuum) m.HandleFunc("POST /postgres/vacuum-analyze", s.handlePostgresVacuumAnalyze) m.HandleFunc("POST /postgres/analyze", s.handlePostgresAnalyze) m.HandleFunc("POST /postgres/reindex", s.handlePostgresReindex) m.HandleFunc("POST /postgres/cleanup", s.handlePostgresCleanup) m.HandleFunc("GET /postgres/maintenance/logs", s.handlePostgresMaintenanceLogs) } // requireOperatorStrict is a compatibility shim mapping the legacy "operator" // API-key role to the tenant-settings admin permission for JWT/API-key clients. func (s *Server) requireOperatorStrict(w http.ResponseWriter, a Auth) bool { return s.requirePerm(w, a, "bgp:tenant_settings:admin") } func (s *Server) checkPgMaintRateLimit(tenantID, kind string) bool { key := tenantID + ":" + kind pgMaintRateMu.Lock() defer pgMaintRateMu.Unlock() if t, ok := pgMaintLastByTK[key]; ok && time.Since(t) < 60*time.Second { return false } pgMaintLastByTK[key] = time.Now().UTC() return true } type pgMaintBody struct { Table string `json:"table"` DryRun bool `json:"dry_run"` Index string `json:"index"` Policy string `json:"policy"` PolicyID string `json:"policy_id"` Limit int `json:"limit"` } func (s *Server) decodePgMaintBody(r *http.Request) (pgMaintBody, bool) { var body pgMaintBody if r.Body == nil || r.ContentLength == 0 { return body, true } if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&body); err != nil && err != io.EOF { return body, false } return body, true } func (s *Server) enqueuePostgresMaint(w http.ResponseWriter, r *http.Request, a Auth, kind string, meta map[string]any) { if !s.requirePostgres(w) || !s.requireOperatorStrict(w, a) { return } if !s.checkPgMaintRateLimit(a.TenantID, kind) { writeProblem(w, http.StatusTooManyRequests, "Too Many Requests", "wait before repeating this maintenance operation") return } idem := strings.TrimSpace(r.Header.Get("Idempotency-Key")) var idemPtr *string if idem != "" { idemPtr = &idem } meta["actor_prefix"] = actorPrefix(a) j, _, err := s.jobs.Enqueue(a.TenantID, kind, idemPtr, nil, meta) if err != nil { writeInternalError(w, "postgres_maint_enqueue", err) return } w.Header().Set("Location", "/v1/jobs/"+j.ID) snap := j.Snapshot() writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]}) } func (s *Server) handlePostgresVacuum(w http.ResponseWriter, r *http.Request) { a, ok := authFromContext(r.Context()) if !ok { writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") return } body, ok2 := s.decodePgMaintBody(r) if !ok2 { writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) return } s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresVacuum, map[string]any{ "table": body.Table, "dry_run": body.DryRun, "job_title": "PostgreSQL VACUUM", }) } func (s *Server) handlePostgresVacuumAnalyze(w http.ResponseWriter, r *http.Request) { a, ok := authFromContext(r.Context()) if !ok { writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") return } body, ok2 := s.decodePgMaintBody(r) if !ok2 { writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) return } s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresVacuumAnalyze, map[string]any{ "table": body.Table, "dry_run": body.DryRun, "job_title": "PostgreSQL VACUUM ANALYZE", }) } func (s *Server) handlePostgresAnalyze(w http.ResponseWriter, r *http.Request) { a, ok := authFromContext(r.Context()) if !ok { writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") return } body, ok2 := s.decodePgMaintBody(r) if !ok2 { writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) return } s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresAnalyze, map[string]any{ "table": body.Table, "dry_run": body.DryRun, "job_title": "PostgreSQL ANALYZE", }) } func (s *Server) handlePostgresReindex(w http.ResponseWriter, r *http.Request) { a, ok := authFromContext(r.Context()) if !ok { writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") return } body, ok2 := s.decodePgMaintBody(r) if !ok2 { writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) return } table := body.Table if table == "" { table = body.Index } s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresReindex, map[string]any{ "table": table, "dry_run": body.DryRun, "job_title": "PostgreSQL REINDEX", }) } func (s *Server) handlePostgresCleanup(w http.ResponseWriter, r *http.Request) { a, ok := authFromContext(r.Context()) if !ok { writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") return } body, ok2 := s.decodePgMaintBody(r) if !ok2 { writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) return } policyID := strings.TrimSpace(body.PolicyID) if policyID == "" { policyID = strings.TrimSpace(body.Policy) } if policyID == "" { writeProblem(w, http.StatusBadRequest, "Bad Request", "policy_id is required") return } if _, err := s.store.GetMaintenancePolicy(policyID); err != nil { writeStoreErr(w, err) return } idem := strings.TrimSpace(r.Header.Get("Idempotency-Key")) var idemPtr *string if idem != "" { idemPtr = &idem } j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindMaintenancePolicyRun, idemPtr, nil, map[string]any{ "policy_id": policyID, "dry_run": body.DryRun, "actor_prefix": actorPrefix(a), "job_title": "PostgreSQL cleanup (deprecated path)", }) if err != nil { writeInternalError(w, "postgres_maint_enqueue", err) return } w.Header().Set("Location", "/v1/jobs/"+j.ID) snap := j.Snapshot() writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]}) } func (s *Server) handlePostgresMaintenanceLogs(w http.ResponseWriter, r *http.Request) { a, ok := authFromContext(r.Context()) if !ok || !s.requirePerm(w, a, "bgp:monitoring:read") || !s.requirePostgres(w) { return } cursor := r.URL.Query().Get("cursor") limit := parseLimitQuery(r, 20, 100) ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) defer cancel() items, next, hasMore, err := pgmonitor.ListMaintenanceLogs(ctx, s.pgMonitor.Pool(), cursor, limit) if err != nil { writeInternalError(w, "postgres_maint_logs", err) return } writeJSON(w, http.StatusOK, map[string]any{"items": items, "next_cursor": next, "has_more": hasMore}) } func actorPrefix(a Auth) string { if len(a.Token) >= 8 { return a.Token[:8] } return a.Role }