package jobs import ( "fmt" "strings" "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.Meta["table"].(string) dryRun, _ := j.Meta["dry_run"].(bool) actor, _ := j.Meta["actor_prefix"].(string) 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) { if w == nil || w.PgPool == nil { 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). 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"}) } }