package main

import (
	"encoding/binary"
	"encoding/json"
	"log"
	"math"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
	"unsafe"

	"github.com/gorilla/websocket"
)

// ─── WebSocket client ─────────────────────────────────────────────────────────

type wsMessage struct {
	msgType int
	data    []byte
}

type wsClient struct {
	hub  *Hub
	conn *websocket.Conn
	send chan wsMessage
}

func (c *wsClient) writePump() {
	pingTicker := time.NewTicker(30 * time.Second)
	defer func() {
		pingTicker.Stop()
		c.conn.Close()
	}()
	for {
		select {
		case msg, ok := <-c.send:
			if !ok {
				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}
			if err := c.conn.WriteMessage(msg.msgType, msg.data); err != nil {
				return
			}
		case <-pingTicker.C:
			if err := c.conn.WriteControl(websocket.PingMessage, []byte{},
				time.Now().Add(10*time.Second)); err != nil {
				return
			}
		}
	}
}

func (c *wsClient) readPump() {
	defer func() {
		c.hub.unregister <- c
		c.conn.Close()
	}()
	c.conn.SetReadLimit(64 * 1024)
	c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
	c.conn.SetPongHandler(func(string) error {
		c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
		return nil
	})
	for {
		_, msg, err := c.conn.ReadMessage()
		if err != nil {
			break
		}
		var env map[string]interface{}
		if json.Unmarshal(msg, &env) == nil {
			if t, ok := env["type"].(string); ok {
				switch t {
				case "ping":
					resp, _ := json.Marshal(map[string]string{"type": "pong"})
					select {
					case c.send <- wsMessage{websocket.TextMessage, resp}:
					default:
					}
				case "addSource":
					label, _ := env["label"].(string)
					addr, _ := env["addr"].(string)
					if addr != "" {
						select {
						case c.hub.commandCh <- hubCmd{op: "wsAddSource", label: label, addr: addr}:
						default:
						}
					}
				case "removeSource":
					id, _ := env["id"].(string)
					if id != "" {
						select {
						case c.hub.commandCh <- hubCmd{op: "wsRemoveSource", sourceID: id}:
						default:
						}
					}
				case "saveSources":
					select {
					case c.hub.commandCh <- hubCmd{op: "wsSaveSources"}:
					default:
					}
				}
			}
		}
		c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
	}
}

// ─── Hub ─────────────────────────────────────────────────────────────────────

var upgrader = websocket.Upgrader{
	ReadBufferSize:  4096,
	WriteBufferSize: 64 * 1024,
	CheckOrigin:     func(r *http.Request) bool { return true },
}

// sourceHubState holds all data for one active data source.
// Only accessed from the Run() goroutine.
type sourceHubState struct {
	id, label, addr, connState string
	signals                    []SignalInfo
	configJS                   []byte

	// Time-signal calibration — only accessed from Run() goroutine.
	timeSigCalib     map[string]float64
	configSeq        uint64
	configSeqAtCalib uint64
}

// taggedSample is a DataSample annotated with its source ID.
type taggedSample struct {
	sourceID string
	sample   DataSample
}

// hubCmd carries a command to the Run() goroutine.
type hubCmd struct {
	op       string // "addSource","removeSource","setSourceState","updateConfig",
	//                "wsAddSource","wsRemoveSource","wsSaveSources"
	sourceID string
	label    string
	addr     string
	state    string
	sigs     []SignalInfo
}

// Hub is the central broker between UDP clients and WebSocket clients.
// All map state is accessed exclusively from the Run() goroutine, except
// ringsMu/rings which are also read by HTTP handler goroutines.
type Hub struct {
	clients     map[*wsClient]bool
	register    chan *wsClient
	unregister  chan *wsClient
	broadcastCh chan []byte
	dataCh      chan taggedSample
	commandCh   chan hubCmd

	sm *SourceManager // set after construction; used for WS-initiated source changes

	// Ring buffers for hi-res zoom data.
	// ringsMu protects the map structure; each sigRing has its own RWMutex for data.
	ringsMu sync.RWMutex
	rings   map[string]*sigRing // "sourceId:signalKey" → ring

	// lastZoomAt tracks the last time a zoom request was served.
	// Ring buffer writes are skipped when no zoom has been requested
	// in the last 10 s, saving substantial CPU on LTTB + ring writes.
	lastZoomAt time.Time
	zoomAtMu   sync.Mutex

	statsMu  sync.RWMutex
	statsMap map[string]*SourceStat
}

// NewHub creates an initialised Hub.
func NewHub() *Hub {
	return &Hub{
		clients:     make(map[*wsClient]bool),
		register:    make(chan *wsClient, 8),
		unregister:  make(chan *wsClient, 8),
		broadcastCh: make(chan []byte, 256),
		dataCh:      make(chan taggedSample, 65536), // large buffer: absorbs bursts at high sample rates
		commandCh:   make(chan hubCmd, 64),
		rings:       make(map[string]*sigRing),
		statsMap:    make(map[string]*SourceStat),
	}
}

// getRing returns the ring buffer for a fully-prefixed signal key, or nil.
func (h *Hub) getRing(key string) *sigRing {
	h.ringsMu.RLock()
	rb := h.rings[key]
	h.ringsMu.RUnlock()
	return rb
}

// shouldWriteRing returns true if zoom was requested within the last 10 seconds.
// When false, the hot path can skip LTTB decimation for the ring buffer entirely.
func (h *Hub) shouldWriteRing() bool {
	h.zoomAtMu.Lock()
	ok := time.Since(h.lastZoomAt) < 10*time.Second
	h.zoomAtMu.Unlock()
	return ok
}

// HandleZoom serves GET /api/zoom?... It also records the access time
// so the ring buffer knows zoom is active and worth populating.
func (h *Hub) HandleZoom(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	t0, err0 := strconv.ParseFloat(q.Get("t0"), 64)
	t1, err1 := strconv.ParseFloat(q.Get("t1"), 64)
	if err0 != nil || err1 != nil || t1 <= t0 {
		http.Error(w, "invalid t0/t1", http.StatusBadRequest)
		return
	}
	// n=0 (or explicit "0") means no LTTB decimation — return all ring data in range.
	// n omitted / invalid → default 2400 (display quality).
	var n int
	if nStr := q.Get("n"); nStr == "" {
		n = 2400
	} else {
		n, _ = strconv.Atoi(nStr)
		if n <= 0 {
			n = 1 << 30 // no decimation
		} else if n < 10 {
			n = 2400
		}
	}

	// Record zoom access time so ring writes stay active.
	// Skip n=0 requests (full-data exports / trigger snapshots) — only
	// interactive zooms should keep the ring buffer populated.
	if n > 0 {
		h.zoomAtMu.Lock()
		h.lastZoomAt = time.Now()
		h.zoomAtMu.Unlock()
	}

	keys := strings.Split(q.Get("signals"), ",")

	// Collect ring references under a brief RLock.
	h.ringsMu.RLock()
	refs := make(map[string]*sigRing, len(keys))
	for _, k := range keys {
		k = strings.TrimSpace(k)
		if k == "" {
			continue
		}
		if rb, ok := h.rings[k]; ok {
			refs[k] = rb
		}
	}
	h.ringsMu.RUnlock()

	result := make(map[string]sigData, len(refs))
	for k, rb := range refs {
		rt, rv := rb.slice(t0, t1)
		if len(rt) == 0 {
			continue
		}
		dt, dv := lttbDecimate(rt, rv, n)
		result[k] = sigData{T: dt, V: dv}
	}

	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(map[string]any{
		"type":    "zoom",
		"signals": result,
	}); err != nil {
		log.Printf("hub: zoom encode: %v", err)
	}
}

// AddSource notifies the Hub that a new source has been registered.
func (h *Hub) AddSource(id, label, addr string) {
	select {
	case h.commandCh <- hubCmd{op: "addSource", sourceID: id, label: label, addr: addr}:
	default:
	}
}

// RemoveSource notifies the Hub that a source has been removed.
func (h *Hub) RemoveSource(id string) {
	select {
	case h.commandCh <- hubCmd{op: "removeSource", sourceID: id}:
	default:
	}
}

// SetSourceState updates the connection state of a source.
func (h *Hub) SetSourceState(id, state string) {
	select {
	case h.commandCh <- hubCmd{op: "setSourceState", sourceID: id, state: state}:
	default:
	}
}

// UpdateConfigForSource stores a new signal config for a source and broadcasts it.
func (h *Hub) UpdateConfigForSource(sourceID string, sigs []SignalInfo) {
	select {
	case h.commandCh <- hubCmd{op: "updateConfig", sourceID: sourceID, sigs: sigs}:
	default:
	}
}

// PushDataForSource enqueues a data sample from a specific source.
func (h *Hub) PushDataForSource(sourceID string, s DataSample) {
	select {
	case h.dataCh <- taggedSample{sourceID: sourceID, sample: s}:
	default:
	}
}

// broadcast enqueues a message for delivery to all WebSocket clients.
func (h *Hub) broadcast(msg []byte) {
	select {
	case h.broadcastCh <- msg:
	default:
	}
}

// HandleWebSocket upgrades an HTTP request to a WebSocket connection.
func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("ws upgrade: %v", err)
		return
	}
	c := &wsClient{hub: h, conn: conn, send: make(chan wsMessage, 64)}
	h.register <- c
	go c.writePump()
	go c.readPump()
}

// buildSourcesMsg serialises the current source list as a JSON "sources" message.
func buildSourcesMsg(sm map[string]*sourceHubState) []byte {
	type srcInfo struct {
		ID    string `json:"id"`
		Label string `json:"label"`
		Addr  string `json:"addr"`
		State string `json:"state"`
	}
	list := make([]srcInfo, 0, len(sm))
	for _, src := range sm {
		list = append(list, srcInfo{ID: src.id, Label: src.label, Addr: src.addr, State: src.connState})
	}
	msg, _ := json.Marshal(map[string]interface{}{"type": "sources", "sources": list})
	return msg
}

// Run is the hub's main goroutine. Must be started with go hub.Run().
func (h *Hub) Run() {
	ticker := time.NewTicker(time.Second / 30)
	defer ticker.Stop()

	statsTicker := time.NewTicker(time.Second)
	defer statsTicker.Stop()

	sourcesMap := make(map[string]*sourceHubState)
	var sourcesMsg []byte

	// pending[sourceID] accumulates samples between 30 Hz ticks.
	pending := make(map[string][]DataSample)

	rebuildSources := func() {
		sourcesMsg = buildSourcesMsg(sourcesMap)
		h.broadcast(sourcesMsg)
	}

	for {
		select {
		case c := <-h.register:
			h.clients[c] = true
			// Send current state to the new client.
			if sourcesMsg != nil {
				select { case c.send <- wsMessage{websocket.TextMessage, sourcesMsg}: default: }
			}
			for _, src := range sourcesMap {
				if src.configJS != nil {
					select { case c.send <- wsMessage{websocket.TextMessage, src.configJS}: default: }
				}
			}

		case c := <-h.unregister:
			if _, ok := h.clients[c]; ok {
				delete(h.clients, c)
				close(c.send)
			}

		case msg := <-h.broadcastCh:
			for c := range h.clients {
				select { case c.send <- wsMessage{websocket.TextMessage, msg}: default: }
			}

		case cmd := <-h.commandCh:
			switch cmd.op {
			case "addSource":
				sourcesMap[cmd.sourceID] = &sourceHubState{
					id:           cmd.sourceID,
					label:        cmd.label,
					addr:         cmd.addr,
					connState:    "connecting",
					timeSigCalib: make(map[string]float64),
				}
				h.statsMu.Lock()
				h.statsMap[cmd.sourceID] = &SourceStat{}
				h.statsMu.Unlock()
				rebuildSources()

			case "removeSource":
				delete(sourcesMap, cmd.sourceID)
				delete(pending, cmd.sourceID)
				pfxDel := cmd.sourceID + ":"
				h.ringsMu.Lock()
				for k := range h.rings {
					if strings.HasPrefix(k, pfxDel) {
						delete(h.rings, k)
					}
				}
				h.ringsMu.Unlock()
				h.statsMu.Lock()
				delete(h.statsMap, cmd.sourceID)
				h.statsMu.Unlock()
				rebuildSources()

			case "setSourceState":
				if src, ok := sourcesMap[cmd.sourceID]; ok {
					src.connState = cmd.state
					rebuildSources()
				}

			case "updateConfig":
				src, ok := sourcesMap[cmd.sourceID]
				if !ok {
					continue
				}
				src.signals = cmd.sigs
				src.configSeq++
				cfgMsg, err := json.Marshal(map[string]any{
					"type":     "config",
					"sourceId": cmd.sourceID,
					"signals":  cmd.sigs,
				})
				if err != nil {
					log.Printf("hub: marshal config: %v", err)
					continue
				}
				src.configJS = cfgMsg
				h.broadcast(cfgMsg)
				// Rebuild ring buffers for this source.
				pfxUpd := cmd.sourceID + ":"
				h.ringsMu.Lock()
				for k := range h.rings {
					if strings.HasPrefix(k, pfxUpd) {
						delete(h.rings, k)
					}
				}
				for _, sig := range cmd.sigs {
					ne := sig.NumElements()
					isTemporal := ne > 1 && sig.TimeMode != TimeModePacket
					if isTemporal {
						h.rings[pfxUpd+sig.Name] = newSigRing(ringCapTemporal)
					} else if ne == 1 {
						h.rings[pfxUpd+sig.Name] = newSigRing(ringCapScalar)
					} else {
						for i := 0; i < ne; i++ {
							h.rings[pfxUpd+arrayKey(sig.Name, i)] = newSigRing(ringCapScalar)
						}
					}
				}
				h.ringsMu.Unlock()

			case "wsAddSource":
				if h.sm != nil {
					go func(label, addr string) { h.sm.Add(label, addr) }(cmd.label, cmd.addr)
				}

			case "wsRemoveSource":
				if h.sm != nil {
					go func(id string) { h.sm.Remove(id) }(cmd.sourceID)
				}

			case "wsSaveSources":
				if h.sm != nil {
					if err := h.sm.Save(); err != nil {
						log.Printf("hub: save sources: %v", err)
					}
				}
			}

		case ts := <-h.dataCh:
			pending[ts.sourceID] = append(pending[ts.sourceID], ts.sample)

		case <-ticker.C:
			for srcID, samples := range pending {
				if len(samples) == 0 {
					continue
				}
				src, ok := sourcesMap[srcID]
				if !ok || len(src.signals) == 0 || len(h.clients) == 0 {
					pending[srcID] = pending[srcID][:0]
					continue
				}
				msg := h.buildBinaryDataMessageForSource(src, samples)
				pending[srcID] = pending[srcID][:0]
				if msg != nil {
					for c := range h.clients {
						select {
						case c.send <- wsMessage{websocket.BinaryMessage, msg}:
						default:
						}
					}
				}
			}

		case <-statsTicker.C:
			h.statsMu.RLock()
			snap := make(map[string]StatInfo, len(h.statsMap))
			for id, st := range h.statsMap {
				snap[id] = st.Snapshot()
			}
			h.statsMu.RUnlock()
			if len(snap) > 0 {
				msg, _ := json.Marshal(map[string]any{"type": "stats", "sources": snap})
				h.broadcast(msg)
			}
		}
	}
}

// float64ToBytes reinterprets a []float64 as []byte without copying.
// The caller must ensure the slice is not modified while the byte view is in use.
func float64ToBytes(f []float64) []byte {
	if len(f) == 0 {
		return nil
	}
	return unsafe.Slice((*byte)(unsafe.Pointer(&f[0])), len(f)*8)
}

// writeFloat64s encodes a []float64 as little-endian bytes into buf at offset
// and returns the new offset. Uses unsafe to avoid per-element PutUint64 calls.
func writeFloat64s(buf []byte, off int, f []float64) int {
	copy(buf[off:], float64ToBytes(f))
	return off + len(f)*8
}

// ─── Data serialisation ───────────────────────────────────────────────────────

// maxPushPoints is the LTTB target for data pushed over WebSocket per 30 Hz tick.
// With cascaded LTTB the server selects the most visually significant pts from
// each batch; the browser accumulates ticks × pts/tick for the rolling window.
// For a 1 200 px plot showing a 5 s window at 30 Hz: 2×1200/(5×30) ≈ 16 pts/tick
// needed. 50 gives 4× headroom: 50×30×5 = 7 500 pts → trivial 120 KB buffer.
// Zoom resolution is unaffected — the ring buffer (maxRingPoints) serves /api/zoom.
const maxPushPoints = 50

// maxRingPoints is the LTTB target written into the ring buffer per tick.
// At 5 Msps / 5 kHz packet rate ≈ 167 k raw samples/tick → LTTB to 20 k →
// min Δt ≈ 33 ms / 20 k ≈ 1.65 µs, sufficient for sub-10 µs zoom resolution.
const maxRingPoints = 20_000

// ringCapTemporal is the ring buffer capacity for temporal-array signals.
// At 20 k pts/tick × 30 Hz = 600 k pts/s → 6 M cap gives ~10 s of hi-res
// history — the same temporal coverage as the frontend push buffer.
const ringCapTemporal = 6_000_000

// ringCapScalar is the ring buffer capacity for scalar / spatial-array signals.
// At ≤10 kHz → ~333 pts/tick × 30 Hz ≈ 10 k pts/s → ~10 s of history.
const ringCapScalar = 100_000

// lttbDecimate reduces (tIn, vIn) to at most threshold representative points
// using the Largest-Triangle-Three-Buckets algorithm.
// Returns the original slices unchanged when len(tIn) ≤ threshold.
func lttbDecimate(tIn, vIn []float64, threshold int) ([]float64, []float64) {
	n := len(tIn)
	if n <= threshold || threshold < 3 {
		return tIn, vIn
	}
	outT := make([]float64, threshold)
	outV := make([]float64, threshold)
	outT[0], outV[0] = tIn[0], vIn[0]
	outT[threshold-1], outV[threshold-1] = tIn[n-1], vIn[n-1]

	every := float64(n-2) / float64(threshold-2)
	a := 0
	for i := 0; i < threshold-2; i++ {
		// Centroid of the next bucket
		avgS := int(float64(i+1)*every) + 1
		avgE := int(float64(i+2)*every) + 1
		if avgE > n {
			avgE = n
		}
		avgT, avgV, cnt := 0.0, 0.0, 0
		for j := avgS; j < avgE; j++ {
			avgT += tIn[j]; avgV += vIn[j]; cnt++
		}
		if cnt > 0 {
			avgT /= float64(cnt); avgV /= float64(cnt)
		}
		// Pick the point in the current bucket that forms the largest triangle
		rS := int(float64(i)*every) + 1
		rE := int(float64(i+1)*every) + 1
		if rE > n {
			rE = n
		}
		maxArea, next := -1.0, rS
		aT, aV := tIn[a], vIn[a]
		for j := rS; j < rE; j++ {
			area := math.Abs((aT-avgT)*(vIn[j]-aV) - (aT-tIn[j])*(avgV-aV))
			if area > maxArea {
				maxArea = area; next = j
			}
		}
		outT[i+1], outV[i+1] = tIn[next], vIn[next]
		a = next
	}
	return outT, outV
}

// sigData carries one signal's worth of time+value pairs.
type sigData struct {
	T []float64 `json:"t"`
	V []float64 `json:"v"`
}

// dataMsg is the JSON envelope sent to WebSocket clients.
type dataMsg struct {
	Type     string             `json:"type"`
	SourceID string             `json:"sourceId"`
	Signals  map[string]sigData `json:"signals"`
}

// buildDataMessageForSource serialises a batch of samples for one source.
// Signal keys in the output are prefixed with "sourceId:" so the browser can
// store them in a single flat buffer map without collision.
func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample) []byte {
	if len(batch) == 0 {
		return nil
	}

	// Reset time-signal calibration whenever the signal config changed.
	if src.configSeq != src.configSeqAtCalib {
		src.configSeqAtCalib = src.configSeq
		src.timeSigCalib = make(map[string]float64)
	}

	sigs := src.signals
	pfx := src.id + ":"
	out := make(map[string]sigData, len(sigs)*2)

	for _, sig := range sigs {
		n := sig.NumElements()

		switch {
		case n > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample):
			hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
			var timeSigName string
			// uint64 signals carry nanoseconds (HRT); everything else is assumed microseconds.
			timerToSec := 1e-6
			if hasTimeSig {
				ts := sigs[sig.TimeSignalIdx]
				timeSigName = ts.Name
				if ts.TypeCode == 6 { // uint64 → nanoseconds
					timerToSec = 1e-9
				}
			}
			dt := 0.0
			if sig.SamplingRate > 0 {
				dt = 1.0 / sig.SamplingRate
			}
			allT := make([]float64, 0, len(batch)*n)
			allV := make([]float64, 0, len(batch)*n)

			for _, s := range batch {
				vals, ok := s.Values[sig.Name]
				if !ok || len(vals) < n {
					continue
				}
				var anchorTime float64
				anchorIsFirstSample := sig.TimeMode == TimeModeFirstSample
				if hasTimeSig {
					tVals, tOk := s.Values[timeSigName]
					if tOk && len(tVals) >= 1 {
						timerS := tVals[0] * timerToSec
						wallT := float64(s.WallTime.UnixNano()) / 1e9
						if _, exists := src.timeSigCalib[timeSigName]; !exists {
							src.timeSigCalib[timeSigName] = wallT - timerS
						}
						anchorTime = src.timeSigCalib[timeSigName] + timerS
					} else {
						anchorTime = float64(s.WallTime.UnixNano()) / 1e9
						anchorIsFirstSample = false
					}
				} else {
					anchorTime = float64(s.WallTime.UnixNano()) / 1e9
					anchorIsFirstSample = false
				}
				for k := 0; k < n; k++ {
					var t float64
					if anchorIsFirstSample {
						t = anchorTime + float64(k)*dt
					} else {
						t = anchorTime - float64(n-1-k)*dt
					}
					allT = append(allT, t)
					allV = append(allV, vals[k])
				}
			}

			// Write hi-res LTTB data to ring for on-demand zoom queries.
			ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
			if rb := h.getRing(pfx + sig.Name); rb != nil {
				rb.write(ringT, ringV)
			}
			// Decimate further for WebSocket push (rolling window).
			decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
			out[pfx+sig.Name] = sigData{T: decimT, V: decimV}

		case n > 1 && sig.TimeMode == TimeModeFullArray:
			// The time signal has the same N elements as the data signal.
			// Each element pair (timeSig[k], dataSig[k]) is one (t, v) sample.
			hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
			var timeSigName string
			timerToSec := 1e-6
			if hasTimeSig {
				ts := sigs[sig.TimeSignalIdx]
				timeSigName = ts.Name
				if ts.TypeCode == 6 { // uint64 → nanoseconds
					timerToSec = 1e-9
				}
			}
			allT := make([]float64, 0, len(batch)*n)
			allV := make([]float64, 0, len(batch)*n)

			for _, s := range batch {
				vals, ok := s.Values[sig.Name]
				if !ok || len(vals) < n {
					continue
				}
				if hasTimeSig {
					tVals, tOk := s.Values[timeSigName]
					if tOk && len(tVals) >= n {
						// Calibrate once: map timer ticks to wall-clock seconds.
						if _, exists := src.timeSigCalib[timeSigName]; !exists {
							wallT := float64(s.WallTime.UnixNano()) / 1e9
							src.timeSigCalib[timeSigName] = wallT - tVals[0]*timerToSec
						}
						calib := src.timeSigCalib[timeSigName]
						for k := 0; k < n; k++ {
							allT = append(allT, calib+tVals[k]*timerToSec)
							allV = append(allV, vals[k])
						}
						continue
					}
				}
				// Fallback: stamp all elements with packet arrival time.
				wallT := float64(s.WallTime.UnixNano()) / 1e9
				for k := 0; k < n; k++ {
					allT = append(allT, wallT)
					allV = append(allV, vals[k])
				}
			}

			ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
			if rb := h.getRing(pfx + sig.Name); rb != nil {
				rb.write(ringT, ringV)
			}
			decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
			out[pfx+sig.Name] = sigData{T: decimT, V: decimV}

		case n == 1:
			ts := make([]float64, 0, len(batch))
			vs := make([]float64, 0, len(batch))
			for _, s := range batch {
				vals, ok := s.Values[sig.Name]
				if !ok || len(vals) < 1 {
					continue
				}
				ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
				vs = append(vs, vals[0])
			}
			if rb := h.getRing(pfx + sig.Name); rb != nil {
				rb.write(ts, vs)
			}
			out[pfx+sig.Name] = sigData{T: ts, V: vs}

		default:
			// Spatial / PacketTime array: one stream per element.
			for i := 0; i < n; i++ {
				key := pfx + arrayKey(sig.Name, i)
				ts := make([]float64, 0, len(batch))
				vs := make([]float64, 0, len(batch))
				for _, s := range batch {
					vals, ok := s.Values[sig.Name]
					if !ok || len(vals) <= i {
						continue
					}
					ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
					vs = append(vs, vals[i])
				}
				if rb := h.getRing(key); rb != nil {
					rb.write(ts, vs)
				}
				out[key] = sigData{T: ts, V: vs}
			}
		}
	}

	result, err := json.Marshal(dataMsg{Type: "data", SourceID: src.id, Signals: out})
	if err != nil {
		log.Printf("hub: marshal data: %v", err)
		return nil
	}
	return result
}

// buildBinaryDataMessageForSource encodes a batch of samples as a compact binary
// frame for WebSocket binary messages. Skips the JSON overhead entirely.
//
// Wire format (little-endian):
//
//	uint8  version (1)
//	uint8  source ID length
//	UTF-8  source ID
//	uint32 number of signals
//	for each signal:
//	  uint16 key length
//	  UTF-8  key (relative to source, e.g. "sigName" not "s1:sigName")
//	  uint32 pair count N
//	  float64[N]  t values
//	  float64[N]  v values
func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataSample) []byte {
	if len(batch) == 0 {
		return nil
	}
	if src.configSeq != src.configSeqAtCalib {
		src.configSeqAtCalib = src.configSeq
		src.timeSigCalib = make(map[string]float64)
	}

	sigs := src.signals
	pfx := src.id + ":"
	writeRing := h.shouldWriteRing()

	// ---- Phase 1: collect (t,v) for each signal (same logic as JSON path) ----

	type pairBuf struct {
		t, v []float64
	}
	pairs := make(map[string]pairBuf, len(sigs)*2)

	for _, sig := range sigs {
		n := sig.NumElements()

		switch {
		case n > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample):
			hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
			var timeSigName string
			timerToSec := 1e-6
			if hasTimeSig {
				ts := sigs[sig.TimeSignalIdx]
				timeSigName = ts.Name
				if ts.TypeCode == 6 {
					timerToSec = 1e-9
				}
			}
			dt := 0.0
			if sig.SamplingRate > 0 {
				dt = 1.0 / sig.SamplingRate
			}
			allT := make([]float64, 0, len(batch)*n)
			allV := make([]float64, 0, len(batch)*n)
			for _, s := range batch {
				vals, ok := s.Values[sig.Name]
				if !ok || len(vals) < n {
					continue
				}
				var anchorTime float64
				anchorIsFirstSample := sig.TimeMode == TimeModeFirstSample
				if hasTimeSig {
					tVals, tOk := s.Values[timeSigName]
					if tOk && len(tVals) >= 1 {
						timerS := tVals[0] * timerToSec
						wallT := float64(s.WallTime.UnixNano()) / 1e9
						if _, exists := src.timeSigCalib[timeSigName]; !exists {
							src.timeSigCalib[timeSigName] = wallT - timerS
						}
						anchorTime = src.timeSigCalib[timeSigName] + timerS
					} else {
						anchorTime = float64(s.WallTime.UnixNano()) / 1e9
						anchorIsFirstSample = false
					}
				} else {
					anchorTime = float64(s.WallTime.UnixNano()) / 1e9
					anchorIsFirstSample = false
				}
				for k := 0; k < n; k++ {
					var t float64
					if anchorIsFirstSample {
						t = anchorTime + float64(k)*dt
					} else {
						t = anchorTime - float64(n-1-k)*dt
					}
					allT = append(allT, t)
					allV = append(allV, vals[k])
				}
			}
			// Write hi-res LTTB data to ring (only if zoom is active).
			if writeRing {
				ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
				if rb := h.getRing(pfx + sig.Name); rb != nil {
					rb.write(ringT, ringV)
				}
			}
			// Decimate for push.
			decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
			pairs[sig.Name] = pairBuf{t: decimT, v: decimV}

		case n > 1 && sig.TimeMode == TimeModeFullArray:
			hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
			var timeSigName string
			timerToSec := 1e-6
			if hasTimeSig {
				ts := sigs[sig.TimeSignalIdx]
				timeSigName = ts.Name
				if ts.TypeCode == 6 {
					timerToSec = 1e-9
				}
			}
			allT := make([]float64, 0, len(batch)*n)
			allV := make([]float64, 0, len(batch)*n)
			for _, s := range batch {
				vals, ok := s.Values[sig.Name]
				if !ok || len(vals) < n {
					continue
				}
				if hasTimeSig {
					tVals, tOk := s.Values[timeSigName]
					if tOk && len(tVals) >= n {
						if _, exists := src.timeSigCalib[timeSigName]; !exists {
							wallT := float64(s.WallTime.UnixNano()) / 1e9
							src.timeSigCalib[timeSigName] = wallT - tVals[0]*timerToSec
						}
						calib := src.timeSigCalib[timeSigName]
						for k := 0; k < n; k++ {
							allT = append(allT, calib+tVals[k]*timerToSec)
							allV = append(allV, vals[k])
						}
						continue
					}
				}
				wallT := float64(s.WallTime.UnixNano()) / 1e9
				for k := 0; k < n; k++ {
					allT = append(allT, wallT)
					allV = append(allV, vals[k])
				}
			}
			ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
			if rb := h.getRing(pfx + sig.Name); rb != nil {
				rb.write(ringT, ringV)
			}
			decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
			pairs[sig.Name] = pairBuf{t: decimT, v: decimV}

		case n == 1:
			ts := make([]float64, 0, len(batch))
			vs := make([]float64, 0, len(batch))
			for _, s := range batch {
				vals, ok := s.Values[sig.Name]
				if !ok || len(vals) < 1 {
					continue
				}
				ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
				vs = append(vs, vals[0])
			}
			if writeRing {
				if rb := h.getRing(pfx + sig.Name); rb != nil {
					rb.write(ts, vs)
				}
			}
			pairs[sig.Name] = pairBuf{t: ts, v: vs}

		default:
			for i := 0; i < n; i++ {
				key := arrayKey(sig.Name, i)
				ts := make([]float64, 0, len(batch))
				vs := make([]float64, 0, len(batch))
				for _, s := range batch {
					vals, ok := s.Values[sig.Name]
					if !ok || len(vals) <= i {
						continue
					}
					ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
					vs = append(vs, vals[i])
				}
				if writeRing {
					if rb := h.getRing(pfx + key); rb != nil {
						rb.write(ts, vs)
					}
				}
				pairs[key] = pairBuf{t: ts, v: vs}
			}
		}
	}

	// ---- Phase 2: compute total size and serialize ----
	totalSize := 1 + 1 + len(src.id) + 4 // version + srcIdLen + srcId + numSigs
	for key, p := range pairs {
		totalSize += 2 + len(key) + 4          // keyLen + key + pairCount
		totalSize += len(p.t)*8 + len(p.v)*8  // t + v float64 data
	}

	buf := make([]byte, totalSize)
	buf[0] = 1 // version
	buf[1] = byte(len(src.id))
	copy(buf[2:], src.id)
	off := 2 + len(src.id)
	binary.LittleEndian.PutUint32(buf[off:], uint32(len(pairs)))
	off += 4

	for key, p := range pairs {
		binary.LittleEndian.PutUint16(buf[off:], uint16(len(key)))
		off += 2
		copy(buf[off:], key)
		off += len(key)
		binary.LittleEndian.PutUint32(buf[off:], uint32(len(p.t)))
		off += 4
		off = writeFloat64s(buf, off, p.t)
		off = writeFloat64s(buf, off, p.v)
	}

	return buf
}

// RecordDataFragment is called by UDPClient for every incoming DATA datagram.
func (h *Hub) RecordDataFragment(sourceID string, counter uint32, nBytes int, arrivalNs int64, complete bool) {
	h.statsMu.RLock()
	st := h.statsMap[sourceID]
	h.statsMu.RUnlock()
	if st != nil {
		st.RecordFragment(counter, nBytes, arrivalNs, complete)
	}
}

// arrayKey returns the buffer key for element i of an array signal.
func arrayKey(name string, i int) string {
	return name + "[" + itoa(i) + "]"
}

func itoa(n int) string {
	if n == 0 {
		return "0"
	}
	buf := [20]byte{}
	pos := len(buf)
	for n > 0 {
		pos--
		buf[pos] = byte('0' + n%10)
		n /= 10
	}
	return string(buf[pos:])
}