fathom/internal/web/server.go

// Package web serves Fathom's API and its embedded UI on one port.
package web

import (
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"

	"git.z.gd/zzg/fathom/internal/ingest"
	"git.z.gd/zzg/fathom/internal/insights"
	"git.z.gd/zzg/fathom/internal/query"
	"git.z.gd/zzg/fathom/internal/store"
	"git.z.gd/zzg/fathom/internal/tracetree"
)

// Server wires the store, decoder, and query engine to HTTP.
type Server struct {
	store   *store.Store
	decoder *ingest.Decoder
	engine  *query.Engine
	corr    *tracetree.Correlator
	hub     *Hub
	logger  *log.Logger

	// insights is optional. Nothing else in the server depends on it, so a nil
	// analyzer removes the tab and changes nothing else.
	insights *insights.Analyzer
}

// SetInsights attaches a background analyzer.
func (s *Server) SetInsights(a *insights.Analyzer) { s.insights = a }

// NewServer wires the pieces. A nil hub is allowed and simply means no live
// updates, which keeps tests that do not care about streaming simple.
func NewServer(s *store.Store, d *ingest.Decoder, hub *Hub, logger *log.Logger) *Server {
	if hub == nil {
		hub = NewHub()
	}
	return &Server{
		store:   s,
		decoder: d,
		engine:  query.NewEngine(s),
		corr:    tracetree.NewCorrelator(),
		hub:     hub,
		logger:  logger,
	}
}

// Handler builds the mux. OTLP ingest, the read API, and the UI all live here
// on one port: a second listener is a second thing to configure and firewall,
// for no benefit at this scale.
func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()

	mux.HandleFunc("POST /v1/traces", s.handleTraces)
	mux.HandleFunc("POST /v1/logs", s.handleLogs)
	// Metrics are accepted and discarded. Exporters send all three signals to
	// one endpoint, and 404ing here fills the agent's own logs with errors
	// about a signal Fathom deliberately does not store.
	mux.HandleFunc("POST /v1/metrics", s.handleMetrics)

	mux.HandleFunc("GET /api/traces", s.handleTraceList)
	mux.HandleFunc("GET /api/traces/{id}", s.handleTraceDetail)
	mux.HandleFunc("POST /api/query", s.handleQuery)
	mux.HandleFunc("GET /api/stats", s.handleStats)
	mux.HandleFunc("GET /api/events", s.hub.handleEvents)

	mux.HandleFunc("GET /api/insights", s.handleInsights)
	mux.HandleFunc("POST /api/insights/dismiss", s.handleDismissInsight)
	mux.HandleFunc("POST /api/insights/restore", s.handleRestoreInsights)
	mux.HandleFunc("POST /api/insights/local", s.handleLocalToggle)
	mux.HandleFunc("POST /api/insights/surface", s.handleSurface)
	mux.HandleFunc("POST /api/insights/surface/reset", s.handleResetSurfaced)

	mux.Handle("GET /", uiHandler())
	return logRequests(s.logger, mux)
}

func logRequests(l *log.Logger, next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()
		rec := &statusRecorder{ResponseWriter: w, code: 200}
		next.ServeHTTP(rec, r)
		// Successful ingest is high-frequency and uninteresting; logging only
		// failures keeps the console readable while an agent is running.
		// The event stream is long-lived; logging it on close would report a
		// multi-hour "request" and say nothing useful.
		if r.URL.Path == "/api/events" {
			return
		}
		if rec.code >= 400 || !strings.HasPrefix(r.URL.Path, "/v1/") {
			l.Printf("%s %s %d %s", r.Method, r.URL.Path, rec.code,
				time.Since(start).Round(time.Millisecond))
		}
	})
}

type statusRecorder struct {
	http.ResponseWriter
	code int
}

func (r *statusRecorder) WriteHeader(c int) {
	r.code = c
	r.ResponseWriter.WriteHeader(c)
}

// Flush forwards to the wrapped writer.
//
// Embedding http.ResponseWriter satisfies that interface but hides every other
// one the underlying writer implements, so without this the event stream's
// `w.(http.Flusher)` check fails and SSE silently degrades to an error. This is
// the standard hazard of wrapping a ResponseWriter, and it only shows up on the
// streaming endpoint.
func (r *statusRecorder) Flush() {
	if f, ok := r.ResponseWriter.(http.Flusher); ok {
		f.Flush()
	}
}

const maxIngestBytes = 64 << 20

// --- ingest ---

func (s *Server) handleTraces(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxIngestBytes))
	if err != nil {
		otlpError(w, http.StatusBadRequest, err)
		return
	}
	spans, err := s.decoder.DecodeTraces(body)
	if err != nil {
		otlpError(w, http.StatusBadRequest, err)
		return
	}
	if err := s.store.AddSpans(spans); err != nil {
		otlpError(w, http.StatusInternalServerError, err)
		return
	}
	otlpOK(w)
}

func (s *Server) handleLogs(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxIngestBytes))
	if err != nil {
		otlpError(w, http.StatusBadRequest, err)
		return
	}
	recs, err := s.decoder.DecodeLogs(body)
	if err != nil {
		otlpError(w, http.StatusBadRequest, err)
		return
	}
	if err := s.store.AddLogs(recs); err != nil {
		otlpError(w, http.StatusInternalServerError, err)
		return
	}
	otlpOK(w)
}

func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) {
	io.Copy(io.Discard, http.MaxBytesReader(w, r.Body, maxIngestBytes))
	otlpOK(w)
}

// otlpOK returns the success shape OTLP exporters expect; an empty
// partialSuccess means everything was accepted.
func otlpOK(w http.ResponseWriter) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	w.Write([]byte(`{"partialSuccess":{}}`))
}

func otlpError(w http.ResponseWriter, code int, err error) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	json.NewEncoder(w).Encode(map[string]any{"code": code, "message": err.Error()})
}

// --- query ---

func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Query string `json:"query"`
	}
	if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil {
		httpError(w, http.StatusBadRequest, "reading request: %v", err)
		return
	}
	q, err := query.Parse(req.Query)
	if err != nil {
		// A parse error is the user's to fix, so it goes back verbatim rather
		// than as a generic 400.
		httpError(w, http.StatusBadRequest, "%v", err)
		return
	}
	start := time.Now()
	res, err := s.engine.Run(q)
	if err != nil {
		httpError(w, http.StatusBadRequest, "%v", err)
		return
	}
	res.Stats.ElapsedMicros = time.Since(start).Microseconds()
	writeJSON(w, res)
}

func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) {
	type shardStat struct {
		Table string `json:"table"`
		Date  string `json:"date"`
		Bytes int64  `json:"bytes"`
	}
	var out []shardStat
	var total int64
	for _, table := range []string{"spans", "logs"} {
		shards, err := s.store.ListShards(table)
		if err != nil {
			continue
		}
		for _, sh := range shards {
			out = append(out, shardStat{sh.Table, sh.Date, sh.Bytes})
			total += sh.Bytes
		}
	}
	if out == nil {
		out = []shardStat{}
	}
	writeJSON(w, map[string]any{"shards": out, "total_bytes": total})
}

// --- helpers ---

func writeJSON(w http.ResponseWriter, v any) {
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(v)
}

func httpError(w http.ResponseWriter, code int, format string, args ...any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	json.NewEncoder(w).Encode(map[string]string{"error": fmt.Sprintf(format, args...)})
}

func intParam(r *http.Request, name string, def int) int {
	if v := r.URL.Query().Get(name); v != "" {
		if n, err := strconv.Atoi(v); err == nil {
			return n
		}
	}
	return def
}

func int64Param(r *http.Request, name string, def int64) int64 {
	if v := r.URL.Query().Get(name); v != "" {
		if n, err := strconv.ParseInt(v, 10, 64); err == nil {
			return n
		}
	}
	return def
}

func nowMicros() int64 { return time.Now().UnixMicro() }

// timeRange resolves from/to, defaulting to the last 24 hours.
func timeRange(r *http.Request) (int64, int64) {
	now := time.Now().UnixMicro()
	from := int64Param(r, "from", now-int64(24*time.Hour/time.Microsecond))
	to := int64Param(r, "to", now)
	if from > to {
		from, to = to, from
	}
	return from, to
}