refactor(maintenance): remove hardcoded retention and wire scheduler

RunPeriodicMaintenance и RunCleanup удалены; scheduler политик в StartBackground; deprecated /postgres/cleanup принимает policy_id.

Co-authored-by: Cursor <[email protected]>

internal/dbcli/dbcli.go

@@ -115,22 +115,25 @@ func cmdMaint(args []string, _ string, path string) int {
 
 func cmdCleanup(args []string) int {
 	fs := flag.NewFlagSet("cleanup", flag.ExitOnError)
-	policy := fs.String("policy", "", "cleanup policy name")
+	policyID := fs.String("policy-id", "", "maintenance policy UUID")
 	dryRun := fs.Bool("dry-run", true, "dry run")
-	limit := fs.Int("limit", 10000, "max rows")
 	apiURL := fs.String("api-url", "", "control plane base URL")
 	token := fs.String("token", "", "Bearer token (operator)")
 	_ = fs.Parse(args)
-	if *policy == "" {
+	if *policyID == "" {
-		fmt.Fprintln(os.Stderr, "cleanup: --policy is required")
+		fmt.Fprintln(os.Stderr, "cleanup: --policy-id is required")
 		return 2
 	}
 	if *apiURL == "" || *token == "" {
 		fmt.Fprintln(os.Stderr, "cleanup: --api-url and --token are required")
 		return 2
 	}
-	payload := map[string]any{"policy": *policy, "dry_run": *dryRun, "limit": *limit}
+	path := "/v1/maintenance/run"
-	body, err := apiPOST(*apiURL, *token, "/v1/postgres/cleanup", payload)
+	if *dryRun {
+		path = "/v1/maintenance/dry-run"
+	}
+	payload := map[string]any{"policy_id": *policyID}
+	body, err := apiPOST(*apiURL, *token, path, payload)
 	if err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		return 1

internal/httpapi/routes_postgres_maintenance.go

@@ -47,11 +47,12 @@ func (s *Server) checkPgMaintRateLimit(tenantID, kind string) bool {
 }
 
 type pgMaintBody struct {
-	Table  string `json:"table"`
+	Table    string `json:"table"`
-	DryRun bool   `json:"dry_run"`
+	DryRun   bool   `json:"dry_run"`
-	Index  string `json:"index"`
+	Index    string `json:"index"`
-	Policy string `json:"policy"`
+	Policy   string `json:"policy"`
-	Limit  int    `json:"limit"`
+	PolicyID string `json:"policy_id"`
+	Limit    int    `json:"limit"`
 }
 
 func (s *Server) decodePgMaintBody(r *http.Request) (pgMaintBody, bool) {
@@ -168,14 +169,34 @@ func (s *Server) handlePostgresCleanup(w http.ResponseWriter, r *http.Request) {
 		writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail)
 		return
 	}
-	if strings.TrimSpace(body.Policy) == "" {
+	policyID := strings.TrimSpace(body.PolicyID)
-		writeProblem(w, http.StatusBadRequest, "Bad Request", "policy is required")
+	if policyID == "" {
+		policyID = strings.TrimSpace(body.Policy)
+	}
+	if policyID == "" {
+		writeProblem(w, http.StatusBadRequest, "Bad Request", "policy_id is required")
 		return
 	}
-	s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresCleanup, map[string]any{
+	if _, err := s.store.GetMaintenancePolicy(policyID); err != nil {
-		"policy": body.Policy, "dry_run": body.DryRun, "limit": body.Limit,
+		writeStoreErr(w, err)
-		"job_title": "PostgreSQL cleanup",
+		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) {

internal/httpapi/server.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"net/http"
 	"strings"
+	"time"
 
 	"evobgp/internal/jobs"
 	"evobgp/internal/maintenance"
@@ -107,9 +108,19 @@ 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.
+// StartBackground starts PostgreSQL monitoring and maintenance schedulers until ctx is cancelled.
 func (s *Server) StartBackground(ctx context.Context) {
 	if s != nil && s.pgPool != nil {
 		pgmonitor.StartScheduler(ctx, s.pgPool)
 	}
+	if s != nil && s.maintConfig != nil && s.jobs != nil {
+		maintenance.StartScheduler(ctx, s.maintConfig, func(policyID string, dryRun bool, idem string) {
+			key := idem
+			_, _, _ = s.jobs.Enqueue("", jobs.KindMaintenancePolicyRun, &key, nil, map[string]any{
+				"policy_id": policyID,
+				"dry_run":   dryRun,
+				"trigger":   "scheduler",
+			})
+		}, 30*time.Second)
+	}
 }

internal/ingest/run.go

@@ -17,8 +17,6 @@ type Deps struct {
 	Store store.Backend
 }
 
-var lastMaintenance time.Time
-
 // Run blocks until ctx is cancelled.
 func Run(ctx context.Context, deps *Deps) {
 	cfg := config.Load()
@@ -36,10 +34,6 @@ func Run(ctx context.Context, deps *Deps) {
 			log.Printf("evobgp-ingest: stopped")
 			return
 		case <-t.C:
-			if deps.Store != nil && time.Since(lastMaintenance) > time.Hour {
-				deps.Store.RunPeriodicMaintenance(ctx)
-				lastMaintenance = time.Now()
-			}
 			prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second)
 			err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc)
 			cancel()

internal/jobs/postgres_worker.go

@@ -2,7 +2,6 @@ package jobs
 
 import (
 	"fmt"
-	"strings"
 
 	"evobgp/internal/pgmonitor"
 )
@@ -118,35 +117,7 @@ func (w *Worker) runPostgresMaint(j *Job, kind string) {
 }
 
 func (w *Worker) runPostgresCleanup(j *Job) {
-	if w == nil || w.PgPool == nil {
+	j.Fail("postgres_cleanup deprecated: configure maintenance_policy in UI and use maintenance_policy_run")
-		j.Fail("postgresql not configured")
-		return
-	}
-	policy, _ := j.Meta["policy"].(string)
-	dryRun, _ := j.Meta["dry_run"].(bool)
-	limit := 0
-	if v, ok := j.Meta["limit"].(float64); ok {
-		limit = int(v)
-	}
-	actor, _ := j.Meta["actor_prefix"].(string)
-	ctx, cancel := j.workContext()
-	defer cancel()
-	auditID, _ := pgmonitor.InsertMaintenanceAudit(ctx, w.PgPool, j.TenantID, actor, "cleanup", policy, dryRun)
-	detail, err := pgmonitor.RunCleanup(ctx, w.PgPool, strings.TrimSpace(policy), dryRun, limit)
-	var errMsg *string
-	status := StatusSucceeded
-	if err != nil {
-		s := err.Error()
-		errMsg = &s
-		status = StatusFailed
-		j.Fail(s)
-	} else {
-		j.mergeMeta(map[string]any{"cleanup": detail, "audit_id": auditID})
-		j.Succeed()
-	}
-	if auditID != "" {
-		_ = pgmonitor.FinishMaintenanceAudit(ctx, w.PgPool, auditID, status, detail, errMsg)
-	}
 }
 
 // EnqueuePostgresAnalyzerJobs enqueues periodic analyzer jobs (global tenant id).

internal/pgmonitor/maintenance_audit.go

@@ -3,81 +3,18 @@ package pgmonitor
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
-// CleanupPolicy names safe retention policies.
+// CleanupRequest for deprecated POST /postgres/cleanup (use /v1/maintenance/run).
-type CleanupPolicy string
-
-const (
-	PolicyJobAuditRetention CleanupPolicy = "job_audit_retention"
-	PolicyASNCacheRetention CleanupPolicy = "asn_cache_retention"
-)
-
-// CleanupRequest for POST /postgres/cleanup.
 type CleanupRequest struct {
-	Policy string `json:"policy"`
+	PolicyID string `json:"policy_id"`
-	DryRun bool   `json:"dry_run"`
+	Policy   string `json:"policy"`
-	Limit  int    `json:"limit"`
+	DryRun   bool   `json:"dry_run"`
-}
+	Limit    int    `json:"limit"`
-
-// RunCleanup executes a named retention policy.
-func RunCleanup(ctx context.Context, pool *pgxpool.Pool, policy string, dryRun bool, limit int) (map[string]any, error) {
-	if pool == nil {
-		return nil, fmt.Errorf("pgmonitor: postgres not configured")
-	}
-	if limit <= 0 {
-		limit = 10000
-	}
-	if limit > 100000 {
-		limit = 100000
-	}
-	detail := map[string]any{"policy": policy, "dry_run": dryRun, "limit": limit}
-	switch CleanupPolicy(policy) {
-	case PolicyJobAuditRetention:
-		cutoff := time.Now().UTC().Add(-90 * 24 * time.Hour)
-		if dryRun {
-			var n int64
-			err := pool.QueryRow(ctx, `
-				SELECT count(*) FROM job_audit
-				WHERE created_at < $1 AND status IN ('succeeded', 'failed', 'cancelled')`, cutoff).Scan(&n)
-			detail["would_delete"] = n
-			return detail, err
-		}
-		tag, err := pool.Exec(ctx, `
-			DELETE FROM job_audit
-			WHERE id IN (
-			  SELECT id FROM job_audit
-			  WHERE created_at < $1 AND status IN ('succeeded', 'failed', 'cancelled')
-			  LIMIT $2
-			)`, cutoff, limit)
-		if err != nil {
-			return detail, err
-		}
-		detail["deleted"] = tag.RowsAffected()
-		return detail, nil
-	case PolicyASNCacheRetention:
-		cutoff := time.Now().UTC().Add(-7 * 24 * time.Hour)
-		if dryRun {
-			var n int64
-			err := pool.QueryRow(ctx, `SELECT count(*) FROM asn_prefix_cache WHERE fetched_at < $1`, cutoff).Scan(&n)
-			detail["would_delete"] = n
-			return detail, err
-		}
-		tag, err := pool.Exec(ctx, `
-			DELETE FROM asn_prefix_cache WHERE fetched_at < $1`, cutoff)
-		if err != nil {
-			return detail, err
-		}
-		detail["deleted"] = tag.RowsAffected()
-		return detail, nil
-	default:
-		return nil, fmt.Errorf("pgmonitor: unknown cleanup policy %q", policy)
-	}
 }
 
 // InsertMaintenanceAudit records an audit row at job start.

internal/repository/maintenance.go

@@ -1,29 +1,8 @@
 package repository
 
-import (
+import "context"
-	"context"
-	"time"
-)
 
-const (
+// RunPeriodicMaintenance is a no-op; retention is driven by maintenance_policy rows (UI-configured).
-	jobAuditRetentionDays = 90
-	asnCacheRetentionDays = 7
-)
-
-// RunPeriodicMaintenance prunes stale job_audit and asn_prefix_cache rows (PostgreSQL).
 func (p *Postgres) RunPeriodicMaintenance(ctx context.Context) {
-	if p == nil || p.pool == nil {
+	_ = ctx
-		return
-	}
-	if ctx == nil {
-		ctx = context.Background()
-	}
-	jobCutoff := time.Now().UTC().Add(-time.Duration(jobAuditRetentionDays) * 24 * time.Hour)
-	_, _ = p.pool.Exec(ctx, `
-		DELETE FROM job_audit
-		WHERE created_at < $1
-		  AND status IN ('succeeded', 'failed', 'cancelled')`, jobCutoff)
-	asnCutoff := time.Now().UTC().Add(-time.Duration(asnCacheRetentionDays) * 24 * time.Hour)
-	_, _ = p.pool.Exec(ctx, `
-		DELETE FROM asn_prefix_cache WHERE fetched_at < $1`, asnCutoff)
 }