feat(db): implement PostgreSQL monitoring and maintenance features

Added PostgreSQL monitoring and maintenance capabilities to the API, including new endpoints for instance-level metrics, maintenance operations, and job scheduling. Updated the HTTP API to support PostgreSQL monitoring routes and integrated a background scheduler for metrics collection. Enhanced the CLI with database commands for maintenance tasks. Updated documentation to reflect these changes.

internal/httpapi/bootstrap.go

@@ -51,7 +51,7 @@ func BootstrapWorkers(ctx context.Context, opts Options) (store.Backend, *jobs.R
 	}
 
 	cdnHTTP := NewCDNHTTPClient()
-	wk := &jobs.Worker{Store: backend, HTTPClient: cdnHTTP}
+	wk := &jobs.Worker{Store: backend, PgPool: pool, HTTPClient: cdnHTTP}
 	reg := jobs.NewRegistry(wk.Process)
 	wk.Registry = reg
 	if pool != nil {

internal/httpapi/routes.go

@@ -76,6 +76,8 @@ func (s *Server) registerV1(m *http.ServeMux) {
 	m.HandleFunc("GET /speakers/{speaker_id}/bundle/{revision_id}", s.handleNodeBundle)
 	m.HandleFunc("POST /nodes/enroll", s.handleNodeEnroll)
 	s.registerCRUDRoutes(m)
+	s.registerPostgresMonitoringRoutes(m)
+	s.registerPostgresMaintenanceRoutes(m)
 }
 
 func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {

internal/httpapi/routes_postgres_maintenance.go

@@ -0,0 +1,203 @@
+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)
+}
+
+func (s *Server) requireOperatorStrict(w http.ResponseWriter, a Auth) bool {
+	if strings.ToLower(a.Role) != "operator" {
+		writeProblem(w, http.StatusForbidden, "Forbidden", "operator role required")
+		return false
+	}
+	return true
+}
+
+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"`
+	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
+	}
+	if strings.TrimSpace(body.Policy) == "" {
+		writeProblem(w, http.StatusBadRequest, "Bad Request", "policy is required")
+		return
+	}
+	s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresCleanup, map[string]any{
+		"policy": body.Policy, "dry_run": body.DryRun, "limit": body.Limit,
+		"job_title": "PostgreSQL cleanup",
+	})
+}
+
+func (s *Server) handlePostgresMaintenanceLogs(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok || !s.requireAtLeast(w, a, "viewer") || !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
+}

internal/httpapi/routes_postgres_monitoring.go

@@ -0,0 +1,130 @@
+package httpapi
+
+import (
+	"context"
+	"net/http"
+	"strconv"
+	"time"
+)
+
+func (s *Server) registerPostgresMonitoringRoutes(m *http.ServeMux) {
+	m.HandleFunc("GET /monitoring/postgres/overview", s.handlePostgresOverview)
+	m.HandleFunc("GET /monitoring/postgres/queries", s.handlePostgresQueries)
+	m.HandleFunc("GET /monitoring/postgres/locks", s.handlePostgresLocks)
+	m.HandleFunc("GET /monitoring/postgres/tables", s.handlePostgresTables)
+	m.HandleFunc("GET /monitoring/postgres/recommendations", s.handlePostgresRecommendations)
+	m.HandleFunc("GET /monitoring/correlation", s.handleMonitoringCorrelation)
+}
+
+func (s *Server) requirePostgres(w http.ResponseWriter) bool {
+	if s.pgMonitor == nil {
+		writeProblem(w, http.StatusServiceUnavailable, "Unavailable", "postgresql backend required")
+		return false
+	}
+	return true
+}
+
+func parseLimitQuery(r *http.Request, def, max int) int {
+	if v := r.URL.Query().Get("limit"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil {
+			return n
+		}
+	}
+	return def
+}
+
+func (s *Server) handlePostgresOverview(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) {
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
+	defer cancel()
+	out, err := s.pgMonitor.Overview(ctx)
+	if err != nil {
+		writeInternalError(w, "postgres_overview", err)
+		return
+	}
+	writeJSON(w, http.StatusOK, out)
+}
+
+func (s *Server) handlePostgresQueries(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) {
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
+	defer cancel()
+	out, err := s.pgMonitor.TopQueries(ctx, parseLimitQuery(r, 20, 100))
+	if err != nil {
+		writeInternalError(w, "postgres_queries", err)
+		return
+	}
+	writeJSON(w, http.StatusOK, out)
+}
+
+func (s *Server) handlePostgresLocks(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) {
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
+	defer cancel()
+	out, err := s.pgMonitor.Locks(ctx)
+	if err != nil {
+		writeInternalError(w, "postgres_locks", err)
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"items": out})
+}
+
+func (s *Server) handlePostgresTables(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) {
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
+	defer cancel()
+	out, err := s.pgMonitor.Tables(ctx, parseLimitQuery(r, 20, 100))
+	if err != nil {
+		writeInternalError(w, "postgres_tables", err)
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"items": out})
+}
+
+func (s *Server) handlePostgresRecommendations(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) {
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
+	defer cancel()
+	out, err := s.pgMonitor.Recommendations(ctx)
+	if err != nil {
+		writeInternalError(w, "postgres_recommendations", err)
+		return
+	}
+	writeJSON(w, http.StatusOK, out)
+}
+
+func (s *Server) handleMonitoringCorrelation(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) {
+		return
+	}
+	window := 60
+	if v := r.URL.Query().Get("window"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil {
+			window = n
+		}
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
+	defer cancel()
+	out, err := s.pgMonitor.Correlation(ctx, window)
+	if err != nil {
+		writeInternalError(w, "monitoring_correlation", err)
+		return
+	}
+	writeJSON(w, http.StatusOK, out)
+}

internal/httpapi/routes_postgres_test.go

@@ -0,0 +1,21 @@
+package httpapi
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestPostgresOverviewMemoryBackend503(t *testing.T) {
+	srv, err := New(Options{SeedDemo: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	req := httptest.NewRequest(http.MethodGet, "/v1/monitoring/postgres/overview", nil)
+	req.Header.Set("Authorization", "Bearer dev")
+	rec := httptest.NewRecorder()
+	srv.Handler().ServeHTTP(rec, req)
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String())
+	}
+}

internal/httpapi/server.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 
 	"evobgp/internal/jobs"
+	"evobgp/internal/pgmonitor"
 	"evobgp/internal/store"
 
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -19,6 +20,7 @@ import (
 type Server struct {
 	store       store.Backend
 	pgPool      *pgxpool.Pool
+	pgMonitor   *pgmonitor.Service
 	jobs        *jobs.Registry
 	bundlePriv  ed25519.PrivateKey
 	keyResolver *apiKeyResolver
@@ -63,9 +65,14 @@ func New(opts Options) (*Server, error) {
 	if err != nil {
 		return nil, err
 	}
+	var pgMon *pgmonitor.Service
+	if pool != nil {
+		pgMon = pgmonitor.NewService(pool)
+	}
 	s := &Server{
 		store:       backend,
 		pgPool:      pool,
+		pgMonitor:   pgMon,
 		jobs:        reg,
 		bundlePriv:  priv,
 		keyResolver: resolver,
@@ -89,3 +96,10 @@ func (s *Server) Store() store.Backend { return s.store }
 
 // Jobs exposes the in-process async job registry (for scheduler / evobgp-all).
 func (s *Server) Jobs() *jobs.Registry { return s.jobs }
+
+// StartBackground starts PostgreSQL monitoring scheduler until ctx is cancelled.
+func (s *Server) StartBackground(ctx context.Context) {
+	if s != nil && s.pgPool != nil {
+		pgmonitor.StartScheduler(ctx, s.pgPool)
+	}
+}