go-otel/internal/collector/handlers.go

// Package collector implements an OpenTelemetry consumer that stores telemetry in history
// and broadcasts it to connected clients.
package collector

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"git.tornberg.me/go-otel/internal/sourcemaps"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

	"git.tornberg.me/go-otel/internal/features"
	"git.tornberg.me/go-otel/internal/issues"
	"git.tornberg.me/go-otel/internal/storage"
	"git.tornberg.me/go-otel/internal/websocket"
	"go.opentelemetry.io/collector/pdata/plog"
	"go.opentelemetry.io/collector/pdata/pmetric"
	"go.opentelemetry.io/collector/pdata/ptrace"
)

// checkIngest applies the shared ingest rate limit and, if configured, the
// ingest key check. It writes an error response and returns false when the
// request should be rejected.
func (c *Collector) checkIngest(w http.ResponseWriter, r *http.Request) bool {
	if !c.ingestLimiter.Allow(clientIP(r)) {
		http.Error(w, "too many requests, slow down", http.StatusTooManyRequests)
		return false
	}
	if c.ingestKey == "" {
		return true
	}
	key := r.Header.Get("X-Otel-Ingest-Key")
	if key == "" {
		key = r.URL.Query().Get("key")
	}
	if key != c.ingestKey {
		http.Error(w, "invalid or missing ingest key", http.StatusUnauthorized)
		return false
	}
	return true
}

// HandleTraces handles OTLP trace requests.
func (c *Collector) HandleTraces(w http.ResponseWriter, r *http.Request) {
	log.Printf("Received traces request: method=%s, content-type=%s", r.Method, r.Header.Get("Content-Type"))
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.checkIngest(w, r) {
		return
	}

	body, err := io.ReadAll(r.Body)
	if err != nil {
		log.Printf("failed to read request body: %v", err)
		http.Error(w, "failed to read request body", http.StatusInternalServerError)
		return
	}

	log.Printf("Received traces body, size=%d", len(body))

	var unmarshaler ptrace.Unmarshaler
	contentType := r.Header.Get("Content-Type")
	switch contentType {
	case "application/x-protobuf":
		unmarshaler = &ptrace.ProtoUnmarshaler{}
	case "application/json":
		unmarshaler = &ptrace.JSONUnmarshaler{}
	default:
		unmarshaler = &ptrace.JSONUnmarshaler{} // default to JSON
	}

	traces, err := unmarshaler.UnmarshalTraces(body)
	if err != nil {
		log.Printf("failed to unmarshal traces: %v", err)
		http.Error(w, "failed to unmarshal traces", http.StatusBadRequest)
		return
	}

	log.Printf("Successfully unmarshaled traces, span_count=%d", traces.SpanCount())

	if err := c.ConsumeTraces(context.Background(), traces); err != nil {
		log.Printf("failed to consume traces: %v", err)
		http.Error(w, "failed to consume traces", http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusOK)
}

// HandleMetrics handles OTLP metric requests.
func (c *Collector) HandleMetrics(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.checkIngest(w, r) {
		return
	}

	body, err := io.ReadAll(r.Body)
	if err != nil {
		log.Printf("failed to read request body: %v", err)
		http.Error(w, "failed to read request body", http.StatusInternalServerError)
		return
	}

	var unmarshaler pmetric.Unmarshaler
	contentType := r.Header.Get("Content-Type")
	switch contentType {
	case "application/x-protobuf":
		unmarshaler = &pmetric.ProtoUnmarshaler{}
	case "application/json":
		unmarshaler = &pmetric.JSONUnmarshaler{}
	default:
		unmarshaler = &pmetric.JSONUnmarshaler{} // default to JSON
	}

	metrics, err := unmarshaler.UnmarshalMetrics(body)
	if err != nil {
		log.Printf("failed to unmarshal metrics: %v", err)
		http.Error(w, "failed to unmarshal metrics", http.StatusBadRequest)
		return
	}

	if err := c.ConsumeMetrics(context.Background(), metrics); err != nil {
		log.Printf("failed to consume metrics: %v", err)
		http.Error(w, "failed to consume metrics", http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusOK)
}

// HandleLogs handles OTLP log requests.
func (c *Collector) HandleLogs(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.checkIngest(w, r) {
		return
	}

	body, err := io.ReadAll(r.Body)
	if err != nil {
		log.Printf("failed to read request body: %v", err)
		http.Error(w, "failed to read request body", http.StatusInternalServerError)
		return
	}

	var unmarshaler plog.Unmarshaler
	contentType := r.Header.Get("Content-Type")
	switch contentType {
	case "application/x-protobuf":
		unmarshaler = &plog.ProtoUnmarshaler{}
	case "application/json":
		unmarshaler = &plog.JSONUnmarshaler{}
	default:
		unmarshaler = &plog.JSONUnmarshaler{} // default to JSON
	}

	logs, err := unmarshaler.UnmarshalLogs(body)
	if err != nil {
		log.Printf("failed to unmarshal logs: %v", err)
		http.Error(w, "failed to unmarshal logs", http.StatusBadRequest)
		return
	}

	if err := c.ConsumeLogs(context.Background(), logs); err != nil {
		log.Printf("failed to consume logs: %v", err)
		http.Error(w, "failed to consume logs", http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusOK)
}

// parseSearchOptions builds a storage.SearchOptions from common query
// parameters shared by the search/wait/history endpoints: q, type, since, until.
func parseSearchOptions(r *http.Request) storage.SearchOptions {
	q := r.URL.Query()
	opts := storage.SearchOptions{
		Query: q.Get("q"),
		Type:  q.Get("type"),
	}
	if since := q.Get("since"); since != "" {
		if v, err := strconv.ParseInt(since, 10, 64); err == nil {
			opts.Since = v
		}
	}
	if until := q.Get("until"); until != "" {
		if v, err := strconv.ParseInt(until, 10, 64); err == nil {
			opts.Until = v
		}
	}
	return opts
}

func parseLimit(r *http.Request, def int) int {
	if limitStr := r.URL.Query().Get("limit"); limitStr != "" {
		if l, err := strconv.Atoi(limitStr); err == nil {
			return l
		}
	}
	return def
}

// HandleHistory returns the telemetry history as JSON. It searches the full
// persistent store when one is configured, so results are not bounded by the
// in-memory ring buffer size.
func (c *Collector) HandleHistory(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	history, err := c.SearchHistory(r.Context(), storage.SearchOptions{}, parseLimit(r, 100))
	if err != nil {
		log.Printf("failed to fetch history: %v", err)
		http.Error(w, "failed to fetch history", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(history); err != nil {
		log.Printf("failed to encode history: %v", err)
		http.Error(w, "failed to encode history", http.StatusInternalServerError)
		return
	}
}

// HandleSearch searches the telemetry history. Supports the free-text 'q'
// parameter plus structured filters: type, since, until (unix nanoseconds).
func (c *Collector) HandleSearch(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	opts := parseSearchOptions(r)
	history, err := c.SearchHistory(r.Context(), opts, parseLimit(r, 100))
	if err != nil {
		log.Printf("failed to search history: %v", err)
		http.Error(w, "failed to search history", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(history); err != nil {
		log.Printf("failed to encode search results: %v", err)
		http.Error(w, "failed to encode history", http.StatusInternalServerError)
		return
	}
}

// HandleWait waits for a specific telemetry signal, optionally restricted by
// the same structured filters as HandleSearch (type, since, until).
func (c *Collector) HandleWait(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	if !c.waitLimiter.Allow(clientIP(r)) {
		http.Error(w, "too many wait requests, slow down", http.StatusTooManyRequests)
		return
	}

	opts := parseSearchOptions(r)
	timeoutSec := 30
	if t := r.URL.Query().Get("timeout"); t != "" {
		if ts, err := strconv.Atoi(t); err == nil {
			timeoutSec = ts
		}
	}

	ctx, cancel := context.WithTimeout(r.Context(), time.Duration(timeoutSec)*time.Second)
	defer cancel()

	// First, check if result already exists in recent history
	matches, err := c.SearchHistory(ctx, opts, 1)
	if err == nil && len(matches) > 0 {
		w.Header().Set("Content-Type", "application/json")
		if err := json.NewEncoder(w).Encode(matches[0]); err != nil {
			log.Printf("failed to encode match: %v", err)
		}
		return
	}

	// Otherwise, subscribe and wait
	lowerQuery := strings.ToLower(opts.Query)
	ch, cleanup := c.Subscribe(func(entry websocket.HistoryEntry) bool {
		if opts.Type != "" && entry.Type != opts.Type {
			return false
		}
		if opts.Query == "" {
			return true
		}
		return strings.Contains(strings.ToLower(string(entry.Data)), lowerQuery)
	})
	defer cleanup()

	select {
	case entry := <-ch:
		w.Header().Set("Content-Type", "application/json")
		if err := json.NewEncoder(w).Encode(entry); err != nil {
			log.Printf("failed to encode signal: %v", err)
		}
	case <-ctx.Done():
		http.Error(w, "timeout waiting for signal", http.StatusGatewayTimeout)
	}
}

// HandleHistoryByType returns telemetry history filtered by type, searching
// the full persistent store when one is configured.
func (c *Collector) HandleHistoryByType(dataType string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodGet {
			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
			return
		}

		opts := storage.SearchOptions{Type: dataType}
		history, err := c.SearchHistory(r.Context(), opts, parseLimit(r, 100))
		if err != nil {
			log.Printf("failed to fetch history for type %s: %v", dataType, err)
			http.Error(w, "failed to fetch history", http.StatusInternalServerError)
			return
		}

		w.Header().Set("Content-Type", "application/json")
		if err := json.NewEncoder(w).Encode(history); err != nil {
			log.Printf("failed to encode filtered history: %v", err)
			http.Error(w, "failed to encode history", http.StatusInternalServerError)
			return
		}
	}
}

// HandleTraceByID returns all stored entries referencing the given trace id,
// so agents don't have to rely on the trace id happening to substring-match
// inside a search result.
func (c *Collector) HandleTraceByID(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	traceID := r.PathValue("traceId")
	if traceID == "" {
		http.Error(w, "traceId is required", http.StatusBadRequest)
		return
	}

	var entries []websocket.HistoryEntry
	var err error
	if c.traceQuerier != nil {
		// Indexed lookup, payloads trimmed to this trace's spans/logs.
		entries, err = c.traceQuerier.TraceEntries(r.Context(), traceID, parseLimit(r, 1000))
	} else {
		entries, err = c.SearchHistory(r.Context(), storage.SearchOptions{Query: traceID}, parseLimit(r, 1000))
	}
	if err != nil {
		log.Printf("failed to fetch trace %s: %v", traceID, err)
		http.Error(w, "failed to fetch trace", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(entries); err != nil {
		log.Printf("failed to encode trace: %v", err)
		http.Error(w, "failed to encode trace", http.StatusInternalServerError)
		return
	}
}

// HandleIssues returns deduplicated error issues derived from ingested
// traces and logs, most recently seen first.
func (c *Collector) HandleIssues(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	opts, err := parseIssueListOptions(r)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	page := c.ListIssuesFiltered(opts)

	w.Header().Set("Content-Type", "application/json")
	if page.HasMore && len(page.Issues) > 0 {
		last := page.Issues[len(page.Issues)-1]
		next := encodeIssueCursor(last.LastSeen, last.ID)
		w.Header().Set("X-Next-Cursor", next)
		nextQuery := r.URL.Query()
		nextQuery.Set("cursor", next)
		w.Header().Set("Link", fmt.Sprintf("<%s?%s>; rel=next", r.URL.Path, nextQuery.Encode()))
	}
	// Keep the response as a JSON array for compatibility. Pagination metadata
	// is exposed in headers so existing UI clients do not need a migration.
	if err := json.NewEncoder(w).Encode(page.Issues); err != nil {
		log.Printf("failed to encode issues: %v", err)
		http.Error(w, "failed to encode issues", http.StatusInternalServerError)
		return
	}
}

const maxIssuePageSize = 200

type issueCursor struct {
	LastSeen int64  `json:"lastSeen"`
	ID       string `json:"id"`
}

func parseIssueListOptions(r *http.Request) (issues.ListOptions, error) {
	q := r.URL.Query()
	opts := issues.ListOptions{
		Service:  q.Get("service"),
		Kind:     q.Get("kind"),
		Status:   q.Get("status"),
		Title:    q.Get("q"),
		SourceIP: q.Get("sourceIp"),
		Route:    q.Get("route"),
		Limit:    parseLimit(r, 100),
	}
	if opts.Limit <= 0 {
		opts.Limit = 100
	}
	if opts.Limit > maxIssuePageSize {
		opts.Limit = maxIssuePageSize
	}
	for name, dst := range map[string]*int64{"since": &opts.Since, "until": &opts.Until} {
		if value := q.Get(name); value != "" {
			parsed, err := strconv.ParseInt(value, 10, 64)
			if err != nil || parsed < 0 {
				return issues.ListOptions{}, fmt.Errorf("invalid %s; expected unix nanoseconds", name)
			}
			*dst = parsed
		}
	}
	if opts.Since != 0 && opts.Until != 0 && opts.Since > opts.Until {
		return issues.ListOptions{}, fmt.Errorf("since must be less than or equal to until")
	}
	if value := q.Get("cursor"); value != "" {
		decoded, err := base64.RawURLEncoding.DecodeString(value)
		if err != nil {
			return issues.ListOptions{}, fmt.Errorf("invalid cursor")
		}
		var cursor issueCursor
		if err := json.Unmarshal(decoded, &cursor); err != nil || cursor.LastSeen <= 0 || cursor.ID == "" {
			return issues.ListOptions{}, fmt.Errorf("invalid cursor")
		}
		opts.AfterLastSeen = cursor.LastSeen
		opts.AfterID = cursor.ID
	}
	return opts, nil
}

func encodeIssueCursor(lastSeen int64, id string) string {
	data, _ := json.Marshal(issueCursor{LastSeen: lastSeen, ID: id})
	return base64.RawURLEncoding.EncodeToString(data)
}

// HandleSourceMap accepts a JavaScript source map for a deployed release.
// Multipart fields: release, source, file.
func (c *Collector) HandleSourceMap(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if c.sourceMaps == nil {
		http.Error(w, "source maps are not configured", http.StatusNotImplemented)
		return
	}
	if err := r.ParseMultipartForm(50 << 20); err != nil {
		http.Error(w, "invalid multipart body", http.StatusBadRequest)
		return
	}
	release, source := r.FormValue("release"), r.FormValue("source")
	file, _, err := r.FormFile("file")
	if err != nil {
		http.Error(w, "file is required", http.StatusBadRequest)
		return
	}
	defer file.Close()
	if err := c.sourceMaps.Upload(release, source, file); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	w.WriteHeader(http.StatusCreated)
}

// HandleListSourceMaps lists the release/source pairs currently loaded, so a
// deploy can verify its uploads (and that they survived a restart).
func (c *Collector) HandleListSourceMaps(w http.ResponseWriter, r *http.Request) {
	list := []sourcemaps.Entry{}
	if c.sourceMaps != nil {
		list = c.sourceMaps.List()
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(list)
}

// HandleResolveStacks maps raw browser stacktraces back to source using the
// uploaded source maps. Body: {"items":[{"release":"...","stacktrace":"..."}]};
// the response has one resolved stacktrace per item, in order.
func (c *Collector) HandleResolveStacks(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Items []struct {
			Release    string `json:"release"`
			Stacktrace string `json:"stacktrace"`
		} `json:"items"`
	}
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&req); err != nil {
		http.Error(w, "invalid body", http.StatusBadRequest)
		return
	}
	out := make([]string, len(req.Items))
	frames := make([][]sourcemaps.Frame, len(req.Items))
	for i, it := range req.Items {
		out[i] = it.Stacktrace
		frames[i] = []sourcemaps.Frame{}
		if c.sourceMaps != nil {
			out[i] = c.sourceMaps.ResolveStack(it.Release, it.Stacktrace)
			frames[i] = c.sourceMaps.ResolveFrames(it.Release, it.Stacktrace)
		}
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{"resolved": out, "frames": frames})
}

// HandleIssueByID returns a single issue by id, including its samples.
func (c *Collector) HandleIssueByID(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	switch r.Method {
	case http.MethodGet:
		issue, ok := c.GetIssue(id)
		if !ok {
			http.Error(w, "issue not found", http.StatusNotFound)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		if err := json.NewEncoder(w).Encode(issue); err != nil {
			log.Printf("failed to encode issue: %v", err)
			http.Error(w, "failed to encode issue", http.StatusInternalServerError)
		}
	case http.MethodPatch:
		c.HandleIssueStatus(w, r, id)
	default:
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
	}
}

// HandlePatterns returns deterministic related-issue groups for visualization.
func (c *Collector) HandlePatterns(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(c.ListPatterns())
}

// HandleRollouts returns normalized Kubernetes Deployment rollout history.
func (c *Collector) HandleRollouts(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.KubernetesEnabled() {
		jsonError(w, http.StatusNotImplemented, c.KubernetesError())
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(c.ListRollouts())
}

func (c *Collector) HandleRolloutByID(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.KubernetesEnabled() {
		jsonError(w, http.StatusNotImplemented, c.KubernetesError())
		return
	}
	rollout, ok := c.GetRollout(r.PathValue("id"))
	if !ok {
		http.Error(w, "rollout not found", http.StatusNotFound)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(rollout)
}

func (c *Collector) HandleRolloutImpact(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.KubernetesEnabled() {
		jsonError(w, http.StatusNotImplemented, c.KubernetesError())
		return
	}
	rollout, ok := c.GetRollout(r.PathValue("id"))
	if !ok {
		http.Error(w, "rollout not found", http.StatusNotFound)
		return
	}
	response := map[string]interface{}{"configured": c.prometheus != nil, "rollout": rollout, "points": []interface{}{}}
	if c.prometheus != nil {
		start := time.Unix(0, rollout.StartedAt).Add(-15 * time.Minute)
		end := time.Now()
		if start.Before(end.Add(-24 * time.Hour)) {
			start = end.Add(-24 * time.Hour)
		}
		points, err := c.prometheus.ErrorRate(r.Context(), rollout.Service, rollout.Release, start, end)
		if err != nil {
			jsonError(w, http.StatusBadGateway, err.Error())
			return
		}
		response["points"] = points
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(response)
}

func jsonError(w http.ResponseWriter, status int, message string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(map[string]string{"error": message})
}

// HandleIssueSimilar returns explainable, non-merging related-issue suggestions.
func (c *Collector) HandleIssueSimilar(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	limit := parseLimit(r, 10)
	result, ok := c.SimilarIssues(r.PathValue("id"), limit)
	if !ok {
		http.Error(w, "issue not found", http.StatusNotFound)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(result)
}

// HandleSimilarityFeedback records a human review label for a pair of issues.
// Body: {"issueId":"...", "decision":"same|different"}.
func (c *Collector) HandleSimilarityFeedback(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	var req struct {
		IssueID  string `json:"issueId"`
		Decision string `json:"decision"`
	}
	if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&req); err != nil {
		http.Error(w, "invalid JSON body", http.StatusBadRequest)
		return
	}
	if !c.SetSimilarityFeedback(r.PathValue("id"), req.IssueID, req.Decision) {
		http.Error(w, "invalid issue pair or decision", http.StatusBadRequest)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]string{"status": "recorded", "decision": req.Decision})
}

// HandleIssueStatus updates the triage state (open/resolved/ignored) of an
// issue. Body: {"status": "resolved"}.
func (c *Collector) HandleIssueStatus(w http.ResponseWriter, r *http.Request, id string) {
	var req struct {
		Status string `json:"status"`
	}
	body, err := io.ReadAll(io.LimitReader(r.Body, 4096))
	if err != nil {
		http.Error(w, "failed to read request body", http.StatusBadRequest)
		return
	}
	if err := json.Unmarshal(body, &req); err != nil {
		http.Error(w, "invalid JSON body", http.StatusBadRequest)
		return
	}

	issue, ok := c.SetIssueStatus(id, req.Status)
	if !ok {
		http.Error(w, "issue not found or invalid status", http.StatusBadRequest)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(issue); err != nil {
		log.Printf("failed to encode issue: %v", err)
	}
}

// HandleHealth returns the health status of the collector.
func (c *Collector) HandleHealth(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	if err := json.NewEncoder(w).Encode(map[string]string{"status": "ok"}); err != nil {
		log.Printf("failed to encode health check: %v", err)
	}
}

// HandleMLDetectorTest runs a deterministic, in-memory spike through the
// production feature and label builder. It stores no telemetry and exists so
// an operator can verify the detector wiring from the authenticated UI.
func (c *Collector) HandleMLDetectorTest(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	const service = "ml-verification"
	base := time.Date(2026, 1, 8, 12, 0, 0, 0, time.UTC)
	events := make([]features.Event, 0, 100)
	for day := 1; day <= 7; day++ {
		for i := 0; i < 10; i++ {
			events = append(events, features.Event{Timestamp: base.AddDate(0, 0, -day).Add(time.Duration(i) * time.Second), Service: service, Kind: "span", IsError: i == 0, DurationMS: float64(i + 1)})
		}
	}
	for i := 0; i < 10; i++ {
		events = append(events, features.Event{Timestamp: base.Add(time.Duration(i) * time.Second), Service: service, Release: "verification", Kind: "span", DurationMS: float64(i + 1)})
		events = append(events, features.Event{Timestamp: base.Add(5 * time.Minute).Add(time.Duration(i) * time.Second), Service: service, Release: "verification", Kind: "log", IsError: i < 5, IssueFingerprint: "verification-spike"})
	}
	rows := features.Build(events, features.Config{ObservedUntil: base.Add(20 * time.Minute)})
	for _, row := range rows {
		if row.WindowStart.Equal(base) && row.Service == service && row.Release == "verification" {
			w.Header().Set("Content-Type", "application/json")
			_ = json.NewEncoder(w).Encode(map[string]interface{}{
				"passed":   row.LabelKnown && row.Label,
				"scenario": "five errors in ten future events against a seven-day 10% baseline",
				"row":      row,
			})
			return
		}
	}
	jsonError(w, http.StatusInternalServerError, "detector test did not generate its verification row")
}

// statsEnabled reports whether an aggregate-capable store is configured.
func (c *Collector) statsEnabled() bool { return c.statsQuerier != nil }

// statsNotEnabled writes the response used when the file store is active:
// a stable 501 the UI treats as "view not available".
func statsNotEnabled(w http.ResponseWriter) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusNotImplemented)
	_ = json.NewEncoder(w).Encode(map[string]string{
		"error": "stats require OTEL_CLICKHOUSE_ADDR (ClickHouse storage) to be configured",
	})
}

// parseStatsCommon reads the shared service/hours params for stats endpoints.
func parseStatsCommon(r *http.Request) (service string, hours int) {
	service = r.URL.Query().Get("service")
	hours = 72
	if h, err := strconv.Atoi(r.URL.Query().Get("hours")); err == nil && h > 0 && h <= 24*90 {
		hours = h
	}
	return service, hours
}

// HandleStatsErrors returns hourly/daily total and error counts for the
// trends graphs. Only available with ClickHouse storage.
func (c *Collector) HandleStatsErrors(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.statsEnabled() {
		statsNotEnabled(w)
		return
	}

	service, hours := parseStatsCommon(r)
	daily := hours > 24*14 // beyond two weeks, daily buckets read better
	points, err := c.statsQuerier.ErrorStats(r.Context(), service, hours, daily)
	if err != nil {
		log.Printf("failed to compute error stats: %v", err)
		http.Error(w, "failed to compute error stats", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]interface{}{
		"granularity": map[bool]string{true: "day", false: "hour"}[daily],
		"points":      points,
	})
}

// HandleStatsReleases returns per-release error-count time series so two
// deploys can be compared line-by-line. Only available with ClickHouse.
func (c *Collector) HandleStatsReleases(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.statsEnabled() {
		statsNotEnabled(w)
		return
	}

	service, hours := parseStatsCommon(r)
	series, err := c.statsQuerier.ReleaseSeries(r.Context(), service, hours)
	if err != nil {
		log.Printf("failed to compute release stats: %v", err)
		http.Error(w, "failed to compute release stats", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]interface{}{
		"series": series,
	})
}

// HandleStatsServices lists the distinct services present in the rollup.
func (c *Collector) HandleStatsServices(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !c.statsEnabled() {
		statsNotEnabled(w)
		return
	}

	services, err := c.statsQuerier.Services(r.Context())
	if err != nil {
		log.Printf("failed to list services: %v", err)
		http.Error(w, "failed to list services", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]interface{}{"services": services})
}

// needleAddr returns the configured or default address for the Needle 3 sidecar.
func needleAddr() string {
	addr := os.Getenv("OTEL_NEEDLE_ADDR")
	if addr == "" {
		addr = "http://127.0.0.1:5005"
	}
	return strings.TrimRight(addr, "/")
}

// HandleNeedleStatus checks if the Needle 3 sidecar is reachable and reports its health.
func (c *Collector) HandleNeedleStatus(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	target := needleAddr() + "/health"
	client := &http.Client{Timeout: 2 * time.Second}
	resp, err := client.Get(target)
	w.Header().Set("Content-Type", "application/json")
	if err != nil {
		w.WriteHeader(http.StatusServiceUnavailable)
		_ = json.NewEncoder(w).Encode(map[string]interface{}{
			"available": false,
			"error":     fmt.Sprintf("Needle sidecar unreachable at %s: %v", target, err),
		})
		return
	}
	defer resp.Body.Close()
	w.WriteHeader(resp.StatusCode)
	_, _ = io.Copy(w, resp.Body)
}

// HandleNeedleClassify proxies telemetry triage requests to the Needle 3 sidecar.
func (c *Collector) HandleNeedleClassify(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "failed to read request body", http.StatusBadRequest)
		return
	}

	// If the payload specifies an issue_id and no text, look up the issue
	var reqMap map[string]interface{}
	if err := json.Unmarshal(body, &reqMap); err == nil {
		if issueID, ok := reqMap["issue_id"].(string); ok && issueID != "" {
			if issue, ok := c.GetIssue(issueID); ok {
				var promptParts []string
				promptParts = append(promptParts, fmt.Sprintf("Service: %s", issue.Service))
				promptParts = append(promptParts, fmt.Sprintf("Title: %s", issue.Title))
				promptParts = append(promptParts, fmt.Sprintf("Kind: %s", issue.Kind))
				if len(issue.Samples) > 0 {
					latest := issue.Samples[len(issue.Samples)-1]
					if latest.Message != "" {
						promptParts = append(promptParts, fmt.Sprintf("Message: %s", latest.Message))
					}
					if latest.URL != "" {
						promptParts = append(promptParts, fmt.Sprintf("URL: %s", latest.URL))
					}
					if latest.HTTPStatus > 0 {
						promptParts = append(promptParts, fmt.Sprintf("Status: %d", latest.HTTPStatus))
					}
					if latest.HTTPRoute != "" {
						promptParts = append(promptParts, fmt.Sprintf("Route: %s", latest.HTTPRoute))
					}
				}
				reqMap["text"] = strings.Join(promptParts, " | ")
				reqMap["service"] = issue.Service
				reqMap["title"] = issue.Title
				body, _ = json.Marshal(reqMap)
			}
		}
	}

	target := needleAddr() + "/classify"
	client := &http.Client{Timeout: 60 * time.Second}
	proxyReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost, target, strings.NewReader(string(body)))
	if err != nil {
		http.Error(w, "failed to create proxy request", http.StatusInternalServerError)
		return
	}
	proxyReq.Header.Set("Content-Type", "application/json")

	resp, err := client.Do(proxyReq)
	if err != nil {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusServiceUnavailable)
		_ = json.NewEncoder(w).Encode(map[string]interface{}{
			"success": false,
			"error":   fmt.Sprintf("Needle sidecar unreachable at %s: %v", target, err),
		})
		return
	}
	defer resp.Body.Close()

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(resp.StatusCode)
	_, _ = io.Copy(w, resp.Body)
}