MARTe-Integrated-Components/Common/Client/go/wshub/history.go

package wshub

// Disk-backed circular history storage, wire- and file-compatible with the C++
// StreamHub's HistoryWriter. Long live windows (up to 600 s) hold far more
// samples than the in-memory zoom rings can, so the samples go to a
// pre-allocated circular file per signal and are read back on demand.
//
// File layout (per signal, little-endian):
//
//	Offset  Size  Field
//	0       4     Magic "SHR1"
//	4       4     uint32 version (1)
//	8       4     uint32 capacity (max pairs)
//	12      4     uint32 head (next write position, wraps)
//	16      4     uint32 count (valid entries, <= capacity)
//	20      4     uint32 decimation
//	24      8     float64 tOldest
//	32      8     float64 tNewest
//	40      24    reserved (pad to 64)
//	64      ...   data: capacity × 16 bytes (float64 time, float64 value)

import (
	"encoding/binary"
	"encoding/json"
	"fmt"
	"log"
	"math"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"

	"marte2/common/udpsprotocol"
)

const (
	histHeaderSize = 64
	histPairSize   = 16
	histVersion    = 1
	// histMinCapacity floors the per-signal file so a slow or unknown-rate
	// signal still keeps a usable amount of history.
	histMinCapacity = 1000
	// histMaxCapacity caps one signal's file at 16 GB of pairs, so a bogus
	// sampling rate in a CONFIG packet cannot ask for an unbounded file.
	histMaxCapacity = 1 << 30
	// histDefaultMaxPoints is the default per-signal file budget, in stored
	// points. 16 Mi points is 256 MB of (time, value) pairs.
	histDefaultMaxPoints = 16 << 20
	// histTypeCodeUint64 is the UDPS type code for uint64; those signals are
	// time references (TimeArray), not data, so they are not archived.
	histTypeCodeUint64 = 6
)

// histHeadroom is how much more than the window a file is sized to hold. The
// window has to still be on disk when it is read back, and a capture is read
// back a whole window after its first sample was written — a file sized to
// exactly the window has overwritten that sample by then.
const histHeadroom = 1.25

var histMagic = [4]byte{'S', 'H', 'R', '1'}

// HistoryConfig configures the disk-backed history. A zero Directory disables
// history entirely.
type HistoryConfig struct {
	Directory string
	// WindowSec is the timespan one signal's file has to hold — the live or
	// trigger window the clients are displaying, not a fixed retention period.
	// The hub keeps it up to date as clients change what they display, so this
	// field is only the value used until the first client says otherwise.
	WindowSec        float64 // default 10 s, the assumed live window
	Decimation       int     // keep every Nth sample; default 1
	FlushIntervalSec int     // header flush period; default 5
	// MinDiskFreeMB pauses writing below this much free space; default 500.
	// A negative value disables the check. Unlike the C++ HistoryWriter, zero
	// cannot mean "disabled" here: it is also the unset value of the field, and
	// silently dropping the guard for a caller who just did not fill it in is
	// the worse failure.
	MinDiskFreeMB int
	// MaxPointsPerSignal caps one signal's file in stored points (16 bytes
	// each); default 16 Mi = 256 MB. The window alone cannot bound the size,
	// because the file is sized from the sample rate: 600 s of a 1 MSps signal
	// is 9.6 GB. The budget is spent on resolution, not on span — a signal too
	// fast to archive whole is stored as a min/max envelope wide enough to fit,
	// so the window is always covered. This is the knob the web UI exposes, in
	// MPts per signal.
	MaxPointsPerSignal int
}

func (c HistoryConfig) withDefaults() HistoryConfig {
	if c.WindowSec <= 0 {
		c.WindowSec = defaultLiveWindowSec
	}
	if c.Decimation <= 0 {
		c.Decimation = 1
	}
	if c.FlushIntervalSec <= 0 {
		c.FlushIntervalSec = 5
	}
	if c.MinDiskFreeMB == 0 {
		c.MinDiskFreeMB = 500
	}
	if c.MaxPointsPerSignal <= 0 {
		c.MaxPointsPerSignal = histDefaultMaxPoints
	}
	return c
}

// histFile is one signal's circular file. Writes come from the hub's Run()
// goroutine, reads from the WebSocket read goroutines, hence the RWMutex.
type histFile struct {
	mu       sync.RWMutex
	f        *os.File
	capacity uint32
	head     uint32
	count    uint32
	tOldest  float64
	tNewest  float64

	decimCount int
	dirty      bool

	// bucket is how many incoming samples collapse into one min/max pair; 1
	// archives every sample verbatim. bMin/bMax accumulate the bucket in
	// progress, which spans as many write() batches as it takes to fill.
	bucket int
	bCount int
	bMinT  float64
	bMinV  float64
	bMaxT  float64
	bMaxV  float64

	// reduction is the header's decimation field: how many source samples one
	// bucket consumes, so a reader knows what the stored resolution is.
	reduction uint32

	// rateHz is the sampling rate the file was sized from, kept so a later
	// budget change can re-size it without waiting to measure the rate again.
	rateHz float64

	sourceID string
	signal   string
}

// foldBucket accumulates a batch into the min/max bucket in progress and returns
// the pairs of whichever buckets completed, in time order. Called with hf.mu
// held by the caller's write path.
func (hf *histFile) foldBucket(t, v []float64) ([]float64, []float64) {
	outT := make([]float64, 0, 2*(len(t)/hf.bucket+1))
	outV := make([]float64, 0, cap(outT))
	for i := range t {
		if hf.bCount == 0 {
			hf.bMinT, hf.bMinV = t[i], v[i]
			hf.bMaxT, hf.bMaxV = t[i], v[i]
		} else {
			if v[i] < hf.bMinV {
				hf.bMinT, hf.bMinV = t[i], v[i]
			}
			if v[i] > hf.bMaxV {
				hf.bMaxT, hf.bMaxV = t[i], v[i]
			}
		}
		hf.bCount++
		if hf.bCount < hf.bucket {
			continue
		}
		hf.bCount = 0
		switch {
		case hf.bMinT == hf.bMaxT:
			// A bucket whose samples are all equal has one extreme, not two.
			outT = append(outT, hf.bMinT)
			outV = append(outV, hf.bMinV)
		case hf.bMinT < hf.bMaxT:
			outT = append(outT, hf.bMinT, hf.bMaxT)
			outV = append(outV, hf.bMinV, hf.bMaxV)
		default:
			outT = append(outT, hf.bMaxT, hf.bMinT)
			outV = append(outV, hf.bMaxV, hf.bMinV)
		}
	}
	return outT, outV
}

// HistorySignalInfo is the per-signal metadata reported in a historyInfo event.
type HistorySignalInfo struct {
	T0       float64 `json:"t0"`
	T1       float64 `json:"t1"`
	Count    uint32  `json:"count"`
	Capacity uint32  `json:"capacity"`
	// Bucket is how many source samples collapse into one min/max pair; 1 means
	// the signal is archived verbatim. Clients show it so the budget control has
	// a visible effect to tune against.
	Bucket int `json:"bucket"`
}

// pendingHistSignal is a signal whose producer declared no sampling rate, so
// its file cannot be sized until the rate has been measured on the live stream.
type pendingHistSignal struct {
	sourceID string
	signal   string
}

// historyWriter owns every open signal file.
type historyWriter struct {
	cfg HistoryConfig

	mu      sync.RWMutex
	files   map[string]*histFile         // "sourceId:signalName" → file
	pending map[string]pendingHistSignal // awaiting a measured rate
	// windowSec is the timespan the files are sized to hold, kept in step with
	// what the clients display by the hub's retune sweep.
	windowSec float64

	diskMu  sync.RWMutex
	diskLow bool

	// capMu guards captures, the per-signal copies of the last trigger window.
	capMu    sync.RWMutex
	captures map[string]*histFile
}

// newHistoryWriter opens (creating it if needed) the history directory. It
// returns nil when history is not configured, which every call site treats as
// "history disabled" so the hub runs unchanged without it.
func newHistoryWriter(cfg HistoryConfig) (*historyWriter, error) {
	if strings.TrimSpace(cfg.Directory) == "" {
		return nil, nil
	}
	cfg = cfg.withDefaults()
	cfg.Directory = strings.TrimRight(cfg.Directory, "/")
	if err := os.MkdirAll(cfg.Directory, 0o755); err != nil {
		return nil, fmt.Errorf("history directory %q: %w", cfg.Directory, err)
	}
	return &historyWriter{
		cfg:       cfg,
		windowSec: cfg.WindowSec,
		files:     make(map[string]*histFile),
		pending:   make(map[string]pendingHistSignal),
	}, nil
}

func (hw *historyWriter) enabled() bool { return hw != nil }

// onSourceConfigured opens a file per data signal of a source, sizing it from
// the declared sampling rate. Signals already archived under the same key keep
// their file, so a repeated CONFIG does not discard history.
func (hw *historyWriter) onSourceConfigured(sourceID string, sigs []udpsprotocol.SignalInfo) {
	if !hw.enabled() {
		return
	}
	for _, sig := range sigs {
		// Time-reference signals are the clock for the others, not data.
		if sig.TypeCode == histTypeCodeUint64 {
			continue
		}
		key := sourceID + ":" + sig.Name

		hw.mu.RLock()
		_, exists := hw.files[key]
		_, waiting := hw.pending[key]
		hw.mu.RUnlock()
		if exists || waiting {
			continue
		}

		// The rate the producer declares is what the file has to hold. Many
		// producers (the C++ UDPStreamer among them) declare none, and a fixed
		// guess would be orders of magnitude out at megasample rates — so those
		// signals wait for openPending to size them from the measured rate.
		if sig.SamplingRate <= 0 {
			hw.mu.Lock()
			hw.pending[key] = pendingHistSignal{sourceID: sourceID, signal: sig.Name}
			hw.mu.Unlock()
			continue
		}
		if _, err := hw.open(key, sourceID, sig.Name, sig.SamplingRate); err != nil {
			log.Printf("wshub/history: %s:%s: %v", sourceID, sig.Name, err)
		}
	}
}

// open creates (or reopens) one signal's file and registers it.
func (hw *historyWriter) open(key, sourceID, signal string, rateHz float64) (*histFile, error) {
	dir := filepath.Join(hw.cfg.Directory, sanitizeHistName(sourceID))
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return nil, fmt.Errorf("mkdir %s: %w", dir, err)
	}
	path := filepath.Join(dir, sanitizeHistName(signal)+".shist")
	hf, err := hw.openSignalFile(path, sourceID, signal, rateHz)
	if err != nil {
		return nil, err
	}
	hw.mu.Lock()
	hw.files[key] = hf
	delete(hw.pending, key)
	hw.mu.Unlock()
	return hf, nil
}

// pendingKeys lists the signals still waiting for a measured sampling rate.
func (hw *historyWriter) pendingKeys() []string {
	if !hw.enabled() {
		return nil
	}
	hw.mu.RLock()
	defer hw.mu.RUnlock()
	if len(hw.pending) == 0 {
		return nil
	}
	keys := make([]string, 0, len(hw.pending))
	for k := range hw.pending {
		keys = append(keys, k)
	}
	return keys
}

// openPending sizes and opens a deferred signal's file from a measured rate.
func (hw *historyWriter) openPending(key string, rateHz float64) bool {
	if !hw.enabled() || rateHz <= 0 {
		return false
	}
	hw.mu.RLock()
	p, waiting := hw.pending[key]
	hw.mu.RUnlock()
	if !waiting {
		return false
	}
	if _, err := hw.open(key, p.sourceID, p.signal, rateHz); err != nil {
		log.Printf("wshub/history: %s: %v", key, err)
		// Drop it rather than retrying every sweep: the error is a filesystem
		// one, and it would otherwise be logged once a second forever.
		hw.mu.Lock()
		delete(hw.pending, key)
		hw.mu.Unlock()
		return false
	}
	return true
}

// sanitizeHistName keeps a source or signal name usable as a path component:
// separators and traversal would otherwise let a producer-supplied name write
// outside the history directory.
func sanitizeHistName(s string) string {
	out := strings.Map(func(r rune) rune {
		switch {
		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
			return r
		case r == '-', r == '_', r == '.', r == '[', r == ']':
			return r
		default:
			return '_'
		}
	}, s)
	// ".", ".." and the empty string are not usable component names.
	if out == "" || strings.Trim(out, ".") == "" {
		return "_"
	}
	return out
}

// histCapacityFor sizes one signal's file and picks its min/max bucket width.
//
// What the file has to hold is the window being displayed, not a retention
// period: the archive exists so that a zoom or a trigger capture can be served
// after the in-memory rings have rolled past it, and neither ever asks for more
// than the window. So the file is `windowSec` of samples long — verbatim while
// that fits the budget, and otherwise as an envelope of `bucket` samples per
// min/max pair, the narrowest bucket that does fit. Sizing it for hours instead
// meant a 1 s live window was archived at a thousandth of its resolution, since
// the same budget had to stretch over 3600 times as much time.
//
// A returned bucket of 1 means every sample is archived verbatim.
func histCapacityFor(windowSec, rateHz float64, decimation, maxPts int) (uint32, int) {
	need := windowSec * rateHz / float64(decimation)
	if !(need > 0) { // also catches NaN
		return histMinCapacity, 1
	}
	budget := float64(maxPts)
	if maxPts <= 0 || budget > histMaxCapacity {
		budget = histMaxCapacity
	}
	if want := math.Ceil(need * histHeadroom); want <= budget {
		if want < histMinCapacity {
			return histMinCapacity, 1
		}
		return uint32(want), 1
	}
	// Two points per bucket, so `bucket` samples collapse to a min and a max.
	bucket := int(math.Ceil(2 * need * histHeadroom / budget))
	n := math.Ceil(2 * need * histHeadroom / float64(bucket))
	if n > budget {
		n = budget
	}
	if n < histMinCapacity {
		n = histMinCapacity
	}
	return uint32(n), bucket
}

// histCoverageSec is how much time a file of this geometry holds, at the rate it
// was sized from. A bucket of 2 stores both of its samples, so it reaches no
// further back than a bucket of 1.
func histCoverageSec(capacity uint32, bucket, decimation int, rateHz float64) float64 {
	if rateHz <= 0 {
		return 0
	}
	pairs := float64(capacity)
	if bucket > 2 {
		pairs = float64(capacity/2) * float64(bucket)
	}
	return pairs * float64(decimation) / rateHz
}

// window reports the timespan the files are currently sized for.
func (hw *historyWriter) window() float64 {
	if !hw.enabled() {
		return 0
	}
	hw.mu.RLock()
	defer hw.mu.RUnlock()
	return hw.windowSec
}

// setWindow points the archive at the timespan the clients are looking at, and
// re-sizes the files that no longer match it. It reports whether any file's
// geometry changed, which invalidates what clients know about the archive.
//
// As in setBudget, the archived samples do not survive a re-size: a file's
// capacity and bucket width are fixed when it is created. Widening the window
// therefore costs the history collected so far, which is the price of storing
// the window at the best resolution the budget can buy rather than storing
// hours of it at the worst.
func (hw *historyWriter) setWindow(sec float64) bool {
	if !hw.enabled() || !(sec > 0) {
		return false
	}
	hw.mu.Lock()
	if sec == hw.windowSec {
		hw.mu.Unlock()
		return false
	}
	hw.windowSec = sec
	hw.mu.Unlock()
	return hw.resizeForWindow(sec)
}

// resizeForWindow re-creates every file whose geometry no longer suits `window`.
func (hw *historyWriter) resizeForWindow(window float64) bool {
	hw.mu.RLock()
	decim := hw.cfg.Decimation
	keys := make([]string, 0, len(hw.files))
	files := make([]*histFile, 0, len(hw.files))
	for k, hf := range hw.files {
		keys = append(keys, k)
		files = append(files, hf)
	}
	hw.mu.RUnlock()

	changed := false
	for i, hf := range files {
		hf.mu.RLock()
		capacity, bucket, rate := hf.capacity, hf.bucket, hf.rateHz
		hf.mu.RUnlock()
		if rate <= 0 {
			continue
		}

		cov := histCoverageSec(capacity, bucket, decim, rate)
		newCap, newBucket := histCapacityFor(window, rate, decim, hw.budget())
		switch {
		case cov < window:
			// Too short to answer for the window: it has to grow, and losing what
			// it holds is the price of that.
		case bucket > 1 && newBucket < bucket && cov > 2*window:
			// Longer than it needs to be, and enveloped: a narrower window buys
			// resolution back. Twice the window of slack, as for the rings, so a
			// client nudging its window by a few percent does not cost it the
			// archive.
		default:
			// Longer than the window but already verbatim. Shrinking it would
			// discard history and buy nothing — a trigger arming with a 1 s window
			// must not throw away the seconds it is about to be asked for.
			continue
		}
		// A file at the minimum capacity, or already at the budget ceiling, can
		// come out of the sizer unchanged however far the window moved.
		if newCap == capacity && newBucket == bucket {
			continue
		}

		// Detach before closing: write() and readRange() look the key up under
		// this lock, so from here on they skip the signal rather than touching a
		// descriptor that is about to go away.
		hw.mu.Lock()
		if hw.files[keys[i]] != hf {
			hw.mu.Unlock()
			continue
		}
		delete(hw.files, keys[i])
		hw.mu.Unlock()

		hf.mu.Lock()
		if hf.dirty {
			_ = hf.flushHeaderLocked()
		}
		_ = hf.f.Close()
		hf.mu.Unlock()

		if _, err := hw.open(keys[i], hf.sourceID, hf.signal, rate); err != nil {
			log.Printf("wshub/history: re-size %s: %v", keys[i], err)
			continue
		}
		log.Printf("wshub/history: %s re-sized for a %.1f s window (cap %d→%d, min/max over %d→%d); archive reset",
			keys[i], window, capacity, newCap, bucket, newBucket)
		changed = true
	}
	return changed
}

// budget reports the current per-signal point budget. It is read under the lock
// because setBudget can change it from a WebSocket goroutine while the hub's Run
// goroutine is opening a deferred file.
func (hw *historyWriter) budget() int {
	if !hw.enabled() {
		return 0
	}
	hw.mu.RLock()
	defer hw.mu.RUnlock()
	return hw.cfg.MaxPointsPerSignal
}

// setBudget changes the per-signal point budget and re-creates every open file
// at the new size, returning the budget actually applied.
//
// The archived samples do not survive: a file's geometry — its capacity and its
// min/max bucket width — is fixed when it is created, and re-bucketing an
// existing envelope into a different one would be inventing samples. Callers
// must make that clear to the user before asking for it.
func (hw *historyWriter) setBudget(maxPts int) int {
	if !hw.enabled() {
		return 0
	}
	if maxPts <= 0 {
		maxPts = histDefaultMaxPoints
	}
	if maxPts > histMaxCapacity {
		maxPts = histMaxCapacity
	}

	hw.mu.Lock()
	if maxPts == hw.cfg.MaxPointsPerSignal {
		hw.mu.Unlock()
		return maxPts
	}
	hw.cfg.MaxPointsPerSignal = maxPts
	// Detach the old files before reopening: write() and readRange() look the
	// key up under this lock, so from here on they find nothing and skip the
	// signal rather than touching a descriptor that is about to be closed.
	old := hw.files
	hw.files = make(map[string]*histFile, len(old))
	hw.mu.Unlock()

	for key, hf := range old {
		hf.mu.Lock()
		if hf.dirty {
			_ = hf.flushHeaderLocked()
		}
		_ = hf.f.Close()
		rate := hf.rateHz
		hf.mu.Unlock()
		if _, err := hw.open(key, hf.sourceID, hf.signal, rate); err != nil {
			log.Printf("wshub/history: resize %s: %v", key, err)
		}
	}
	log.Printf("wshub/history: budget set to %.3f MPts per signal, %d file(s) re-sized",
		float64(maxPts)/1e6, len(old))
	return maxPts
}

// openSignalFile reopens an existing file whose capacity still matches, and
// otherwise creates a fresh pre-allocated one.
func (hw *historyWriter) openSignalFile(path, sourceID, signal string, rateHz float64) (*histFile, error) {
	capacity, bucket := histCapacityFor(hw.window(), rateHz, hw.cfg.Decimation, hw.budget())
	reduction := uint32(bucket * hw.cfg.Decimation)

	if f, err := os.OpenFile(path, os.O_RDWR, 0o644); err == nil {
		hdr := make([]byte, histHeaderSize)
		if _, rerr := f.ReadAt(hdr, 0); rerr == nil &&
			string(hdr[0:4]) == string(histMagic[:]) &&
			binary.LittleEndian.Uint32(hdr[4:]) == histVersion &&
			binary.LittleEndian.Uint32(hdr[8:]) == capacity &&
			// Same size but a different resolution would splice two
			// incompatible envelopes into one file.
			binary.LittleEndian.Uint32(hdr[20:]) == reduction {
			hf := &histFile{
				f:         f,
				capacity:  capacity,
				head:      binary.LittleEndian.Uint32(hdr[12:]),
				count:     binary.LittleEndian.Uint32(hdr[16:]),
				tOldest:   math.Float64frombits(binary.LittleEndian.Uint64(hdr[24:])),
				tNewest:   math.Float64frombits(binary.LittleEndian.Uint64(hdr[32:])),
				bucket:    bucket,
				reduction: reduction,
				rateHz:    rateHz,
				sourceID:  sourceID,
				signal:    signal,
			}
			if hf.head < capacity && hf.count <= capacity {
				log.Printf("wshub/history: reopened %s (cap=%d, count=%d, bucket=%d)",
					path, capacity, hf.count, bucket)
				return hf, nil
			}
		}
		// Wrong capacity, wrong version, wrong resolution or a corrupt header:
		// start over.
		f.Close()
	}

	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o644)
	if err != nil {
		return nil, err
	}
	size := int64(histHeaderSize) + int64(capacity)*histPairSize
	if err := f.Truncate(size); err != nil {
		f.Close()
		return nil, fmt.Errorf("preallocate %d bytes: %w", size, err)
	}
	hf := &histFile{f: f, capacity: capacity, bucket: bucket, reduction: reduction,
		rateHz: rateHz, sourceID: sourceID, signal: signal}
	if err := hf.flushHeaderLocked(); err != nil {
		f.Close()
		return nil, err
	}
	if bucket > 1 {
		log.Printf("wshub/history: created %s (cap=%d, %.1f MB, min/max over %d samples = %.0f Sps stored)",
			path, capacity, float64(size)/(1024*1024), bucket, 2*rateHz/float64(bucket))
	} else {
		log.Printf("wshub/history: created %s (cap=%d, %.1f MB, full resolution)",
			path, capacity, float64(size)/(1024*1024))
	}
	return hf, nil
}

// write archives a batch of full-resolution samples for one signal. Unknown
// keys and a disabled writer are no-ops, so callers need no guard.
func (hw *historyWriter) write(key string, t, v []float64) {
	if !hw.enabled() || len(t) == 0 || len(t) != len(v) {
		return
	}
	hw.diskMu.RLock()
	low := hw.diskLow
	hw.diskMu.RUnlock()
	if low {
		return
	}
	hw.mu.RLock()
	hf := hw.files[key]
	hw.mu.RUnlock()
	if hf == nil {
		return
	}

	if hw.cfg.Decimation > 1 {
		// Decimate into scratch rather than in place: t/v are the same slices
		// the zoom ring and the trigger were handed.
		kt := make([]float64, 0, len(t)/hw.cfg.Decimation+1)
		kv := make([]float64, 0, cap(kt))
		hf.mu.Lock()
		c := hf.decimCount
		for i := range t {
			c++
			if c >= hw.cfg.Decimation {
				kt = append(kt, t[i])
				kv = append(kv, v[i])
				c = 0
			}
		}
		hf.decimCount = c
		hf.mu.Unlock()
		t, v = kt, kv
		if len(t) == 0 {
			return
		}
	}

	// Too fast to archive sample-for-sample within the budget: store the min/max
	// envelope instead, so the file still covers the configured duration and a
	// spike survives at whatever resolution the budget allows.
	if hf.bucket > 1 {
		hf.mu.Lock()
		t, v = hf.foldBucket(t, v)
		hf.mu.Unlock()
		if len(t) == 0 {
			// The bucket in progress is not complete yet.
			return
		}
	}

	if err := hf.writePairs(t, v); err != nil {
		log.Printf("wshub/history: write %s: %v", key, err)
	}
}

// captureRange copies [t0, t1] out of every archive file into a capture file of
// its own, which nothing overwrites until the next trigger. The in-memory rings
// roll past a captured window within seconds, so this copy is what a zoom into
// the capture reads from once they have.
//
// Protecting the window in place instead — pinning the region and refusing to
// wrap onto it — was tried and cannot work: the archive is circular and sized in
// seconds, so a capture held for longer than the file covers stops the archive
// dead, and the hole then lands exactly where the *next* capture's pre-trigger
// window belongs. Copying costs one window's worth of disk per signal and leaves
// the archive rolling untouched.
func (hw *historyWriter) captureRange(t0, t1 float64) {
	if !hw.enabled() || t1 <= t0 {
		return
	}
	hw.mu.RLock()
	files := make(map[string]*histFile, len(hw.files))
	for k, hf := range hw.files {
		files[k] = hf
	}
	hw.mu.RUnlock()

	// The copy runs on the hub's goroutine, so how long it takes is how long
	// ingest is held up: report it rather than leaving a stall to be guessed at.
	started := time.Now()
	made := make(map[string]*histFile, len(files))
	for key, hf := range files {
		cf, err := hw.copyRange(hf, t0, t1)
		if err != nil {
			log.Printf("wshub/history: capture %s: %v", key, err)
			continue
		}
		if cf != nil {
			made[key] = cf
		}
	}

	hw.capMu.Lock()
	old := hw.captures
	hw.captures = made
	hw.capMu.Unlock()
	for _, cf := range old {
		_ = cf.f.Close()
	}
	log.Printf("wshub/history: captured [%.3f, %.3f] for %d/%d signal(s) in %s",
		t0, t1, len(made), len(files), time.Since(started).Round(time.Millisecond))
}

// histCopyChunk is how many pairs one copyRange pass moves. The archive's lock
// is dropped between passes so a megasample stream keeps being written while a
// capture of it is copied out.
const histCopyChunk = 65536

// copyRange writes the archived samples in [t0, t1] to the signal's capture
// file, returning it ready to read. A capture file is laid out as a full,
// non-wrapping histFile (capacity == count, head == 0), so readRange serves it
// with no special case. It returns nil when the archive holds nothing in range.
func (hw *historyWriter) copyRange(hf *histFile, t0, t1 float64) (*histFile, error) {
	path := filepath.Join(hw.cfg.Directory, sanitizeHistName(hf.sourceID),
		sanitizeHistName(hf.signal)+".cap")
	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o644)
	if err != nil {
		return nil, err
	}

	// The copy inherits the archive's resolution: it is the same samples.
	out := &histFile{f: f, sourceID: hf.sourceID, signal: hf.signal,
		bucket: hf.bucket, reduction: hf.reduction}
	// after is the exclusive lower bound of the next pass: resuming by time
	// rather than by index keeps the copy correct even if the archive wrapped
	// between passes.
	after := math.Inf(-1)
	first := true
	for {
		buf, lastT, err := hf.readAfter(after, t0, t1, histCopyChunk)
		if err != nil {
			f.Close()
			return nil, err
		}
		if len(buf) == 0 {
			break
		}
		if _, err := f.WriteAt(buf, int64(histHeaderSize)+int64(out.count)*histPairSize); err != nil {
			f.Close()
			return nil, err
		}
		if first {
			out.tOldest = math.Float64frombits(binary.LittleEndian.Uint64(buf))
			first = false
		}
		out.count += uint32(len(buf) / histPairSize)
		out.tNewest, after = lastT, lastT
	}
	if out.count == 0 {
		f.Close()
		os.Remove(path)
		return nil, nil
	}
	out.capacity = out.count
	if err := out.flushHeaderLocked(); err != nil {
		f.Close()
		return nil, err
	}
	return out, nil
}

// readAfter returns up to max encoded pairs whose timestamp is in (after, t1]
// and at or after t0, plus the timestamp of the last one.
func (hf *histFile) readAfter(after, t0, t1 float64, max int) ([]byte, float64, error) {
	hf.mu.RLock()
	defer hf.mu.RUnlock()
	if hf.count == 0 {
		return nil, 0, nil
	}
	capacity, count := hf.capacity, hf.count
	oldest := (hf.head + capacity - count) % capacity
	timeAt := func(i uint32) float64 {
		var b [8]byte
		if _, err := hf.f.ReadAt(b[:], int64(histHeaderSize)+int64((oldest+i)%capacity)*histPairSize); err != nil {
			return math.NaN()
		}
		return math.Float64frombits(binary.LittleEndian.Uint64(b[:]))
	}

	lo := histSearch(0, count, func(i uint32) bool {
		t := timeAt(i)
		return t < t0 || t <= after
	})
	hi := histSearch(lo, count, func(i uint32) bool { return timeAt(i) <= t1 })
	n := int(hi - lo)
	if n <= 0 {
		return nil, 0, nil
	}
	if n > max {
		n = max
	}

	// The run wraps at most once, so it costs at most two reads.
	buf := make([]byte, n*histPairSize)
	start := (oldest + lo) % capacity
	head := int(capacity-start) * histPairSize
	if head > len(buf) {
		head = len(buf)
	}
	if _, err := hf.f.ReadAt(buf[:head], int64(histHeaderSize)+int64(start)*histPairSize); err != nil {
		return nil, 0, err
	}
	if head < len(buf) {
		if _, err := hf.f.ReadAt(buf[head:], histHeaderSize); err != nil {
			return nil, 0, err
		}
	}
	lastT := math.Float64frombits(binary.LittleEndian.Uint64(buf[(n-1)*histPairSize:]))
	return buf, lastT, nil
}

// captureFor returns the capture file to answer a [t0, t1] query, or nil when
// the archive should answer it. The capture is only preferred when it holds the
// whole range: a query wider than one trigger window belongs to the archive.
func (hw *historyWriter) captureFor(key string, t0, t1 float64) *histFile {
	hw.capMu.RLock()
	cf := hw.captures[key]
	hw.capMu.RUnlock()
	if cf == nil || cf.count == 0 {
		return nil
	}
	// The client asks for the window it was given, whose edges are the first and
	// last samples in the capture; float arithmetic on the way there can put the
	// request a sample or two outside it.
	tol := 0.01 * (t1 - t0)
	if cf.tOldest <= t0+tol && cf.tNewest >= t1-tol {
		return cf
	}
	return nil
}

// writePairs appends samples at the head, wrapping. A batch spans at most two
// contiguous runs, so it costs at most two writes however large it is —
// per-sample writes would not keep up with a megasample-per-second signal.
func (hf *histFile) writePairs(t, v []float64) error {
	hf.mu.Lock()
	defer hf.mu.Unlock()

	n := len(t)
	// Only the last `capacity` samples can survive; writing the rest would
	// just overwrite them within this same call.
	if n > int(hf.capacity) {
		t, v = t[n-int(hf.capacity):], v[n-int(hf.capacity):]
		n = len(t)
	}
	buf := make([]byte, n*histPairSize)
	for i := range t {
		binary.LittleEndian.PutUint64(buf[i*histPairSize:], math.Float64bits(t[i]))
		binary.LittleEndian.PutUint64(buf[i*histPairSize+8:], math.Float64bits(v[i]))
	}

	first := int(hf.capacity - hf.head)
	if first > n {
		first = n
	}
	off := int64(histHeaderSize) + int64(hf.head)*histPairSize
	if _, err := hf.f.WriteAt(buf[:first*histPairSize], off); err != nil {
		return err
	}
	if first < n {
		if _, err := hf.f.WriteAt(buf[first*histPairSize:], histHeaderSize); err != nil {
			return err
		}
	}

	hf.head = (hf.head + uint32(n)) % hf.capacity
	if hf.count+uint32(n) < hf.capacity {
		hf.count += uint32(n)
	} else {
		hf.count = hf.capacity
	}

	hf.tNewest = t[n-1]
	if hf.count <= uint32(n) {
		hf.tOldest = t[0]
	} else {
		oldest := (hf.head + hf.capacity - hf.count) % hf.capacity
		var b [8]byte
		if _, err := hf.f.ReadAt(b[:], int64(histHeaderSize)+int64(oldest)*histPairSize); err == nil {
			hf.tOldest = math.Float64frombits(binary.LittleEndian.Uint64(b[:]))
		}
	}
	hf.dirty = true
	return nil
}

// flushHeaders persists the in-memory header of every dirty file and refreshes
// the disk-space guard. Until this runs, a crash loses the samples written
// since the last flush — the data region is on disk but the header does not
// account for it yet.
func (hw *historyWriter) flushHeaders() {
	if !hw.enabled() {
		return
	}
	low := !hw.hasSufficientDisk()
	hw.diskMu.Lock()
	wasLow := hw.diskLow
	hw.diskLow = low
	hw.diskMu.Unlock()
	if low && !wasLow {
		log.Printf("wshub/history: free space below %d MB, writing paused", hw.cfg.MinDiskFreeMB)
	} else if !low && wasLow {
		log.Printf("wshub/history: free space recovered, writing resumed")
	}

	hw.mu.RLock()
	files := make([]*histFile, 0, len(hw.files))
	for _, hf := range hw.files {
		files = append(files, hf)
	}
	hw.mu.RUnlock()

	for _, hf := range files {
		hf.mu.Lock()
		if hf.dirty {
			if err := hf.flushHeaderLocked(); err != nil {
				log.Printf("wshub/history: flush %s:%s: %v", hf.sourceID, hf.signal, err)
			}
		}
		hf.mu.Unlock()
	}
}

func (hf *histFile) flushHeaderLocked() error {
	hdr := make([]byte, histHeaderSize)
	copy(hdr[0:4], histMagic[:])
	binary.LittleEndian.PutUint32(hdr[4:], histVersion)
	binary.LittleEndian.PutUint32(hdr[8:], hf.capacity)
	binary.LittleEndian.PutUint32(hdr[12:], hf.head)
	binary.LittleEndian.PutUint32(hdr[16:], hf.count)
	binary.LittleEndian.PutUint32(hdr[20:], hf.reduction)
	binary.LittleEndian.PutUint64(hdr[24:], math.Float64bits(hf.tOldest))
	binary.LittleEndian.PutUint64(hdr[32:], math.Float64bits(hf.tNewest))
	if _, err := hf.f.WriteAt(hdr, 0); err != nil {
		return err
	}
	hf.dirty = false
	return hf.f.Sync()
}

// close flushes and closes every file.
func (hw *historyWriter) close() {
	if !hw.enabled() {
		return
	}
	hw.mu.Lock()
	defer hw.mu.Unlock()
	for _, hf := range hw.files {
		hf.mu.Lock()
		if hf.dirty {
			_ = hf.flushHeaderLocked()
		}
		_ = hf.f.Close()
		hf.mu.Unlock()
	}
	hw.files = make(map[string]*histFile)

	hw.capMu.Lock()
	for _, cf := range hw.captures {
		_ = cf.f.Close()
	}
	hw.captures = nil
	hw.capMu.Unlock()
}

// readRange returns every archived pair in [t0, t1] for one signal, capped at
// maxOut samples.
func (hw *historyWriter) readRange(key string, t0, t1 float64, maxOut int) ([]float64, []float64) {
	if !hw.enabled() || t1 < t0 || maxOut <= 0 {
		return nil, nil
	}
	hw.mu.RLock()
	hf := hw.files[key]
	hw.mu.RUnlock()
	// A zoom into a trigger window is served from the capture copy: the archive
	// is circular and has usually rolled past that window by the time anyone
	// zooms into it a second time.
	if cf := hw.captureFor(key, t0, t1); cf != nil {
		hf = cf
	}
	if hf == nil {
		return nil, nil
	}

	hf.mu.RLock()
	defer hf.mu.RUnlock()
	if hf.count == 0 {
		return nil, nil
	}

	capacity, count := hf.capacity, hf.count
	oldest := (hf.head + capacity - count) % capacity
	// timeAt reads one timestamp by logical index; the binary searches touch
	// only log2(count) of them, which is what makes a query over a multi-
	// gigabyte file cheap.
	timeAt := func(i uint32) float64 {
		var b [8]byte
		phys := (oldest + i) % capacity
		if _, err := hf.f.ReadAt(b[:], int64(histHeaderSize)+int64(phys)*histPairSize); err != nil {
			return math.NaN()
		}
		return math.Float64frombits(binary.LittleEndian.Uint64(b[:]))
	}

	lo := histSearch(0, count, func(i uint32) bool { return timeAt(i) < t0 })
	hi := histSearch(lo, count, func(i uint32) bool { return timeAt(i) <= t1 })
	span := int(hi - lo)
	if span <= 0 {
		return nil, nil
	}

	// A range wider than the cap is thinned out across its whole width. Taking
	// the first maxOut samples instead would answer a 10 s query with its first
	// 2 ms, which reads as an empty plot to a client and sends it back to its
	// own coarse copy of the data.
	stride := 1
	if span > maxOut {
		stride = (span + maxOut - 1) / maxOut
	}
	n := (span + stride - 1) / stride

	outT := make([]float64, n)
	outV := make([]float64, n)
	if stride > 1 {
		// Reading the whole range to throw most of it away costs stride times
		// the I/O, and at megasample rates that is hundreds of megabytes per
		// zoom, so the wanted samples are read one by one instead.
		var b [histPairSize]byte
		for i := 0; i < n; i++ {
			phys := (oldest + lo + uint32(i*stride)) % capacity
			if _, err := hf.f.ReadAt(b[:], int64(histHeaderSize)+int64(phys)*histPairSize); err != nil {
				return nil, nil
			}
			outT[i] = math.Float64frombits(binary.LittleEndian.Uint64(b[0:]))
			outV[i] = math.Float64frombits(binary.LittleEndian.Uint64(b[8:]))
		}
		return outT, outV
	}

	// Read in contiguous runs: the range wraps at most once.
	buf := make([]byte, n*histPairSize)
	start := (oldest + lo) % capacity
	first := int(capacity - start)
	if first > n {
		first = n
	}
	if _, err := hf.f.ReadAt(buf[:first*histPairSize],
		int64(histHeaderSize)+int64(start)*histPairSize); err != nil {
		return nil, nil
	}
	if first < n {
		if _, err := hf.f.ReadAt(buf[first*histPairSize:], histHeaderSize); err != nil {
			return nil, nil
		}
	}
	for i := 0; i < n; i++ {
		outT[i] = math.Float64frombits(binary.LittleEndian.Uint64(buf[i*histPairSize:]))
		outV[i] = math.Float64frombits(binary.LittleEndian.Uint64(buf[i*histPairSize+8:]))
	}
	return outT, outV
}

// histSearch returns the first index in [a, b) for which pred is false; pred
// must be monotonically decreasing over the range.
func histSearch(a, b uint32, pred func(uint32) bool) uint32 {
	for a < b {
		mid := a + (b-a)/2
		if pred(mid) {
			a = mid + 1
		} else {
			b = mid
		}
	}
	return a
}

// info snapshots the per-signal metadata for a historyInfo event.
func (hw *historyWriter) info() map[string]HistorySignalInfo {
	out := make(map[string]HistorySignalInfo)
	if !hw.enabled() {
		return out
	}
	hw.mu.RLock()
	files := make(map[string]*histFile, len(hw.files))
	for k, hf := range hw.files {
		files[k] = hf
	}
	hw.mu.RUnlock()
	for k, hf := range files {
		hf.mu.RLock()
		// Reported even at count==0: the file exists but has not received data
		// yet, and clients need the entry to enable their history UI.
		out[k] = HistorySignalInfo{
			T0: hf.tOldest, T1: hf.tNewest,
			Count: hf.count, Capacity: hf.capacity, Bucket: hf.bucket,
		}
		hf.mu.RUnlock()
	}
	return out
}

/* ─── Hub integration ─────────────────────────────────────────────────────── */

// histOpenIntervalSec throttles the deferred-open sweep: the rate can only be
// measured once data flows, so the sweep repeats until every signal is sized.
const histOpenIntervalSec = 1.0

// histRateMinSpan is the shortest stretch of ring data a rate is measured over.
// Shorter than this and a burst of packets reads as a far higher rate than the
// stream really has, which would size the file many times too large.
const histRateMinSpan = 0.5

// openPendingHistoryFiles sizes the history files of signals whose producer
// declared no sampling rate, using the rate measured on the live stream. The
// alternative — a fixed guess — is three orders of magnitude out for a 1 MSps
// stream, leaving a file sized for the window holding a fraction of a second.
func (h *Hub) openPendingHistoryFiles(nowSec float64) {
	if !h.hist.enabled() || nowSec < h.histOpenAt {
		return
	}
	h.histOpenAt = nowSec + histOpenIntervalSec

	opened := false
	for _, key := range h.hist.pendingKeys() {
		rb := h.getRing(key)
		if rb == nil {
			continue
		}
		// The rate of the samples arriving, not of the points the ring keeps: a
		// bucketed ring stores one pair per bucket, and sizing the file from that
		// would leave it covering a fraction of the window at the source rate it
		// is actually written at.
		rate := rb.sourceRate()
		if rate <= 0 {
			continue
		}
		// Enough elapsed time to trust the measurement: a burst of packets over a
		// few milliseconds reads as a far higher rate than the stream has.
		if count, span := rb.stats(); count < 2 || span < histRateMinSpan {
			continue
		}
		if h.hist.openPending(key, rate) {
			opened = true
		}
	}
	if opened {
		h.broadcast(h.buildHistoryInfoMsg())
	}
}

// buildHistoryInfoMsg encodes the "historyInfo" event. Clients use it to decide
// whether a requested window can be served from disk at all.
func (h *Hub) buildHistoryInfoMsg() []byte {
	m := map[string]any{
		"type":    "historyInfo",
		"enabled": h.hist.enabled(),
		"signals": map[string]HistorySignalInfo{},
	}
	if h.hist.enabled() {
		// The span the files are sized for, which is what the clients are
		// displaying — not a retention period.
		m["windowSec"] = h.hist.window()
		m["decimation"] = h.hist.cfg.Decimation
		// In MPts, matching the unit of the UI control and of -history-max-mpts.
		m["maxMPtsPerSignal"] = float64(h.hist.budget()) / 1e6
		m["signals"] = h.hist.info()
	} else {
		m["windowSec"] = 0.0
		m["decimation"] = 1
		m["maxMPtsPerSignal"] = 0.0
	}
	msg, err := json.Marshal(m)
	if err != nil {
		log.Printf("wshub/history: marshal historyInfo: %v", err)
		return nil
	}
	return msg
}

// handleHistoryCommand processes the history WS commands, returning false when
// the message type is not one of them.
func (h *Hub) handleHistoryCommand(c *wsClient, t string, env map[string]interface{}) bool {
	switch t {
	case "historyInfo":
		if msg := h.buildHistoryInfoMsg(); msg != nil {
			c.sendText(msg)
		}
	case "historyZoom":
		h.handleHistoryZoom(c, env)
	case "setHistoryBudget":
		h.handleSetHistoryBudget(env)
	default:
		return false
	}
	return true
}

// handleSetHistoryBudget applies a new per-signal archive budget, in MPts, and
// tells every client the resulting geometry. The change is broadcast rather than
// answered to the requester alone because it re-creates the files: every client's
// view of what history exists has just been invalidated.
func (h *Hub) handleSetHistoryBudget(env map[string]interface{}) {
	if !h.hist.enabled() {
		return
	}
	mpts, ok := env["maxMPtsPerSignal"].(float64)
	if !ok || math.IsNaN(mpts) || mpts < 0 {
		return
	}
	h.hist.setBudget(int(mpts * 1e6))
	if msg := h.buildHistoryInfoMsg(); msg != nil {
		h.broadcast(msg)
	}
}

// handleHistoryZoom answers a historyZoom request from disk. Same request and
// reply shape as "zoom", so clients can fall back to it transparently when a
// window reaches further back than the in-memory rings hold.
func (h *Hub) handleHistoryZoom(c *wsClient, env map[string]interface{}) {
	if !h.hist.enabled() {
		msg, _ := json.Marshal(map[string]any{
			"type": "historyZoom", "reqId": env["reqId"],
			"error": "history not enabled",
		})
		c.sendText(msg)
		return
	}
	t0, ok0 := env["t0"].(float64)
	t1, ok1 := env["t1"].(float64)
	if !ok0 || !ok1 || t1 <= t0 {
		return
	}
	nF, nOK := env["n"].(float64)
	n := zoomPoints(int(nF), nOK)
	sigCSV, _ := env["signals"].(string)

	// The envelope can only keep a peak it was shown, so the disk read is
	// oversampled relative to the plot's point budget and thinned afterwards.
	// The cap keeps a request for "no decimation" over a multi-hour window from
	// pulling the whole file into memory.
	readCap := n * histReadOversample
	if readCap > histMaxReadPoints {
		readCap = histMaxReadPoints
	}

	signals := make(map[string]sigData)
	for _, k := range strings.Split(sigCSV, ",") {
		k = strings.TrimSpace(k)
		if k == "" {
			continue
		}
		rt, rv := h.hist.readRange(k, t0, t1, readCap)
		if len(rt) == 0 {
			continue
		}
		dt, dv := minMaxDecimate(rt, rv, n)
		signals[k] = sigData{T: dt, V: dv}
	}

	msg, err := json.Marshal(map[string]any{
		"type": "historyZoom", "reqId": env["reqId"], "signals": signals,
	})
	if err != nil {
		log.Printf("wshub/history: marshal historyZoom: %v", err)
		return
	}
	c.sendText(msg)
}

// histMaxReadPoints bounds one historyZoom read per signal.
const histMaxReadPoints = 2_000_000

// histReadOversample is how many disk samples are read per plotted point. The
// reads are ascending, so the kernel's readahead still covers them at moderate
// strides; going much higher turns a zoom into a full scan of the range.
const histReadOversample = 8

// hasSufficientDisk reports whether the history filesystem still has the
// configured free-space margin.
func (hw *historyWriter) hasSufficientDisk() bool {
	if hw.cfg.MinDiskFreeMB < 0 {
		return true
	}
	var st syscall.Statfs_t
	if err := syscall.Statfs(hw.cfg.Directory, &st); err != nil {
		return true // cannot tell — do not stop archiving on a guess
	}
	freeMB := int64(st.Bavail) * st.Bsize / (1024 * 1024)
	return freeMB >= int64(hw.cfg.MinDiskFreeMB)
}