EvoBGP/internal/jobs/postgres_worker.go

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"})
	}
}