package main

import (
	"context"
	"crypto/rand"
	"encoding/binary"
	"fmt"
	"hash/fnv"
	"html/template"
	"log"
	"net/http"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/starfederation/datastar-go/datastar"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

// ---- public trace view: sanitized, realtime span feed --------------------
//
// This is the read-only educational/transparency surface discussed in
// slop-essays/observability.md, served at GET /traces and morphed live like
// every other route. It is PUBLIC and on by default (-trace-view=false opts
// out): the safety guarantee is not the gate but the sanitizer — every span
// passes a deny-by-default allow-list before it can reach a browser. The raw
// otelhttp server spans carry the client IP (network.peer.address /
// client.address from X-Forwarded-For), the User-Agent, and the real issue id
// in url.path — none of which belong on a public feed. Only low-cardinality,
// content-free SHAPE attributes survive (see traceAttrAllow).

// traceAttrAllow is the allow-list of span attributes that may be shown
// publicly. Every key here is low-cardinality and content-free by construction
// (query shapes, row counts, the access strategy, push accounting) — never user
// text, issue ids, IPs, or User-Agents. The list is exhaustive and deny-by-
// default: an attribute not named here (client.address, network.peer.*,
// user_agent.original, url.path, url.query, exception.*, server.address, …) is
// dropped, so newly-added instrumentation stays invisible until someone opts it
// in here on purpose.
var traceAttrAllow = map[attribute.Key]bool{
	"db.strategy":               true,
	"db.rows":                   true,
	"query.has_text":            true,
	"query.has_author":          true,
	"query.labels":              true,
	"query.status":              true,
	"query.sort":                true,
	"query.limit":               true,
	"push.sent":                 true,
	"push.raw_bytes":            true,
	"http.response.status_code": true,
	"inval.scoped":              true, // batch invalidated by predicate vs global bump
}

// traceAttr is one sanitized attribute (already stringified for display).
type traceAttr struct{ k, v string }

// traceSpan is a sanitized, display-ready snapshot of one finished span. It
// holds only what survives traceAttrAllow plus structural fields needed to
// render the tree (span/parent ids resolve depth; start orders rows). No raw
// OTel span is retained, so there is nothing sensitive left to leak downstream.
type traceSpan struct {
	traceID  string
	spanID   string
	parentID string
	name     string
	start    time.Time
	dur      time.Duration
	isError  bool
	attrs    []traceAttr
}

// traceSink is an OpenTelemetry SpanProcessor that captures finished spans,
// sanitizes them, and feeds them to the trace view hub. OnEnd runs synchronously
// inside span.End(), so it does only the cheap copy + ring append + dirty flag;
// rendering and fan-out happen off this path (the hub's frame renderer, which
// only runs while viewers are connected).
//
// IMPORTANT: nothing on the trace render/push path may itself create spans —
// each push would mint a span that wakes viewers that push again, a feedback
// loop. The /traces/stream SSE is therefore excluded from HTTP tracing (see
// main) and the push path is uninstrumented.
type traceSink struct{ hub *traceHub }

func (t traceSink) OnStart(context.Context, sdktrace.ReadWriteSpan) {}
func (t traceSink) Shutdown(context.Context) error                  { return nil }
func (t traceSink) ForceFlush(context.Context) error                { return nil }

func (t traceSink) OnEnd(s sdktrace.ReadOnlySpan) {
	t.hub.add(sanitizeSpan(s))
}

// sanitizeSpan copies a finished span into a traceSpan, keeping only allow-listed
// attributes. The error status is reduced to a bool: span.RecordError records an
// exception event whose message can carry arbitrary content, so the description
// is dropped and only the fact of an error survives.
func sanitizeSpan(s sdktrace.ReadOnlySpan) traceSpan {
	var attrs []traceAttr
	for _, kv := range s.Attributes() {
		if traceAttrAllow[kv.Key] {
			attrs = append(attrs, traceAttr{k: string(kv.Key), v: kv.Value.Emit()})
		}
	}
	sc := s.SpanContext()
	return traceSpan{
		traceID:  sc.TraceID().String(),
		spanID:   sc.SpanID().String(),
		parentID: s.Parent().SpanID().String(),
		name:     s.Name(),
		start:    s.StartTime(),
		dur:      s.EndTime().Sub(s.StartTime()),
		isError:  s.Status().Code == codes.Error,
		attrs:    attrs,
	}
}

// traceCap is how many recent spans the public view retains. A small ring: the
// view is a live window, not a store, and the bound holds memory flat regardless
// of span rate.
const traceCap = 512

// traceHub is the pub/sub + ring buffer behind the /traces view. It is a
// separate fan-out from the issue Hub: every viewer sees the same global span
// feed (no per-viewer state), so there is no session map — just a ring of recent
// spans and subscriber channels woken once per frame. The frame coalescing
// mirrors the issue Hub: a burst of spans collapses into one push.
//
// Because the view is identical for everyone, the hub renders it ONCE per frame
// (cur/curSeq below) and the streams just write the shared string — N viewers
// cost one render, not N.
type traceHub struct {
	mu    sync.Mutex
	ring  [traceCap]traceSpan
	n     int // total spans ever appended; live slots are [max(0,n-cap), n)
	subs  map[chan struct{}]struct{}
	dirty atomic.Bool
	salt  uint64 // per-process seed for the anonymized trace display handle (anonTrace)

	// The shared render: the last #traces region produced, its hash (dedup —
	// an unchanged render publishes nothing), and a sequence number bumped on
	// every publish so each stream knows whether it has written cur yet.
	cur     string
	curHash uint64
	curSeq  uint64
}

func newTraceHub(frame time.Duration) *traceHub {
	h := &traceHub{subs: make(map[chan struct{}]struct{}), salt: randomSalt()}
	h.render() // seed cur (the empty state) so a first paint never races the ticker
	go h.runFrames(frame)
	return h
}

func (h *traceHub) add(s traceSpan) {
	h.mu.Lock()
	h.ring[h.n%traceCap] = s
	h.n++
	h.mu.Unlock()
	h.dirty.Store(true)
}

// snapshot returns the retained spans in arrival order (oldest first).
func (h *traceHub) snapshot() []traceSpan {
	h.mu.Lock()
	defer h.mu.Unlock()
	start := 0
	if h.n > traceCap {
		start = h.n - traceCap
	}
	out := make([]traceSpan, 0, h.n-start)
	for i := start; i < h.n; i++ {
		out = append(out, h.ring[i%traceCap])
	}
	return out
}

// subscribe registers a viewer. Presence is part of the rendered view (the
// "N watching" count), so joining marks the region dirty like a new span does.
func (h *traceHub) subscribe() chan struct{} {
	ch := make(chan struct{}, 1)
	h.mu.Lock()
	h.subs[ch] = struct{}{}
	h.mu.Unlock()
	h.dirty.Store(true)
	return ch
}

func (h *traceHub) unsubscribe(ch chan struct{}) {
	h.mu.Lock()
	delete(h.subs, ch)
	h.mu.Unlock()
	h.dirty.Store(true)
}

func (h *traceHub) viewers() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return len(h.subs)
}

// current returns the shared rendered region and its sequence number. A stream
// pushes only when the seq has moved past what it last wrote.
func (h *traceHub) current() (string, uint64) {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.cur, h.curSeq
}

// render builds the #traces region once from the current ring + presence count
// and, if the bytes changed, publishes it and wakes every subscriber
// (non-blocking, coalescing — a stream with a refresh already pending is
// skipped). This is the single render all viewers share.
func (h *traceHub) render() {
	// Clear dirty BEFORE snapshotting: a span landing mid-render re-marks it
	// and the next frame picks it up, instead of being lost.
	h.dirty.Store(false)
	spans := h.snapshot()
	region, err := renderToString("traces", buildTraceView(spans, h.salt, h.viewers()))
	if err != nil {
		log.Printf("trace view render: %v", err)
		return
	}
	hash := hashStr(region)
	h.mu.Lock()
	if hash != h.curHash {
		h.cur, h.curHash = region, hash
		h.curSeq++
		for ch := range h.subs {
			select {
			case ch <- struct{}{}:
			default:
			}
		}
	}
	h.mu.Unlock()
}

// runFrames renders the shared region once per frame when something changed
// (new spans or presence) — but only while someone is watching. With zero
// viewers dirty just stays set and the ring keeps filling, so an unwatched
// view costs span capture only, never a render.
func (h *traceHub) runFrames(frame time.Duration) {
	t := time.NewTicker(frame)
	defer t.Stop()
	for range t.C {
		if h.dirty.Load() && h.viewers() > 0 {
			h.render()
		}
	}
}

// ---- view models ----------------------------------------------------------

type traceRowView struct {
	Depth   int
	Name    string
	Dur     string
	IsError bool
	Attrs   string
	Cat     string       // span category (db/cache/render/push/write/http/other) → bar color
	Bar     template.CSS // "left:X%;width:Y%" — waterfall position within the trace window
}

type traceGroupView struct {
	ShortID string
	Root    string
	When    string
	Count   int
	Dur     string
	IsError bool
	Spans   []traceRowView
}

type traceView struct {
	Groups  []traceGroupView
	Spans   int // total spans retained in the ring
	Viewers int // live presence: how many streams are watching this page right now
}

// maxTraceGroups bounds how many recent traces the view shows at once (the rest
// of the ring still counts toward Spans, it just isn't drawn).
const maxTraceGroups = 16

// buildTraceView groups the retained spans by trace id into newest-first cards.
// Within a card spans are ordered by start time and indented by their depth in
// the parent chain (computed only over spans present in the ring, so a card
// whose root has aged out still renders sensibly rooted at its shallowest span).
func buildTraceView(spans []traceSpan, salt uint64, viewers int) traceView {
	idx := map[string]int{}
	type group struct {
		id    string
		spans []traceSpan
	}
	var groups []*group
	for _, s := range spans {
		i, ok := idx[s.traceID]
		if !ok {
			i = len(groups)
			idx[s.traceID] = i
			groups = append(groups, &group{id: s.traceID})
		}
		groups[i].spans = append(groups[i].spans, s)
	}

	type ranked struct {
		v   traceGroupView
		end time.Time
	}
	ranks := make([]ranked, 0, len(groups))
	for _, g := range groups {
		byID := make(map[string]traceSpan, len(g.spans))
		for _, s := range g.spans {
			byID[s.spanID] = s
		}
		ordered := append([]traceSpan(nil), g.spans...)
		sort.Slice(ordered, func(i, j int) bool { return ordered[i].start.Before(ordered[j].start) })

		// Pass 1: the trace's time window + per-span depth + headline root. The
		// waterfall bars (pass 2) are positioned relative to [minStart, maxEnd].
		gv := traceGroupView{ShortID: anonTrace(g.id, salt), Count: len(ordered)}
		depths := make([]int, len(ordered))
		var minStart, maxEnd time.Time
		rootDur := time.Duration(-1)
		for i, s := range ordered {
			depths[i] = spanDepth(s, byID)
			if s.isError {
				gv.IsError = true
			}
			if minStart.IsZero() || s.start.Before(minStart) {
				minStart = s.start
			}
			if e := s.start.Add(s.dur); e.After(maxEnd) {
				maxEnd = e
			}
			// Headline span is the longest depth-0 span (the request / flush / push
			// root); its duration is the wall-clock the trace took.
			if depths[i] == 0 && s.dur > rootDur {
				rootDur, gv.Root = s.dur, s.name
			}
		}
		total := maxEnd.Sub(minStart)

		// Pass 2: the rows, each with a time-positioned waterfall bar.
		for i, s := range ordered {
			off, w := barPct(s.start.Sub(minStart), s.dur, total)
			gv.Spans = append(gv.Spans, traceRowView{
				Depth:   depths[i],
				Name:    s.name,
				Dur:     durStr(s.dur),
				IsError: s.isError,
				Attrs:   joinAttrs(s.attrs),
				Cat:     spanCat(s.name),
				Bar:     template.CSS(fmt.Sprintf("left:%.2f%%;width:%.2f%%", off, w)),
			})
		}
		if gv.Root == "" && len(ordered) > 0 {
			gv.Root = ordered[0].name
		}
		if rootDur >= 0 {
			gv.Dur = durStr(rootDur)
		} else {
			gv.Dur = durStr(total)
		}
		gv.When = maxEnd.Format("15:04:05.000")
		ranks = append(ranks, ranked{v: gv, end: maxEnd})
	}
	sort.Slice(ranks, func(i, j int) bool { return ranks[i].end.After(ranks[j].end) })

	out := traceView{Spans: len(spans), Viewers: viewers}
	for i := range ranks {
		if i >= maxTraceGroups {
			break
		}
		out.Groups = append(out.Groups, ranks[i].v)
	}
	return out
}

// spanDepth counts how many ancestors of s are present in the ring snapshot. A
// span whose parent has aged out (or has no parent) is depth 0. The seen guard
// is belt-and-suspenders against a pathological parent cycle.
func spanDepth(s traceSpan, byID map[string]traceSpan) int {
	depth := 0
	cur := s
	seen := map[string]bool{cur.spanID: true}
	for {
		p, ok := byID[cur.parentID]
		if !ok || seen[p.spanID] {
			return depth
		}
		depth++
		seen[p.spanID] = true
		cur = p
	}
}

// barPct positions a span's waterfall bar within its trace window: a left
// offset and a width, both as percentages of [minStart, maxEnd]. A zero-duration
// window (a single instantaneous span) renders full width; a tiny but non-zero
// span keeps a minimum width so it stays visible, clamped to not overflow.
func barPct(offset, dur, total time.Duration) (off, w float64) {
	if total <= 0 {
		return 0, 100
	}
	off = float64(offset) / float64(total) * 100
	w = float64(dur) / float64(total) * 100
	if off < 0 {
		off = 0
	} else if off > 100 {
		off = 100
	}
	if w < 1.5 {
		w = 1.5
	}
	if off+w > 100 {
		w = 100 - off
	}
	return off, w
}

// spanCat buckets a span into a waterfall color category from its name (the only
// field both safe and stable). Mirrors the span families the app emits; unknown
// names fall through to "other".
func spanCat(name string) string {
	switch {
	case strings.HasPrefix(name, "db.") || name == "detail.read": // detail.read is a GetIssue DB read
		return "db"
	case strings.HasSuffix(name, "cache.fill"):
		return "cache"
	case strings.HasPrefix(name, "render."):
		return "render"
	case strings.HasPrefix(name, "sse."): // sse.push (root) + sse.send (the send half)
		return "push"
	case strings.HasPrefix(name, "writer."):
		return "write"
	case isHTTPName(name):
		return "http"
	default:
		return "other"
	}
}

func isHTTPName(name string) bool {
	for _, m := range [...]string{"GET ", "POST ", "PUT ", "DELETE ", "PATCH ", "HEAD ", "OPTIONS "} {
		if strings.HasPrefix(name, m) {
			return true
		}
	}
	return false
}

// anonWidth is the fixed length of the per-trace display handle.
const anonWidth = 6

// anonTrace derives a trace card's display handle: a base36 token hashed from
// the real trace id mixed with a per-process random salt. It is stable within a
// run (a trace always renders the same handle, so a card keeps its id across
// morphs) yet reveals nothing about the real trace id — and because the salt is
// process-private, a client that chose its own id via an inbound traceparent
// cannot reproduce or recognize its handle here. Grouping still keys on the real
// id; only the displayed label is anonymized.
func anonTrace(id string, salt uint64) string {
	h := fnv.New64a()
	var s [8]byte
	binary.LittleEndian.PutUint64(s[:], salt)
	_, _ = h.Write(s[:])
	_, _ = h.Write([]byte(id))
	const digits = "0123456789abcdefghijklmnopqrstuvwxyz"
	n := h.Sum64()
	var buf [anonWidth]byte
	for i := anonWidth - 1; i >= 0; i-- {
		buf[i] = digits[n%36]
		n /= 36
	}
	return string(buf[:])
}

// randomSalt seeds anonTrace once per process. crypto/rand does not realistically
// fail here; a zero salt on the impossible error is a harmless fallback.
func randomSalt() uint64 {
	var b [8]byte
	_, _ = rand.Read(b[:])
	return binary.LittleEndian.Uint64(b[:])
}

func joinAttrs(attrs []traceAttr) string {
	if len(attrs) == 0 {
		return ""
	}
	var b strings.Builder
	for i, a := range attrs {
		if i > 0 {
			b.WriteString(" · ")
		}
		b.WriteString(a.k)
		b.WriteByte('=')
		b.WriteString(a.v)
	}
	return b.String()
}

func durStr(d time.Duration) string {
	if d < time.Second {
		return fmt.Sprintf("%.2fms", float64(d.Microseconds())/1000)
	}
	return fmt.Sprintf("%.3fs", d.Seconds())
}

// ---- handlers --------------------------------------------------------------

// handleTraces serves the public, read-only trace view page. First paint is
// fully server-rendered from the current ring (no dependence on the stream);
// the SSE stream then morphs #traces live, exactly like the issue routes.
// The render path stays deliberately uninstrumented (no tracer.Start anywhere
// it reaches): a span minted while rendering would itself wake viewers and
// push again, a feedback loop.
func (a *app) handleTraces(w http.ResponseWriter, r *http.Request) {
	// Render on demand: with no stream viewers the frame ticker idles, so cur
	// may predate recent spans. One render per page load is cheap, and a
	// changed result is published to any connected streams too.
	a.traces.render()
	region, _ := a.traces.current()
	if region == "" {
		http.Error(w, "render failed", http.StatusInternalServerError)
		return
	}
	data := struct {
		AssetVer string
		Traces   template.HTML
	}{assetVer, template.HTML(region)}
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	if err := tmpl.ExecuteTemplate(w, "tracepage", data); err != nil {
		log.Printf("traces page: %v", err)
	}
}

// handleTraceStream is the long-lived SSE read stream for the trace view: it
// subscribes to the hub and writes the hub's SHARED render whenever a new one
// is published (coalesced to one per frame) — the stream itself never renders,
// so a thousand viewers still cost one render per frame. Plain (uncompressed):
// the region is small (≤ maxTraceGroups cards) and skipping compression keeps
// the public stream simple.
func (a *app) handleTraceStream(w http.ResponseWriter, r *http.Request) {
	sse := datastar.NewSSE(w, r)
	ch := a.traces.subscribe()
	defer a.traces.unsubscribe(ch)

	var lastSeq uint64
	push := func() error {
		region, seq := a.traces.current()
		if seq == lastSeq {
			return nil
		}
		if err := sse.PatchElements(region); err != nil {
			return err
		}
		lastSeq = seq
		return nil
	}
	if err := push(); err != nil {
		return
	}

	// Keepalive so a NAT/proxy doesn't drop an idle stream. `_ka` is
	// underscore-prefixed → Datastar never echoes it back to the server.
	ka := time.NewTicker(15 * time.Second)
	defer ka.Stop()
	ctx := r.Context()
	kaCount := 0
	for {
		select {
		case <-ctx.Done():
			return
		case <-a.shutdown:
			return
		case <-ch:
			if err := push(); err != nil {
				return
			}
		case <-ka.C:
			if err := sse.MarshalAndPatchSignals(map[string]any{"_ka": kaCount}); err != nil {
				return
			}
			kaCount++
		}
	}
}