package jobs import ( "fmt" "evobgp/internal/pgmonitor" ) func (w *Worker) pgService() *pgmonitor.Service { if w == nil || w.PgPool == nil { return nil } return pgmonitor.NewService(w.PgPool) } func (w *Worker) runPostgresMetricsRefresh(j *Job) { s := w.pgService() if s == nil { j.Fail("postgresql not configured") return } ctx, cancel := j.workContext() defer cancel() if err := s.RefreshMetricsSnapshot(ctx); err != nil { j.Fail(err.Error()) return } j.Succeed() } func (w *Worker) runPostgresSlowQueryAgg(j *Job) { s := w.pgService() if s == nil { j.Fail("postgresql not configured") return } ctx, cancel := j.workContext() defer cancel() if err := s.AggregateSlowQueries(ctx, 30); err != nil { j.Fail(err.Error()) return } j.Succeed() } func (w *Worker) runPostgresTableBloat(j *Job) { s := w.pgService() if s == nil { j.Fail("postgresql not configured") return } ctx, cancel := j.workContext() defer cancel() if err := s.EstimateTableBloat(ctx); err != nil { j.Fail(err.Error()) return } j.Succeed() } func (w *Worker) runPostgresIndexUsage(j *Job) { s := w.pgService() if s == nil { j.Fail("postgresql not configured") return } ctx, cancel := j.workContext() defer cancel() if err := s.AnalyzeIndexUsage(ctx); err != nil { j.Fail(err.Error()) return } j.Succeed() } func (w *Worker) runPostgresAutovacuumLag(j *Job) { s := w.pgService() if s == nil { j.Fail("postgresql not configured") return } ctx, cancel := j.workContext() defer cancel() if err := s.DetectAutovacuumLag(ctx); err != nil { j.Fail(err.Error()) return } j.Succeed() } func (w *Worker) runPostgresMaint(j *Job, kind string) { if w == nil || w.PgPool == nil { j.Fail("postgresql not configured") return } table := j.metaString("table") dryRun := j.metaBool("dry_run") actor := j.metaString("actor_prefix") ctx, cancel := j.workContext() defer cancel() auditID, _ := pgmonitor.InsertMaintenanceAudit(ctx, w.PgPool, j.TenantID, actor, kind, table, dryRun) detail, err := pgmonitor.ExecMaintenance(ctx, w.PgPool, kind, table, dryRun) var errMsg *string status := StatusSucceeded if err != nil { s := err.Error() errMsg = &s status = StatusFailed j.Fail(s) } else { j.mergeMeta(map[string]any{"maintenance": detail, "audit_id": auditID}) j.Succeed() } if auditID != "" { _ = pgmonitor.FinishMaintenanceAudit(ctx, w.PgPool, auditID, status, detail, errMsg) } } func (w *Worker) runPostgresCleanup(j *Job) { j.Fail("postgres_cleanup deprecated: configure maintenance_policy in UI and use maintenance_policy_run") } // EnqueuePostgresAnalyzerJobs enqueues periodic analyzer jobs (global tenant id). func EnqueuePostgresAnalyzerJobs(reg *Registry, tenantID string) { if reg == nil || tenantID == "" { return } kinds := []string{ KindPostgresMetricsRefresh, KindPostgresSlowQueryAgg, KindPostgresTableBloat, KindPostgresIndexUsage, KindPostgresAutovacuumLag, } for _, k := range kinds { key := fmt.Sprintf("pgmon-%s-%s", k, tenantID) idem := key _, _, _ = reg.Enqueue(tenantID, k, &idem, nil, map[string]any{"trigger": "scheduler"}) } }