package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"sort"
	"strconv"
	"sync"
	"syscall"
	"time"
)

// ---------------------------------------------------------------------------
// Distributed collector — HTTP dashboard that ingests events from workers
// ---------------------------------------------------------------------------
//
// Endpoints:
//   POST /events             ingest a wireBatch from a worker container
//   GET  /api/state          JSON snapshot for the live dashboard
//   GET  /api/events?cursor= JSON event log slice past the cursor
//   GET  /                   the dashboard HTML page
//
// Data model:
//   workers keyed by "<containerID>/<workerName>" — same uiStats shape the
//   TUI uses. A 1-second tick goroutine computes throughput deltas and p95
//   over a sliding window of recent latencies.

type collectorWorker struct {
	uiStats
	containerID string
	workerName  string
}

type collectorState struct {
	mu sync.Mutex

	startAt time.Time

	// keyed by "<container>/<worker>"; iteration order preserved via slice
	keys    []string
	workers map[string]*collectorWorker

	events     []eventLine
	nextCursor int64

	inconsistencies     []eventLine
	nextInconsCursor    int64

	// global series for the top-of-dashboard charts
	p95Series           []float64
	okPerSecSeries      []float64
	failPerSecSeries    []float64
	missingPerSecSeries []float64
	globalLat           []time.Duration
	prevTotalOps        int64
	prevTotalFail       int64
	prevTotalMissing    int64
}

func newCollectorState() *collectorState {
	return &collectorState{
		startAt: time.Now(),
		workers: make(map[string]*collectorWorker),
	}
}

// ---- ingest -------------------------------------------------------------

func (c *collectorState) ingest(b wireBatch) {
	c.mu.Lock()
	defer c.mu.Unlock()

	for _, ev := range b.Events {
		switch ev.Type {
		case "op":
			c.applyOp(b.ContainerID, ev)
		case "sub":
			c.applySub(b.ContainerID, ev)
		case "ledger":
			c.applyLedger(b.ContainerID, ev)
		case "info":
			c.appendEvent(ev.Level, ev.At, ev.Text)
		}
	}
}

func (c *collectorState) workerByKey(containerID, workerName string) *collectorWorker {
	key := containerID + "/" + workerName
	w, ok := c.workers[key]
	if !ok {
		w = &collectorWorker{containerID: containerID, workerName: workerName}
		c.workers[key] = w
		c.keys = append(c.keys, key)
		sort.Strings(c.keys)
	}
	return w
}

func (c *collectorState) applyOp(containerID string, ev wireEvent) {
	w := c.workerByKey(containerID, ev.Worker)
	if w.role == "" {
		w.role = roleFromOp(ev.Op)
	}
	w.ops++
	if !ev.OK {
		// Surface every individual failure in the dashboard event log so
		// the operator can see what's actually breaking, not just the
		// summarized OUTAGE START/END boundaries.
		c.appendEvent("err", ev.At,
			fmt.Sprintf("[%s/%s] %s FAIL: %s",
				containerID, ev.Worker, ev.Op, trim(ev.Err, 200)))
		w.fail++
		w.consecFail++
		if !w.inOutage {
			w.inOutage = true
			w.outageStart = ev.At
			w.outageOpsLost = 1
			c.appendEvent("err", ev.At,
				fmt.Sprintf("[%s/%s] OUTAGE START", containerID, ev.Worker))
		} else {
			w.outageOpsLost++
		}
		return
	}
	w.ok++
	if w.inOutage {
		w.outages = append(w.outages, outageRec{
			start: w.outageStart, end: ev.At, opsLost: w.outageOpsLost,
		})
		w.inOutage = false
		w.consecFail = 0
		w.outageOpsLost = 0
	}
	if ev.DurUS > 0 {
		dur := time.Duration(ev.DurUS) * time.Microsecond
		w.recentLat = appendBoundedDur(w.recentLat, dur, maxRecentLat)
		c.globalLat = appendBoundedDur(c.globalLat, dur, maxGlobalLat)
	}
}

func (c *collectorState) applySub(containerID string, ev wireEvent) {
	w := c.workerByKey(containerID, ev.Worker)
	if w.role == "" {
		w.role = "subscriber"
	}
	w.msgs++
	if w.lastSeq != 0 && ev.Seq > w.lastSeq+1 {
		gap := ev.Seq - w.lastSeq - 1
		w.gaps += gap
	}
	if !w.lastSeqAt.IsZero() {
		if d := ev.At.Sub(w.lastSeqAt); d > w.longestGap {
			w.longestGap = d
		}
	}
	w.lastSeq = ev.Seq
	w.lastSeqAt = ev.At
}

func (c *collectorState) applyLedger(containerID string, ev wireEvent) {
	w := c.workerByKey(containerID, ev.Worker)
	if w.role == "" {
		w.role = "reader"
	}
	switch ev.Kind {
	case "ok":
		w.verified++
	case "missing":
		w.missing++
		c.appendInconsistency("err", ev.At,
			fmt.Sprintf("[%s/%s] LEDGER MISSING seq=%d  %s",
				containerID, ev.Worker, ev.Seq, ev.Detail))
	case "mismatch":
		w.mismatch++
		c.appendInconsistency("err", ev.At,
			fmt.Sprintf("[%s/%s] LEDGER MISMATCH seq=%d  %s",
				containerID, ev.Worker, ev.Seq, ev.Detail))
	}
}

// roleFromOp infers a worker's role from the first op name we see, since
// workers don't currently transmit role with each event. Best-effort only.
func roleFromOp(op string) string {
	switch op {
	case "WRITE":
		return "writer"
	case "RW":
		return "readwriter"
	case "GET", "PING":
		return "reader"
	case "PUB":
		return "publisher"
	case "SUBSCRIBE":
		return "subscriber"
	case "BLOAT":
		return "bloater"
	}
	return ""
}

func (c *collectorState) appendEvent(level string, at time.Time, text string) {
	if at.IsZero() {
		at = time.Now()
	}
	if level == "" {
		level = "info"
	}
	if len(c.events) >= maxEvents {
		copy(c.events, c.events[1:])
		c.events = c.events[:len(c.events)-1]
	}
	c.events = append(c.events, eventLine{at: at, level: level, text: text})
	c.nextCursor++
}

func (c *collectorState) appendInconsistency(level string, at time.Time, text string) {
	if at.IsZero() {
		at = time.Now()
	}
	if level == "" {
		level = "info"
	}
	if len(c.inconsistencies) >= maxEvents {
		copy(c.inconsistencies, c.inconsistencies[1:])
		c.inconsistencies = c.inconsistencies[:len(c.inconsistencies)-1]
	}
	c.inconsistencies = append(c.inconsistencies, eventLine{at: at, level: level, text: text})
	c.nextInconsCursor++
}

// ---- tick (1Hz) ---------------------------------------------------------

func (c *collectorState) tick() {
	c.mu.Lock()
	defer c.mu.Unlock()

	if len(c.globalLat) > 0 {
		cp := make([]time.Duration, len(c.globalLat))
		copy(cp, c.globalLat)
		sort.Slice(cp, func(i, j int) bool { return cp[i] < cp[j] })
		idx := 95 * len(cp) / 100
		if idx >= len(cp) {
			idx = len(cp) - 1
		}
		p95ms := float64(cp[idx]) / float64(time.Millisecond)
		c.p95Series = appendBoundedFloat(c.p95Series, p95ms, maxSeriesPoints)
	} else {
		c.p95Series = appendBoundedFloat(c.p95Series, 0, maxSeriesPoints)
	}

	var totalOps, totalFail, totalMissing int64
	for _, w := range c.workers {
		totalOps += w.ops + w.msgs
		totalFail += w.fail
		totalMissing += w.missing
	}
	okDelta := (totalOps - totalFail) - (c.prevTotalOps - c.prevTotalFail)
	if okDelta < 0 {
		okDelta = 0
	}
	failDelta := totalFail - c.prevTotalFail
	if failDelta < 0 {
		failDelta = 0
	}
	missingDelta := totalMissing - c.prevTotalMissing
	if missingDelta < 0 {
		missingDelta = 0
	}
	c.prevTotalOps = totalOps
	c.prevTotalFail = totalFail
	c.prevTotalMissing = totalMissing
	c.okPerSecSeries = appendBoundedFloat(c.okPerSecSeries, float64(okDelta), maxSeriesPoints)
	c.failPerSecSeries = appendBoundedFloat(c.failPerSecSeries, float64(failDelta), maxSeriesPoints)
	c.missingPerSecSeries = appendBoundedFloat(c.missingPerSecSeries, float64(missingDelta), maxSeriesPoints)

	for _, w := range c.workers {
		current := w.ops + w.msgs
		delta := current - w.prevOps
		if delta < 0 {
			delta = 0
		}
		w.prevOps = current
		w.opsHist = appendBoundedFloat(w.opsHist, float64(delta), maxSparkSamples)
	}
}

// ---- snapshot view ------------------------------------------------------

type stateView struct {
	StartAt    time.Time         `json:"start_at"`
	Elapsed    string            `json:"elapsed"`
	Containers int               `json:"containers"`
	InOutage   int               `json:"in_outage"`
	Workers    []workerView      `json:"workers"`
	Charts     chartsView        `json:"charts"`
}

type workerView struct {
	Container  string  `json:"container"`
	Name       string  `json:"name"`
	Role       string  `json:"role"`
	Ops        int64   `json:"ops"`
	OK         int64   `json:"ok"`
	Fail       int64   `json:"fail"`
	UptimePct  float64 `json:"uptime_pct"`
	P50MS      float64 `json:"p50_ms"`
	P95MS      float64 `json:"p95_ms"`
	OpsHist    []float64 `json:"ops_hist"`
	InOutage   bool    `json:"in_outage"`
	Outages    int     `json:"outages"`
	LongestMS  float64 `json:"longest_outage_ms"`
	// reader-only
	Verified int64 `json:"verified,omitempty"`
	Missing  int64 `json:"missing,omitempty"`
	Mismatch int64 `json:"mismatch,omitempty"`
	// subscriber-only
	Msgs         int64   `json:"msgs,omitempty"`
	Gaps         int64   `json:"gaps,omitempty"`
	LongestGapMS float64 `json:"longest_gap_ms,omitempty"`
}

type chartsView struct {
	P95           []float64 `json:"p95"`
	OkPerSec      []float64 `json:"ok_per_sec"`
	FailPerSec    []float64 `json:"fail_per_sec"`
	MissingPerSec []float64 `json:"missing_per_sec"`
	P95Peak       float64   `json:"p95_peak"`
	OkPeak        float64   `json:"ok_peak"`
	FailPeak      float64   `json:"fail_peak"`
	MissingPeak   float64   `json:"missing_peak"`
}

func (c *collectorState) snapshot() stateView {
	c.mu.Lock()
	defer c.mu.Unlock()

	containers := map[string]struct{}{}
	for _, k := range c.keys {
		w := c.workers[k]
		containers[w.containerID] = struct{}{}
	}

	view := stateView{
		StartAt:    c.startAt,
		Elapsed:    time.Since(c.startAt).Round(time.Second).String(),
		Containers: len(containers),
		Charts: chartsView{
			P95:           append([]float64(nil), c.p95Series...),
			OkPerSec:      append([]float64(nil), c.okPerSecSeries...),
			FailPerSec:    append([]float64(nil), c.failPerSecSeries...),
			MissingPerSec: append([]float64(nil), c.missingPerSecSeries...),
			P95Peak:       maxFloat(c.p95Series),
			OkPeak:        maxFloat(c.okPerSecSeries),
			FailPeak:      maxFloat(c.failPerSecSeries),
			MissingPeak:   maxFloat(c.missingPerSecSeries),
		},
	}
	for _, k := range c.keys {
		w := c.workers[k]
		uptime := 100.0
		if w.ops > 0 {
			uptime = float64(w.ok) / float64(w.ops) * 100
		}
		var longest time.Duration
		for _, o := range w.outages {
			if d := o.end.Sub(o.start); d > longest {
				longest = d
			}
		}
		if w.inOutage {
			view.InOutage++
		}
		wv := workerView{
			Container: w.containerID,
			Name:      w.workerName,
			Role:      w.role,
			Ops:       w.ops,
			OK:        w.ok,
			Fail:      w.fail,
			UptimePct: uptime,
			P50MS:     float64(percentileLat(w.recentLat, 50)) / float64(time.Millisecond),
			P95MS:     float64(percentileLat(w.recentLat, 95)) / float64(time.Millisecond),
			OpsHist:   append([]float64(nil), w.opsHist...),
			InOutage:  w.inOutage,
			Outages:   len(w.outages),
			LongestMS: float64(longest) / float64(time.Millisecond),
		}
		switch w.role {
		case "reader":
			wv.Verified = w.verified
			wv.Missing = w.missing
			wv.Mismatch = w.mismatch
		case "subscriber":
			wv.Msgs = w.msgs
			wv.Gaps = w.gaps
			wv.LongestGapMS = float64(w.longestGap) / float64(time.Millisecond)
		}
		view.Workers = append(view.Workers, wv)
	}
	return view
}

type eventOut struct {
	ID    int64  `json:"id"`
	At    string `json:"at"`
	Level string `json:"level"`
	Text  string `json:"text"`
}

func (c *collectorState) eventsSince(cursor int64) ([]eventOut, int64) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// nextCursor counts events ever inserted; events slice may have dropped older.
	// Compute base id of the first slot in the current slice:
	baseID := c.nextCursor - int64(len(c.events))
	if cursor < baseID {
		cursor = baseID
	}
	startIdx := int(cursor - baseID)
	if startIdx < 0 {
		startIdx = 0
	}
	if startIdx > len(c.events) {
		startIdx = len(c.events)
	}
	out := make([]eventOut, 0, len(c.events)-startIdx)
	for i := startIdx; i < len(c.events); i++ {
		e := c.events[i]
		out = append(out, eventOut{
			ID:    baseID + int64(i),
			At:    e.at.Format("15:04:05.000"),
			Level: e.level,
			Text:  e.text,
		})
	}
	return out, c.nextCursor
}

func (c *collectorState) inconsistenciesSince(cursor int64) ([]eventOut, int64) {
	c.mu.Lock()
	defer c.mu.Unlock()

	baseID := c.nextInconsCursor - int64(len(c.inconsistencies))
	if cursor < baseID {
		cursor = baseID
	}
	startIdx := int(cursor - baseID)
	if startIdx < 0 {
		startIdx = 0
	}
	if startIdx > len(c.inconsistencies) {
		startIdx = len(c.inconsistencies)
	}
	out := make([]eventOut, 0, len(c.inconsistencies)-startIdx)
	for i := startIdx; i < len(c.inconsistencies); i++ {
		e := c.inconsistencies[i]
		out = append(out, eventOut{
			ID:    baseID + int64(i),
			At:    e.at.Format("15:04:05.000"),
			Level: e.level,
			Text:  e.text,
		})
	}
	return out, c.nextInconsCursor
}

// ---- HTTP server --------------------------------------------------------

func runCollector(port int) int {
	state := newCollectorState()

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	go func() {
		t := time.NewTicker(tickInterval)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				state.tick()
			}
		}
	}()

	mux := http.NewServeMux()
	mux.HandleFunc("/events", state.handleIngest)
	mux.HandleFunc("/api/state", state.handleState)
	mux.HandleFunc("/api/events", state.handleEvents)
	mux.HandleFunc("/api/inconsistencies", state.handleInconsistencies)
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusOK)
	})
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/" {
			http.NotFound(w, r)
			return
		}
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		_, _ = w.Write([]byte(dashboardHTML))
	})

	srv := &http.Server{
		Addr:              fmt.Sprintf(":%d", port),
		Handler:           mux,
		ReadHeaderTimeout: 5 * time.Second,
	}

	go func() {
		log.Printf("collector listening on :%d", port)
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("collector: %v", err)
		}
	}()

	<-ctx.Done()
	log.Printf("collector shutting down…")
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = srv.Shutdown(shutdownCtx)
	return 0
}

func (c *collectorState) handleIngest(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "POST only", http.StatusMethodNotAllowed)
		return
	}
	var b wireBatch
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&b); err != nil {
		http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest)
		return
	}
	c.ingest(b)
	w.WriteHeader(http.StatusNoContent)
}

func (c *collectorState) handleState(w http.ResponseWriter, _ *http.Request) {
	view := c.snapshot()
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(view)
}

func (c *collectorState) handleEvents(w http.ResponseWriter, r *http.Request) {
	cursor := int64(0)
	if v := r.URL.Query().Get("cursor"); v != "" {
		if n, err := strconv.ParseInt(v, 10, 64); err == nil {
			cursor = n
		}
	}
	events, next := c.eventsSince(cursor)
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{
		"events":      events,
		"next_cursor": next,
	})
}

func (c *collectorState) handleInconsistencies(w http.ResponseWriter, r *http.Request) {
	cursor := int64(0)
	if v := r.URL.Query().Get("cursor"); v != "" {
		if n, err := strconv.ParseInt(v, 10, 64); err == nil {
			cursor = n
		}
	}
	events, next := c.inconsistenciesSince(cursor)
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{
		"events":      events,
		"next_cursor": next,
	})
}

// ---- embedded dashboard HTML -------------------------------------------

const dashboardHTML = `<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>valkey-ha chaos</title>
<meta name="viewport" content="width=device-width,initial-scale=1">
<style>
  :root {
    --bg: #0c0c10;
    --panel: #15151c;
    --border: #2a2a35;
    --text: #e6e6ea;
    --dim: #6c6c7a;
    --accent: #d75faf;
    --ok: #5fd75f;
    --warn: #ffaf5f;
    --err: #ff5f5f;
    --chart: #af87ff;
    --chart-fail: #ff5f5f;
  }
  * { box-sizing: border-box; }
  body { font-family: ui-monospace, 'JetBrains Mono', Menlo, Consolas, monospace;
         background: var(--bg); color: var(--text); margin: 0; padding: 16px;
         font-size: 13px; line-height: 1.45; }
  h1 { color: var(--accent); margin: 0 0 4px; font-size: 18px; }
  .header-meta { color: var(--dim); margin-bottom: 14px; }
  .badge { padding: 1px 6px; border-radius: 3px; font-weight: 600; }
  .badge.ok   { background: #1c3a1c; color: var(--ok); }
  .badge.err  { background: #401818; color: var(--err); }
  .badge.warn { background: #402a18; color: var(--warn); }
  .grid { display: grid; grid-template-columns: 1fr 1fr 1fr 1fr; gap: 14px; margin-bottom: 16px; }
  @media (max-width: 1400px) { .grid { grid-template-columns: 1fr 1fr; } }
  @media (max-width: 700px)  { .grid { grid-template-columns: 1fr; } }
  .panel { background: var(--panel); border: 1px solid var(--border);
           border-radius: 4px; padding: 10px 12px; }
  .panel h3 { margin: 0 0 6px; color: #5fafff; font-size: 12px;
              text-transform: uppercase; letter-spacing: 0.06em; }
  .panel .meta { color: var(--dim); font-size: 11px; margin-bottom: 4px; }
  table { border-collapse: collapse; width: 100%; font-size: 12px; }
  th, td { padding: 4px 8px; text-align: left; vertical-align: middle;
           border-bottom: 1px solid var(--border); }
  th { color: #5fafff; font-weight: 600; text-transform: uppercase;
       font-size: 10px; letter-spacing: 0.05em; }
  td.num { text-align: right; font-variant-numeric: tabular-nums; }
  tr.outage td:first-child { border-left: 3px solid var(--err); }
  .container-tag { color: var(--accent); }
  .events { background: var(--panel); border: 1px solid var(--border);
            border-radius: 4px; padding: 8px 10px; max-height: 50vh;
            overflow-y: auto; font-size: 12px; }
  .events .ev { padding: 1px 0; white-space: pre-wrap; word-break: break-word; }
  .events .ev .ts { color: var(--dim); margin-right: 6px; }
  .events .ev.err  { color: var(--err); }
  .events .ev.warn { color: var(--warn); }
  .events .ev.info { color: var(--text); }
  svg { display: block; width: 100%; height: 80px; }
  polyline { fill: none; stroke: var(--chart); stroke-width: 1.4; }
  .area    { fill: var(--chart); fill-opacity: 0.18; stroke: none; }
  .spark   { font-family: ui-monospace, monospace; color: var(--chart); }
  .dim { color: var(--dim); }
  .filter { margin-bottom: 8px; }
  .filter input { background: var(--panel); color: var(--text);
                  border: 1px solid var(--border); padding: 4px 8px;
                  border-radius: 3px; font: inherit; }
  .footer { color: var(--dim); margin-top: 12px; font-size: 11px; }
</style>
</head>
<body>
<h1>valkey-ha chaos</h1>
<div class="header-meta" id="header">connecting…</div>

<div class="grid">
  <div class="panel">
    <h3>p95 latency (rolling 2 min)</h3>
    <div class="meta" id="lat-meta">—</div>
    <svg id="chart-lat" viewBox="0 0 100 30" preserveAspectRatio="none"></svg>
  </div>
  <div class="panel">
    <h3>ok ops/sec (rolling 2 min)</h3>
    <div class="meta" id="tp-meta">—</div>
    <svg id="chart-tp" viewBox="0 0 100 30" preserveAspectRatio="none"></svg>
  </div>
  <div class="panel">
    <h3>fail ops/sec (rolling 2 min)</h3>
    <div class="meta" id="fail-meta">—</div>
    <svg id="chart-fail" viewBox="0 0 100 30" preserveAspectRatio="none"></svg>
  </div>
  <div class="panel">
    <h3>missing ledger/sec (rolling 2 min)</h3>
    <div class="meta" id="miss-meta">—</div>
    <svg id="chart-miss" viewBox="0 0 100 30" preserveAspectRatio="none"></svg>
  </div>
</div>

<div class="panel" style="margin-bottom:16px">
  <h3>workers</h3>
  <div class="filter">
    <input id="filter" placeholder="filter by container or worker name" />
  </div>
  <table id="workers">
    <thead><tr>
      <th>CONTAINER</th><th>NAME</th><th>ROLE</th>
      <th class="num">OPS</th><th class="num">OK</th><th class="num">FAIL</th>
      <th class="num">UPTIME</th><th class="num">P50</th><th class="num">P95</th>
      <th>SPARK</th><th class="num">OUTAGES</th><th>STATUS</th><th>NOTES</th>
    </tr></thead>
    <tbody></tbody>
  </table>
</div>

<div>
  <h3 style="color:#5fafff;font-size:12px;text-transform:uppercase;
             letter-spacing:0.06em;margin:0 0 6px">client errors</h3>
  <div class="events" id="events"></div>
</div>

<div style="margin-top:16px">
  <h3 style="color:#5fafff;font-size:12px;text-transform:uppercase;
             letter-spacing:0.06em;margin:0 0 6px">ledger inconsistencies</h3>
  <div class="events" id="inconsistencies"></div>
</div>

<div class="footer" id="footer">collector dashboard · poll 1s</div>

<script>
const sparkChars = [' ','▁','▂','▃','▄','▅','▆','▇','█'];
function spark(values, width) {
  if (!values || !values.length) return '';
  const max = Math.max(...values);
  if (max <= 0) return ' '.repeat(width);
  let out = '';
  for (let col = 0; col < width; col++) {
    const idx = Math.min(values.length - 1, Math.floor(col * values.length / width));
    const v = values[idx];
    const ratio = v / max;
    const i = Math.max(0, Math.min(8, Math.floor(ratio * 8)));
    out += sparkChars[i];
  }
  return out;
}

function drawChart(svgID, values, color) {
  const stroke = color || 'var(--chart)';
  const svg = document.getElementById(svgID);
  svg.innerHTML = '';
  if (!values || !values.length) return;
  const max = Math.max(1, ...values);
  const w = 100, h = 30;
  const step = w / Math.max(1, values.length - 1);
  let path = '';
  for (let i = 0; i < values.length; i++) {
    const x = i * step;
    const y = h - (values[i] / max) * h;
    path += (i === 0 ? 'M' : 'L') + x.toFixed(2) + ',' + y.toFixed(2) + ' ';
  }
  const ns = 'http://www.w3.org/2000/svg';
  const area = document.createElementNS(ns, 'path');
  area.setAttribute('fill', stroke);
  area.setAttribute('fill-opacity', '0.18');
  area.setAttribute('stroke', 'none');
  area.setAttribute('d', path + 'L' + w + ',' + h + ' L0,' + h + ' Z');
  svg.appendChild(area);
  const line = document.createElementNS(ns, 'path');
  line.setAttribute('d', path);
  line.setAttribute('fill', 'none');
  line.setAttribute('stroke', stroke);
  line.setAttribute('stroke-width', '1.2');
  svg.appendChild(line);
}

function fmt(n) { return n == null ? '-' : n.toLocaleString(); }
function fmtMS(v) {
  if (v == null || v === 0) return '-';
  if (v < 1) return (v * 1000).toFixed(0) + 'µs';
  if (v < 1000) return v.toFixed(1) + 'ms';
  return (v / 1000).toFixed(2) + 's';
}

let cursor = 0;
let inconsCursor = 0;
let lastWorkers = [];
let filterStr = '';

document.getElementById('filter').addEventListener('input', e => {
  filterStr = e.target.value.toLowerCase();
  renderTable(lastWorkers);
});

function renderTable(workers) {
  const tbody = document.querySelector('#workers tbody');
  tbody.innerHTML = '';
  for (const w of workers) {
    if (filterStr) {
      const hay = (w.container + ' ' + w.name + ' ' + w.role).toLowerCase();
      if (!hay.includes(filterStr)) continue;
    }
    const tr = document.createElement('tr');
    if (w.in_outage) tr.classList.add('outage');
    const status = w.in_outage
      ? '<span class="badge err">OUTAGE</span>'
      : (w.role === 'subscriber' && w.msgs === 0
          ? '<span class="badge warn">NO MSGS</span>'
          : '<span class="badge ok">OK</span>');
    let notes = '';
    if (w.role === 'reader') {
      notes = '<span class="dim">v=' + fmt(w.verified) +
              ' miss=' + fmt(w.missing) +
              ' mis=' + fmt(w.mismatch) + '</span>';
    } else if (w.role === 'subscriber') {
      notes = '<span class="dim">msgs=' + fmt(w.msgs) +
              ' gaps=' + fmt(w.gaps) +
              ' longest_gap=' + fmtMS(w.longest_gap_ms) + '</span>';
    }
    tr.innerHTML =
      '<td><span class="container-tag">' + w.container + '</span></td>' +
      '<td>' + w.name + '</td>' +
      '<td>' + (w.role || '?') + '</td>' +
      '<td class="num">' + fmt(w.ops) + '</td>' +
      '<td class="num">' + fmt(w.ok) + '</td>' +
      '<td class="num">' + fmt(w.fail) + '</td>' +
      '<td class="num">' + (w.uptime_pct != null ? w.uptime_pct.toFixed(2) + '%' : '-') + '</td>' +
      '<td class="num">' + fmtMS(w.p50_ms) + '</td>' +
      '<td class="num">' + fmtMS(w.p95_ms) + '</td>' +
      '<td><span class="spark">' + spark(w.ops_hist, 14) + '</span></td>' +
      '<td class="num">' + fmt(w.outages) + '</td>' +
      '<td>' + status + '</td>' +
      '<td>' + notes + '</td>';
    tbody.appendChild(tr);
  }
}

async function tickState() {
  try {
    const s = await (await fetch('/api/state')).json();
    const tag = s.in_outage > 0
      ? '<span class="badge err">' + s.in_outage + ' worker(s) in outage</span>'
      : '<span class="badge ok">healthy</span>';
    document.getElementById('header').innerHTML =
      s.containers + ' container(s) · ' + s.workers.length + ' worker(s) · elapsed ' +
      s.elapsed + ' · ' + tag;
    document.getElementById('lat-meta').textContent =
      'peak ' + s.charts.p95_peak.toFixed(1) + 'ms · last ' +
      (s.charts.p95.length ? s.charts.p95[s.charts.p95.length-1].toFixed(1) + 'ms' : '-');
    document.getElementById('tp-meta').textContent =
      'peak ' + Math.round(s.charts.ok_peak) + ' · last ' +
      (s.charts.ok_per_sec.length ? Math.round(s.charts.ok_per_sec[s.charts.ok_per_sec.length-1]) : '-');
    document.getElementById('fail-meta').textContent =
      'peak ' + Math.round(s.charts.fail_peak) + ' · last ' +
      (s.charts.fail_per_sec.length ? Math.round(s.charts.fail_per_sec[s.charts.fail_per_sec.length-1]) : '-');
    document.getElementById('miss-meta').textContent =
      'peak ' + Math.round(s.charts.missing_peak) + ' · last ' +
      (s.charts.missing_per_sec.length ? Math.round(s.charts.missing_per_sec[s.charts.missing_per_sec.length-1]) : '-');
    drawChart('chart-lat', s.charts.p95);
    drawChart('chart-tp', s.charts.ok_per_sec);
    drawChart('chart-fail', s.charts.fail_per_sec, 'var(--chart-fail)');
    drawChart('chart-miss', s.charts.missing_per_sec, 'var(--chart-fail)');
    lastWorkers = s.workers;
    renderTable(s.workers);
  } catch (e) {
    document.getElementById('header').textContent = 'fetch error: ' + e;
  }
}

async function tickEvents() {
  try {
    const r = await (await fetch('/api/events?cursor=' + cursor)).json();
    cursor = r.next_cursor;
    const div = document.getElementById('events');
    const wasAtBottom = div.scrollHeight - div.scrollTop - div.clientHeight < 30;
    for (const e of r.events) {
      const d = document.createElement('div');
      d.className = 'ev ' + (e.level || 'info');
      const ts = document.createElement('span');
      ts.className = 'ts';
      ts.textContent = e.at;
      d.appendChild(ts);
      d.appendChild(document.createTextNode(e.text));
      div.appendChild(d);
    }
    while (div.children.length > 1000) div.removeChild(div.firstChild);
    if (wasAtBottom) div.scrollTop = div.scrollHeight;
  } catch (e) { /* noop */ }
}

async function tickInconsistencies() {
  try {
    const r = await (await fetch('/api/inconsistencies?cursor=' + inconsCursor)).json();
    inconsCursor = r.next_cursor;
    const div = document.getElementById('inconsistencies');
    const wasAtBottom = div.scrollHeight - div.scrollTop - div.clientHeight < 30;
    for (const e of r.events) {
      const d = document.createElement('div');
      d.className = 'ev ' + (e.level || 'info');
      const ts = document.createElement('span');
      ts.className = 'ts';
      ts.textContent = e.at;
      d.appendChild(ts);
      d.appendChild(document.createTextNode(e.text));
      div.appendChild(d);
    }
    while (div.children.length > 1000) div.removeChild(div.firstChild);
    if (wasAtBottom) div.scrollTop = div.scrollHeight;
  } catch (e) { /* noop */ }
}

setInterval(tickState, 1000);
setInterval(tickEvents, 1000);
setInterval(tickInconsistencies, 1000);
tickState();
tickEvents();
tickInconsistencies();
</script>
</body>
</html>
`