EvoBGP/internal/pgmonitor/correlation.go

package pgmonitor

import (
	"context"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

// Correlation builds aligned timeline points from job_audit and overview cache.
func (s *Service) Correlation(ctx context.Context, windowMinutes int) (CorrelationResponse, error) {
	if s == nil || s.pool == nil {
		return CorrelationResponse{}, fmt.Errorf("pgmonitor: postgres not configured")
	}
	if windowMinutes <= 0 {
		windowMinutes = 60
	}
	if windowMinutes > 1440 {
		windowMinutes = 1440
	}
	since := time.Now().UTC().Add(-time.Duration(windowMinutes) * time.Minute)

	rows, err := s.pool.Query(ctx, `
		SELECT date_trunc('minute', finished_at) AS bucket,
		       percentile_cont(0.99) WITHIN GROUP (ORDER BY
		         EXTRACT(EPOCH FROM (finished_at - started_at)) * 1000)
		FROM job_audit
		WHERE finished_at >= $1 AND kind IN ('module_refresh', 'tenant_refresh')
		  AND status = 'succeeded' AND started_at IS NOT NULL
		GROUP BY 1
		ORDER BY 1`, since)
	if err != nil {
		return CorrelationResponse{}, fmt.Errorf("pgmonitor: correlation jobs: %w", err)
	}
	defer rows.Close()

	points := make(map[time.Time]*CorrelationPoint)
	for rows.Next() {
		var bucket time.Time
		var p99 *float64
		if err := rows.Scan(&bucket, &p99); err != nil {
			return CorrelationResponse{}, err
		}
		bucket = bucket.UTC()
		pt := points[bucket]
		if pt == nil {
			pt = &CorrelationPoint{Timestamp: bucket}
			points[bucket] = pt
		}
		if p99 != nil {
			pt.PipelineRefreshP99Ms = *p99
		}
	}

	ov, err := s.Overview(ctx)
	if err == nil && ov.Database.CacheHitPct > 0 {
		now := time.Now().UTC().Truncate(time.Minute)
		pt := points[now]
		if pt == nil {
			pt = &CorrelationPoint{Timestamp: now}
			points[now] = pt
		}
		pt.CacheHitPct = ov.Database.CacheHitPct
	}

	out := make([]CorrelationPoint, 0, len(points))
	for _, p := range points {
		out = append(out, *p)
	}
	// simple sort by time
	for i := 0; i < len(out); i++ {
		for j := i + 1; j < len(out); j++ {
			if out[j].Timestamp.Before(out[i].Timestamp) {
				out[i], out[j] = out[j], out[i]
			}
		}
	}
	return CorrelationResponse{WindowMinutes: windowMinutes, Points: out}, nil
}

// RecordCorrelationSnapshot is a hook for future Prometheus samples (no-op placeholder).
func RecordCorrelationSnapshot(_ *pgxpool.Pool) {}