// This file fetches, validates, caches, normalizes, and archives O11Y metric responses.
package evaluation

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"
)

const (
	O11YExportVersionV1Alpha1 = "v1alpha1"

	defaultO11YHTTPTimeout = 30 * time.Second
	maxO11YResponseBytes   = 64 << 20
	o11yCacheKeyVersion    = "v1"

	metricInstanceLabel = "instance"
)

type O11YMetricSource string

const (
	O11YMetricSourceApplication    O11YMetricSource = "application"
	O11YMetricSourceInfrastructure O11YMetricSource = "k8s-infrastructure"
)

type O11YQueryType string

const (
	O11YQueryTypeRange   O11YQueryType = "range"
	O11YQueryTypeInstant O11YQueryType = "instant"
)

type O11YTargetSpec struct {
	ClusterID string `json:"clusterId"`
	Vendor    string `json:"vendor"`
	Region    string `json:"region"`
}

type O11YCaseSpec struct {
	ID         string         `json:"id"`
	Component  string         `json:"component"`
	DecisionAt time.Time      `json:"decisionAt"`
	Current    Capacity       `json:"current"`
	Baseline   Capacity       `json:"baseline"`
	Oracle     Oracle         `json:"oracle"`
	Annotation CaseAnnotation `json:"annotation,omitempty"`
}

type O11YQueryWindow struct {
	Start       time.Time `json:"start"`
	End         time.Time `json:"end"`
	StepSeconds int64     `json:"stepSeconds"`
}

type O11YQuerySpec struct {
	Name          string        `json:"name"`
	PromQL        string        `json:"promql"`
	Type          O11YQueryType `json:"type"`
	Required      bool          `json:"required"`
	KeepLabels    []string      `json:"keepLabels,omitempty"`
	MaxGapSeconds int64         `json:"maxGapSeconds,omitempty"`
}

type O11YSourceSpec struct {
	Type    O11YMetricSource  `json:"type"`
	BaseURL string            `json:"baseUrl"`
	Headers map[string]string `json:"headers,omitempty"`
	Queries []O11YQuerySpec   `json:"queries"`
}

// O11YExportSpec describes acquisition and human labeling separately from the
// policy input written to CaseBundle.
type O11YExportSpec struct {
	Version string           `json:"version"`
	Target  O11YTargetSpec   `json:"target"`
	Case    O11YCaseSpec     `json:"case"`
	Window  O11YQueryWindow  `json:"window"`
	Sources []O11YSourceSpec `json:"sources"`
}

type O11YExportOptions struct {
	Client             *http.Client
	BearerToken        string
	RequireBearerToken bool
	ArchivePath        string
	CasePath           string
	CacheDir           string
}

// O11YArchivedQuery stores one request and its unmodified response.
type O11YArchivedQuery struct {
	Name          string           `json:"name"`
	Source        O11YMetricSource `json:"source"`
	Type          O11YQueryType    `json:"type"`
	PromQL        string           `json:"promql"`
	RequestDigest string           `json:"requestDigest"`
	SeriesCount   int              `json:"seriesCount"`
	SampleCount   int              `json:"sampleCount"`
	Error         string           `json:"error,omitempty"`
	Response      json.RawMessage  `json:"response,omitempty"`

	// CacheHit describes the current run, not the historical case.
	CacheHit bool `json:"-"`
}

// O11YArchive is the durable source for one historical case. CaseBundle is
// derived from this data and the committed case specification.
type O11YArchive struct {
	Version    string              `json:"version"`
	CaseID     string              `json:"caseId"`
	Component  string              `json:"component"`
	Target     O11YTargetSpec      `json:"target"`
	Window     O11YQueryWindow     `json:"window"`
	Annotation CaseAnnotation      `json:"annotation,omitempty"`
	Queries    []O11YArchivedQuery `json:"queries"`
	Quality    DataQuality         `json:"quality"`
	Errors     []string            `json:"errors,omitempty"`
}

type prometheusResponse struct {
	Status    string         `json:"status"`
	IsPartial bool           `json:"isPartial"`
	ErrorType string         `json:"errorType"`
	Error     string         `json:"error"`
	Data      prometheusData `json:"data"`
}

type prometheusData struct {
	ResultType string             `json:"resultType"`
	Result     []prometheusResult `json:"result"`
}

type prometheusResult struct {
	Metric map[string]string   `json:"metric"`
	Values [][]json.RawMessage `json:"values"`
	Value  []json.RawMessage   `json:"value"`
}

// FetchO11Y collects every response for one case into a single archive and
// derives the CaseBundle used by manually constructed cases.
func FetchO11Y(ctx context.Context, spec *O11YExportSpec, options O11YExportOptions) (*CaseBundle, *O11YArchive, error) {
	if err := validateO11YExportSpec(spec); err != nil {
		return nil, nil, err
	}
	if options.ArchivePath == "" {
		return nil, nil, fmt.Errorf("evaluation: O11Y archive path is required")
	}
	if options.Client == nil {
		options.Client = &http.Client{Timeout: defaultO11YHTTPTimeout}
	}
	if err := makePrivateDir(filepath.Dir(options.ArchivePath)); err != nil {
		return nil, nil, fmt.Errorf("evaluation: prepare O11Y archive directory: %w", err)
	}
	if options.CasePath != "" {
		if err := makePrivateDir(filepath.Dir(options.CasePath)); err != nil {
			return nil, nil, fmt.Errorf("evaluation: prepare O11Y case directory: %w", err)
		}
		if err := os.Remove(options.CasePath); err != nil && !errors.Is(err, os.ErrNotExist) {
			return nil, nil, fmt.Errorf("evaluation: remove stale case: %w", err)
		}
	}
	if options.CacheDir != "" {
		if err := makePrivateDir(options.CacheDir); err != nil {
			return nil, nil, fmt.Errorf("evaluation: prepare O11Y cache directory: %w", err)
		}
	}

	archive := &O11YArchive{
		Version:    O11YExportVersionV1Alpha1,
		CaseID:     spec.Case.ID,
		Component:  spec.Case.Component,
		Target:     spec.Target,
		Window:     spec.Window,
		Annotation: spec.Case.Annotation,
		Queries:    make([]O11YArchivedQuery, 0, queryCount(spec.Sources)),
		Quality: DataQuality{
			MissingSeries: make([]string, 0),
		},
	}
	for i := range spec.Sources {
		source := &spec.Sources[i]
		for j := range source.Queries {
			query := &source.Queries[j]
			record := O11YArchivedQuery{
				Name:   query.Name,
				Source: source.Type,
				Type:   query.Type,
				PromQL: query.PromQL,
			}

			requestURL, err := buildO11YQueryURL(source, query, &spec.Window, spec.Case.DecisionAt)
			if err != nil {
				record.Error = err.Error()
				archive.Errors = append(archive.Errors, fmt.Sprintf("%s: %v", query.Name, err))
				archive.Queries = append(archive.Queries, record)
				continue
			}
			record.RequestDigest = o11yRequestDigest(requestURL)

			body, cacheHit, cacheErr := readO11YCache(options.CacheDir, record.RequestDigest)
			if cacheErr != nil {
				record.Error = cacheErr.Error()
				archive.Errors = append(archive.Errors, fmt.Sprintf("%s: %v", query.Name, cacheErr))
				archive.Queries = append(archive.Queries, record)
				continue
			}
			record.CacheHit = cacheHit

			var fetchErr error
			if !cacheHit {
				if options.RequireBearerToken && options.BearerToken == "" {
					fetchErr = fmt.Errorf("cache miss requires O11Y bearer token")
				} else {
					body, fetchErr = fetchO11YQuery(ctx, options.Client, options.BearerToken, requestURL)
				}
			}
			if len(body) > 0 {
				record.Response = append(json.RawMessage(nil), body...)
			}
			if fetchErr != nil {
				record.Error = fetchErr.Error()
				archive.Errors = append(archive.Errors, fmt.Sprintf("%s: %v", query.Name, fetchErr))
				archive.Queries = append(archive.Queries, record)
				if ctx.Err() != nil {
					return finishO11YExportError(&options, archive, ctx.Err())
				}
				continue
			}

			response, err := decodePrometheusResponse(body)
			if err != nil {
				record.Error = err.Error()
				archive.Errors = append(archive.Errors, fmt.Sprintf("%s: %v", query.Name, err))
				archive.Queries = append(archive.Queries, record)
				continue
			}
			if response.IsPartial {
				archive.Quality.IsPartial = true
				record.Error = "partial Prometheus response"
				archive.Queries = append(archive.Queries, record)
				continue
			}

			// The response is authoritative for a fixed historical window, so
			// cache it here rather than after the gap and required-series
			// checks below. Those checks describe the data itself: re-fetching
			// cannot change their outcome, and caching only on success would
			// force every rejected case to hit O11Y again on each rerun.
			if !cacheHit && options.CacheDir != "" {
				if err := writeO11YCache(options.CacheDir, record.RequestDigest, body); err != nil {
					return nil, archive, fmt.Errorf("evaluation: cache O11Y response %q: %w", query.Name, err)
				}
			}

			series, sampleCount, err := normalizePrometheusResponse(response, query, &spec.Window, spec.Case.DecisionAt)
			if err != nil {
				record.Error = err.Error()
				archive.Errors = append(archive.Errors, fmt.Sprintf("%s: %v", query.Name, err))
				archive.Queries = append(archive.Queries, record)
				continue
			}
			record.SeriesCount = len(series)
			record.SampleCount = sampleCount
			maxGap := time.Duration(query.MaxGapSeconds) * time.Second
			if len(series) > 0 && maxGap > 0 {
				if err := validateMetricGaps(series, maxGap); err != nil {
					record.Error = err.Error()
					archive.Quality.GappedSeries = append(archive.Quality.GappedSeries, query.Name)
					archive.Queries = append(archive.Queries, record)
					continue
				}
			}
			if query.Required && len(series) == 0 {
				archive.Queries = append(archive.Queries, record)
				archive.Quality.MissingSeries = append(archive.Quality.MissingSeries, query.Name)
				continue
			}
			// The decision reads the state at decisionAt, so that is where
			// coverage has to hold. A per-instance query reporting fewer
			// instances than the cluster had replicas is silently unusable:
			// skew and hotspot checks would read the missing instances as idle
			// rather than as absent.
			if query.Required && hasInstanceLabel(query) {
				live := countInstances(series)
				if maxGap > 0 {
					live = countLiveInstances(series, spec.Case.DecisionAt, maxGap)
				}
				if live < int(spec.Case.Current.Replicas) {
					record.Error = fmt.Sprintf("%d instances reporting at decisionAt, want at least %d", live, spec.Case.Current.Replicas)
					archive.Quality.IncompleteSeries = append(archive.Quality.IncompleteSeries, query.Name)
					archive.Queries = append(archive.Queries, record)
					continue
				}
			}
			archive.Queries = append(archive.Queries, record)
		}
	}

	if archive.Quality.IsPartial {
		return finishO11YExportError(&options, archive, fmt.Errorf("partial Prometheus response"))
	}
	if len(archive.Quality.MissingSeries) > 0 {
		return finishO11YExportError(&options, archive, fmt.Errorf("missing required series: %v", archive.Quality.MissingSeries))
	}
	if len(archive.Quality.GappedSeries) > 0 {
		return finishO11YExportError(&options, archive, fmt.Errorf("gapped required series: %v", archive.Quality.GappedSeries))
	}
	if len(archive.Quality.IncompleteSeries) > 0 {
		return finishO11YExportError(&options, archive, fmt.Errorf("incomplete required series: %v", archive.Quality.IncompleteSeries))
	}
	if len(archive.Errors) > 0 {
		return finishO11YExportError(&options, archive, fmt.Errorf("O11Y query errors: %s", strings.Join(archive.Errors, "; ")))
	}

	bundle, err := ReplayO11YArchive(spec, archive)
	if err != nil {
		return finishO11YExportError(&options, archive, err)
	}
	if err := writeO11YArchive(options.ArchivePath, archive); err != nil {
		return nil, archive, err
	}
	if options.CasePath != "" {
		if err := writePrivateJSON(options.CasePath, bundle); err != nil {
			return nil, archive, fmt.Errorf("evaluation: write O11Y case: %w", err)
		}
	}
	return bundle, archive, nil
}

// writeO11YArchive keeps embedded responses compact to avoid multiplying the
// committed archive size with indentation.
func writeO11YArchive(path string, archive *O11YArchive) error {
	data, err := json.Marshal(archive)
	if err != nil {
		return fmt.Errorf("evaluation: encode O11Y archive: %w", err)
	}
	if err := writePrivateFile(path, append(data, '\n')); err != nil {
		return fmt.Errorf("evaluation: write O11Y archive: %w", err)
	}
	return nil
}

// SeedO11YCache rebuilds the request-keyed cache from committed archives.
func SeedO11YCache(casesDir, cacheDir string) (int, error) {
	paths, err := filepath.Glob(filepath.Join(casesDir, "*", PointInTimeCasesDirectory, "*.json"))
	if err != nil {
		return 0, fmt.Errorf("evaluation: scan case archives: %w", err)
	}
	if len(paths) == 0 {
		return 0, fmt.Errorf("evaluation: no case archives under %q", casesDir)
	}
	if err := makePrivateDir(cacheDir); err != nil {
		return 0, fmt.Errorf("evaluation: prepare cache directory: %w", err)
	}

	sort.Strings(paths)
	seeded := 0
	for _, path := range paths {
		data, err := os.ReadFile(path)
		if err != nil {
			return seeded, fmt.Errorf("evaluation: read archive %q: %w", path, err)
		}
		var archive O11YArchive
		if err := json.Unmarshal(data, &archive); err != nil {
			return seeded, fmt.Errorf("evaluation: decode archive %q: %w", path, err)
		}
		if archive.Version != O11YExportVersionV1Alpha1 {
			return seeded, fmt.Errorf("evaluation: archive %q has unsupported version %q", path, archive.Version)
		}

		for i := range archive.Queries {
			query := &archive.Queries[i]
			if query.RequestDigest == "" || len(query.Response) == 0 {
				continue
			}
			if err := writeO11YCache(cacheDir, query.RequestDigest, query.Response); err != nil {
				return seeded, fmt.Errorf("evaluation: seed cache for %q: %w", archive.CaseID, err)
			}
			seeded++
		}
	}
	return seeded, nil
}

func validateO11YExportSpec(spec *O11YExportSpec) error {
	if spec == nil {
		return fmt.Errorf("evaluation: O11Y export spec is required")
	}
	if spec.Version != O11YExportVersionV1Alpha1 {
		return fmt.Errorf("evaluation: O11Y export version %q is unsupported", spec.Version)
	}
	if spec.Target.ClusterID == "" || spec.Target.Vendor == "" || spec.Target.Region == "" {
		return fmt.Errorf("evaluation: O11Y target clusterId, vendor, and region are required")
	}
	if spec.Case.ID == "" {
		return fmt.Errorf("evaluation: O11Y case id is required")
	}
	if spec.Case.Component == "" {
		return fmt.Errorf("evaluation: O11Y case component is required")
	}
	if spec.Case.DecisionAt.IsZero() {
		return fmt.Errorf("evaluation: O11Y case decisionAt is required")
	}
	if err := validateCaseCapacity(&spec.Case.Current, "case.current"); err != nil {
		return err
	}
	if err := validateCaseCapacity(&spec.Case.Baseline, "case.baseline"); err != nil {
		return err
	}
	switch spec.Case.Oracle.Class {
	case OracleClassUnnecessaryScaleOut, OracleClassUsefulScaleOut:
	default:
		return fmt.Errorf("evaluation: O11Y case oracle class %q is unsupported", spec.Case.Oracle.Class)
	}
	if err := validateCaseCapacity(&spec.Case.Oracle.Expected, "case.oracle.expected"); err != nil {
		return err
	}
	if spec.Window.Start.IsZero() || spec.Window.End.IsZero() {
		return fmt.Errorf("evaluation: O11Y query window start and end are required")
	}
	if !spec.Window.Start.Before(spec.Window.End) {
		return fmt.Errorf("evaluation: O11Y query window start must be before end")
	}
	if spec.Window.Start.After(spec.Case.DecisionAt) || spec.Window.End.Before(spec.Case.DecisionAt) {
		return fmt.Errorf("evaluation: O11Y query window must contain decisionAt")
	}
	if spec.Window.StepSeconds < 1 {
		return fmt.Errorf("evaluation: O11Y query stepSeconds must be greater than zero")
	}
	if len(spec.Sources) == 0 {
		return fmt.Errorf("evaluation: O11Y sources are required")
	}

	queryNames := make(map[string]struct{})
	for i := range spec.Sources {
		source := &spec.Sources[i]
		switch source.Type {
		case O11YMetricSourceApplication, O11YMetricSourceInfrastructure:
		default:
			return fmt.Errorf("evaluation: O11Y sources[%d].type %q is unsupported", i, source.Type)
		}
		parsedURL, err := url.Parse(source.BaseURL)
		if err != nil || parsedURL.Scheme == "" || parsedURL.Host == "" {
			return fmt.Errorf("evaluation: O11Y sources[%d].baseUrl is invalid", i)
		}
		if parsedURL.Scheme != "http" && parsedURL.Scheme != "https" {
			return fmt.Errorf("evaluation: O11Y sources[%d].baseUrl scheme %q is unsupported", i, parsedURL.Scheme)
		}
		if err := validateO11YQueries(i, source.Queries, queryNames); err != nil {
			return err
		}
	}
	return nil
}

func validateO11YQueries(sourceIndex int, queries []O11YQuerySpec, queryNames map[string]struct{}) error {
	if len(queries) == 0 {
		return fmt.Errorf("evaluation: O11Y sources[%d].queries are required", sourceIndex)
	}
	for j := range queries {
		query := &queries[j]
		if !isSafeQueryName(query.Name) {
			return fmt.Errorf("evaluation: O11Y sources[%d].queries[%d].name %q is invalid", sourceIndex, j, query.Name)
		}
		if _, duplicate := queryNames[query.Name]; duplicate {
			return fmt.Errorf("evaluation: O11Y query name %q is duplicated", query.Name)
		}
		queryNames[query.Name] = struct{}{}
		if query.PromQL == "" {
			return fmt.Errorf("evaluation: O11Y query %q promql is required", query.Name)
		}
		switch query.Type {
		case O11YQueryTypeRange, O11YQueryTypeInstant:
		default:
			return fmt.Errorf("evaluation: O11Y query %q type %q is unsupported", query.Name, query.Type)
		}
		if query.MaxGapSeconds < 0 {
			return fmt.Errorf("evaluation: O11Y query %q maxGapSeconds cannot be negative", query.Name)
		}
		if query.Type == O11YQueryTypeInstant && query.MaxGapSeconds != 0 {
			return fmt.Errorf("evaluation: O11Y instant query %q cannot set maxGapSeconds", query.Name)
		}
	}
	return nil
}

func buildO11YQueryURL(source *O11YSourceSpec, query *O11YQuerySpec, window *O11YQueryWindow, decisionAt time.Time) (string, error) {
	endpoint := "query_range"
	if query.Type == O11YQueryTypeInstant {
		endpoint = "query"
	}
	requestURL, err := url.Parse(strings.TrimRight(source.BaseURL, "/") + "/" + endpoint)
	if err != nil {
		return "", fmt.Errorf("build request URL: %w", err)
	}
	params := requestURL.Query()
	params.Set("query", query.PromQL)
	if query.Type == O11YQueryTypeRange {
		params.Set("start", strconv.FormatInt(window.Start.Unix(), 10))
		params.Set("end", strconv.FormatInt(window.End.Unix(), 10))
		params.Set("step", strconv.FormatInt(window.StepSeconds, 10))
	} else {
		params.Set("time", strconv.FormatInt(decisionAt.Unix(), 10))
	}
	requestURL.RawQuery = params.Encode()
	return requestURL.String(), nil
}

func fetchO11YQuery(ctx context.Context, client *http.Client, bearerToken, requestURL string) ([]byte, error) {
	return fetchO11YQueryWithHeaders(ctx, client, bearerToken, requestURL, nil)
}

func fetchO11YQueryWithHeaders(ctx context.Context, client *http.Client, bearerToken, requestURL string, headers map[string]string) ([]byte, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, requestURL, nil)
	if err != nil {
		return nil, fmt.Errorf("build request: %w", err)
	}
	if bearerToken != "" {
		req.Header.Set("Authorization", "Bearer "+bearerToken)
	}
	for name, value := range headers {
		req.Header.Set(name, value)
	}

	resp, err := client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("request: %w", err)
	}
	defer resp.Body.Close()

	body, err := io.ReadAll(io.LimitReader(resp.Body, maxO11YResponseBytes+1))
	if err != nil {
		return nil, fmt.Errorf("read response: %w", err)
	}
	if len(body) > maxO11YResponseBytes {
		return nil, fmt.Errorf("response exceeds %d bytes", maxO11YResponseBytes)
	}
	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
		return body, fmt.Errorf("HTTP status %s", resp.Status)
	}
	return body, nil
}

func o11yRequestDigest(requestURL string) string {
	return o11yRequestDigestWithHeaders(requestURL, nil)
}

func o11yRequestDigestWithHeaders(requestURL string, headers map[string]string) string {
	names := make([]string, 0, len(headers))
	for name := range headers {
		names = append(names, name)
	}
	sort.Strings(names)

	var key strings.Builder
	key.WriteString(o11yCacheKeyVersion + "\nGET\n" + requestURL)
	for _, name := range names {
		key.WriteString("\n" + name + ":" + headers[name])
	}
	digest := sha256.Sum256([]byte(key.String()))
	return fmt.Sprintf("%x", digest)
}

func readO11YCache(cacheDir, requestDigest string) ([]byte, bool, error) {
	if cacheDir == "" {
		return nil, false, nil
	}
	body, err := os.ReadFile(filepath.Join(cacheDir, requestDigest+".json"))
	if errors.Is(err, os.ErrNotExist) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, fmt.Errorf("read O11Y cache: %w", err)
	}
	return body, true, nil
}

func writeO11YCache(cacheDir, requestDigest string, body []byte) error {
	return writePrivateFile(filepath.Join(cacheDir, requestDigest+".json"), body)
}

func decodePrometheusResponse(body []byte) (*prometheusResponse, error) {
	var response prometheusResponse
	if err := json.Unmarshal(body, &response); err != nil {
		return nil, fmt.Errorf("decode Prometheus response: %w", err)
	}
	if response.Status != "success" {
		return nil, fmt.Errorf("Prometheus status %q: %s %s", response.Status, response.ErrorType, response.Error)
	}
	return &response, nil
}

func normalizePrometheusResponse(response *prometheusResponse, query *O11YQuerySpec, window *O11YQueryWindow, decisionAt time.Time) ([]MetricSeries, int, error) {
	wantResultType := "matrix"
	if query.Type == O11YQueryTypeInstant {
		wantResultType = "vector"
	}
	if response.Data.ResultType != wantResultType {
		return nil, 0, fmt.Errorf("resultType %q, want %q", response.Data.ResultType, wantResultType)
	}

	series := make([]MetricSeries, 0, len(response.Data.Result))
	sampleCount := 0
	for i := range response.Data.Result {
		result := &response.Data.Result[i]
		samples := make([]MetricSample, 0, len(result.Values)+1)
		if query.Type == O11YQueryTypeRange {
			for j := range result.Values {
				sample, err := decodePrometheusSample(result.Values[j])
				if err != nil {
					return nil, 0, fmt.Errorf("series[%d].values[%d]: %w", i, j, err)
				}
				if sample.Timestamp.Before(window.Start) || sample.Timestamp.After(window.End) || sample.Timestamp.After(decisionAt) {
					continue
				}
				samples = append(samples, sample)
			}
		} else {
			sample, err := decodePrometheusSample(result.Value)
			if err != nil {
				return nil, 0, fmt.Errorf("series[%d].value: %w", i, err)
			}
			if !sample.Timestamp.Before(window.Start) && !sample.Timestamp.After(window.End) && !sample.Timestamp.After(decisionAt) {
				samples = append(samples, sample)
			}
		}
		if len(samples) == 0 {
			continue
		}
		sort.Slice(samples, func(i, j int) bool {
			return samples[i].Timestamp.Before(samples[j].Timestamp)
		})
		series = append(series, MetricSeries{
			Name:    query.Name,
			Labels:  filterMetricLabels(result.Metric, query.KeepLabels),
			Samples: samples,
		})
		sampleCount += len(samples)
	}
	return series, sampleCount, nil
}

func decodePrometheusSample(raw []json.RawMessage) (MetricSample, error) {
	if len(raw) != 2 {
		return MetricSample{}, fmt.Errorf("sample must contain timestamp and value")
	}
	timestampValue, err := parsePrometheusFloat(raw[0])
	if err != nil {
		return MetricSample{}, fmt.Errorf("timestamp: %w", err)
	}
	if math.IsNaN(timestampValue) || math.IsInf(timestampValue, 0) {
		return MetricSample{}, fmt.Errorf("timestamp must be finite")
	}
	value, err := parsePrometheusFloat(raw[1])
	if err != nil {
		return MetricSample{}, fmt.Errorf("value: %w", err)
	}
	if math.IsNaN(value) || math.IsInf(value, 0) {
		return MetricSample{}, fmt.Errorf("value must be finite")
	}
	seconds, fraction := math.Modf(timestampValue)
	return MetricSample{
		Timestamp: time.Unix(int64(seconds), int64(math.Round(fraction*float64(time.Second)))).UTC(),
		Value:     value,
	}, nil
}

// validateMetricGaps rejects a break in collection inside one series.
//
// A series that starts or ends part-way through the window is deliberately not
// a gap. Pods are created and removed during a scaling event, so a new pod's
// series begins mid-window and a removed pod's series ends mid-window. Treating
// either as a data-quality failure rejects exactly the cases that carry the
// most signal. Coverage at the decision point is enforced separately, by
// countLiveInstances.
func validateMetricGaps(series []MetricSeries, maxGap time.Duration) error {
	for i := range series {
		samples := series[i].Samples
		for j := 1; j < len(samples); j++ {
			if samples[j].Timestamp.Sub(samples[j-1].Timestamp) > maxGap {
				return fmt.Errorf("series[%d] has a gap longer than %s", i, maxGap)
			}
		}
	}
	return nil
}

// countLiveInstances counts distinct instances still reporting within maxGap of
// the decision point. An instance whose series stopped earlier in the window
// was removed before the decision was made, so it must not count towards the
// coverage that decision relied on.
func countLiveInstances(series []MetricSeries, decisionAt time.Time, maxGap time.Duration) int {
	live := make(map[string]struct{}, len(series))
	for i := range series {
		samples := series[i].Samples
		instance := series[i].Labels[metricInstanceLabel]
		if instance != "" && decisionAt.Sub(samples[len(samples)-1].Timestamp) <= maxGap {
			live[instance] = struct{}{}
		}
	}
	return len(live)
}

func parsePrometheusFloat(raw json.RawMessage) (float64, error) {
	value := strings.Trim(string(raw), "\"")
	parsed, err := strconv.ParseFloat(value, 64)
	if err != nil {
		return 0, fmt.Errorf("parse %q: %w", value, err)
	}
	return parsed, nil
}

func hasInstanceLabel(query *O11YQuerySpec) bool {
	for _, label := range query.KeepLabels {
		if label == metricInstanceLabel {
			return true
		}
	}
	return false
}

func countInstances(series []MetricSeries) int {
	instances := make(map[string]struct{}, len(series))
	for i := range series {
		if instance := series[i].Labels[metricInstanceLabel]; instance != "" {
			instances[instance] = struct{}{}
		}
	}
	return len(instances)
}

func filterMetricLabels(labels map[string]string, keep []string) map[string]string {
	if keep == nil {
		filtered := make(map[string]string, len(labels))
		for key, value := range labels {
			filtered[key] = value
		}
		return filtered
	}
	filtered := make(map[string]string, len(keep))
	for _, key := range keep {
		if value, ok := labels[key]; ok {
			filtered[key] = value
		}
	}
	return filtered
}

func sortMetricSeries(series []MetricSeries) {
	sort.Slice(series, func(i, j int) bool {
		if series[i].Name != series[j].Name {
			return series[i].Name < series[j].Name
		}
		return metricLabelsKey(series[i].Labels) < metricLabelsKey(series[j].Labels)
	})
}

func metricLabelsKey(labels map[string]string) string {
	keys := make([]string, 0, len(labels))
	for key := range labels {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	var builder strings.Builder
	for _, key := range keys {
		builder.WriteString(key)
		builder.WriteByte('=')
		builder.WriteString(labels[key])
		builder.WriteByte(0)
	}
	return builder.String()
}

// finishO11YExportError still writes the archive so rejected data remains
// inspectable and replayable.
func finishO11YExportError(options *O11YExportOptions, archive *O11YArchive, exportErr error) (*CaseBundle, *O11YArchive, error) {
	if err := writeO11YArchive(options.ArchivePath, archive); err != nil {
		return nil, archive, fmt.Errorf("evaluation: O11Y export failed (%v) and archive write failed: %w", exportErr, err)
	}
	return nil, archive, fmt.Errorf("evaluation: O11Y export: %w", exportErr)
}

func writePrivateJSON(path string, value any) error {
	data, err := json.MarshalIndent(value, "", "  ")
	if err != nil {
		return err
	}
	return writePrivateFile(path, append(data, '\n'))
}

func writePrivateFile(path string, data []byte) error {
	if err := os.WriteFile(path, data, 0o600); err != nil {
		return err
	}
	return os.Chmod(path, 0o600)
}

func makePrivateDir(path string) error {
	if err := os.MkdirAll(path, 0o700); err != nil {
		return err
	}
	return os.Chmod(path, 0o700)
}

func isSafeQueryName(name string) bool {
	if name == "" {
		return false
	}
	for _, character := range name {
		if character >= 'a' && character <= 'z' || character >= 'A' && character <= 'Z' || character >= '0' && character <= '9' || character == '_' || character == '-' || character == '.' {
			continue
		}
		return false
	}
	return name != "." && name != ".."
}

func queryCount(sources []O11YSourceSpec) int {
	count := 0
	for i := range sources {
		count += len(sources[i].Queries)
	}
	return count
}