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

package wshub

import (
	"encoding/binary"
	"math"
	"os"
	"path/filepath"
	"testing"

	"marte2/common/udpsprotocol"
)

// newTestHistory opens a writer in a temp dir with one signal file of the given
// declared rate, and returns the writer plus that signal's key.
func newTestHistory(t *testing.T, cfg HistoryConfig, rate float64) (*historyWriter, string) {
	t.Helper()
	if cfg.Directory == "" {
		cfg.Directory = t.TempDir()
	}
	hw, err := newHistoryWriter(cfg)
	if err != nil {
		t.Fatalf("newHistoryWriter: %v", err)
	}
	hw.onSourceConfigured("src", []udpsprotocol.SignalInfo{
		{Name: "sig", TypeCode: 8, SamplingRate: rate},
	})
	t.Cleanup(hw.close)
	return hw, "src:sig"
}

func ramp(t0 float64, dt float64, n int) ([]float64, []float64) {
	ts := make([]float64, n)
	vs := make([]float64, n)
	for i := range ts {
		ts[i] = t0 + float64(i)*dt
		vs[i] = float64(i)
	}
	return ts, vs
}

// A budget that cannot hold the window at full rate must buy the window by
// widening the min/max bucket, not by archiving a shorter stretch: a user
// looking at 600 s wants 600 s of it archived, coarser if need be.
func TestHistCapacityKeepsWindowByBucketing(t *testing.T) {
	const mega = 1 << 20
	cases := []struct {
		name       string
		window     float64
		rate       float64
		maxPts     int
		wantBucket int
	}{
		// 60 s of 1 kSps is 60 k samples — well inside 1 MPt, so stored verbatim.
		{"slow signal keeps full resolution", 60, 1000, mega, 1},
		// 600 s of 1 MSps is 600 M samples against 16 Mi points: at 2 points per
		// bucket and the headroom, ceil(2 × 1.25 × 600e6 / 16Mi) = 90 per bucket.
		{"fast signal is enveloped", 600, 1e6, 16 * mega, 90},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			capacity, bucket := histCapacityFor(c.window, c.rate, 1, c.maxPts)
			if bucket != c.wantBucket {
				t.Errorf("bucket = %d, want %d", bucket, c.wantBucket)
			}
			if capacity > uint32(c.maxPts) {
				t.Errorf("capacity %d exceeds the %d-point budget", capacity, c.maxPts)
			}
			// The whole window has to fit, which is the entire point.
			if covered := histCoverageSec(capacity, bucket, 1, c.rate); covered < c.window {
				t.Errorf("archive covers %.1f s, want the %.1f s window", covered, c.window)
			}
		})
	}
}

// The file exists to serve the window, so it must track it: a client that widens
// what it displays must not be left reading an archive sized for the old span.
func TestHistorySetWindowResizesFiles(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{WindowSec: 10}, 1000)
	before := hw.files[key]
	if before.bucket != 1 || histCoverageSec(before.capacity, 1, 1, 1000) < 10 {
		t.Fatalf("initial geometry = cap %d bucket %d, want 10 s verbatim",
			before.capacity, before.bucket)
	}

	if !hw.setWindow(600) {
		t.Fatal("setWindow reported no change for a 60× wider window")
	}
	after := hw.files[key]
	if after == before {
		t.Fatal("the file was not re-created")
	}
	if cov := histCoverageSec(after.capacity, after.bucket, 1, 1000); cov < 600 {
		t.Fatalf("archive covers %.1f s, want the new 600 s window", cov)
	}

	// Same window again: nothing to do, and re-creating the file would throw the
	// archive away for nothing.
	if hw.setWindow(600) {
		t.Fatal("setWindow re-sized for an unchanged window")
	}
	// A nudge inside the hysteresis band must not either.
	if hw.setWindow(610) {
		t.Fatal("setWindow re-sized for a 2 % window change")
	}
	if hw.files[key] != after {
		t.Fatal("the file was re-created despite the hysteresis")
	}
}

// The archive is what a zoom beyond the rings reads, so a spike that only the
// archive still holds must survive being written to it.
func TestHistoryBucketedWriteKeepsPeaks(t *testing.T) {
	// 1 kSps for 1 s = 1000 samples, plus headroom, into a 100-point budget →
	// buckets of ceil(2 × 1.25 × 1000 / 100) = 25.
	hw, key := newTestHistory(t, HistoryConfig{
		WindowSec: 1, MinDiskFreeMB: -1, MaxPointsPerSignal: 100,
	}, 1000)
	hf := hw.files[key]
	if hf.bucket != 25 {
		t.Fatalf("bucket = %d, want 25", hf.bucket)
	}

	ts := make([]float64, 1000)
	vs := make([]float64, 1000)
	for i := range ts {
		ts[i] = float64(i) * 0.001
	}
	vs[137] = 7.5  // a one-sample positive spike
	vs[500] = -3.5 // and a negative one
	hw.write(key, ts, vs)

	rt, rv := hw.readRange(key, 0, 1, 1000)
	if len(rt) == 0 {
		t.Fatal("nothing archived")
	}
	hi, lo := false, false
	for i := range rv {
		if rv[i] == 7.5 && rt[i] == ts[137] {
			hi = true
		}
		if rv[i] == -3.5 && rt[i] == ts[500] {
			lo = true
		}
	}
	if !hi || !lo {
		t.Errorf("archive lost a spike (positive kept=%v, negative kept=%v)", hi, lo)
	}
	// A partial bucket is not written until it completes, so the last few
	// samples may be missing; everything before them must be there.
	if hf.count == 0 || hf.count > hf.capacity {
		t.Errorf("archived %d points into a %d-point file", hf.count, hf.capacity)
	}
}

func TestHistoryDisabledWithoutDirectory(t *testing.T) {
	hw, err := newHistoryWriter(HistoryConfig{})
	if err != nil {
		t.Fatalf("newHistoryWriter: %v", err)
	}
	if hw != nil {
		t.Fatal("empty Directory must disable history")
	}
	// Every method must stay usable on the nil writer, which is how the hub
	// avoids guarding each call site.
	if hw.enabled() {
		t.Fatal("nil writer reports enabled")
	}
	hw.write("src:sig", []float64{1}, []float64{1})
	hw.flushHeaders()
	hw.close()
	if rt, _ := hw.readRange("src:sig", 0, 1, 10); rt != nil {
		t.Fatal("nil writer returned data")
	}
	if len(hw.info()) != 0 {
		t.Fatal("nil writer returned info entries")
	}
}

func TestHistoryWriteReadRoundTrip(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{}, 100)
	ts, vs := ramp(10, 0.01, 500)
	hw.write(key, ts, vs)

	rt, rv := hw.readRange(key, 10.5, 11.0, 10000)
	if len(rt) != 51 { // inclusive both ends, 0.01 s spacing
		t.Fatalf("read %d points, want 51", len(rt))
	}
	if rt[0] < 10.5-1e-9 || rt[len(rt)-1] > 11.0+1e-9 {
		t.Fatalf("range [%v, %v] escapes the request", rt[0], rt[len(rt)-1])
	}
	for i := range rt {
		wantV := math.Round((rt[i] - 10) / 0.01)
		if math.Abs(rv[i]-wantV) > 1e-6 {
			t.Fatalf("point %d: value %v, want %v", i, rv[i], wantV)
		}
	}
}

func TestHistoryReadRangeOutsideDataIsEmpty(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{}, 100)
	ts, vs := ramp(10, 0.01, 100)
	hw.write(key, ts, vs)

	if rt, _ := hw.readRange(key, 100, 200, 1000); len(rt) != 0 {
		t.Fatalf("read %d points past the newest sample", len(rt))
	}
	if rt, _ := hw.readRange(key, 0, 5, 1000); len(rt) != 0 {
		t.Fatalf("read %d points before the oldest sample", len(rt))
	}
	if rt, _ := hw.readRange("src:missing", 10, 11, 1000); rt != nil {
		t.Fatal("unknown key returned data")
	}
	if rt, _ := hw.readRange(key, 11, 10, 1000); rt != nil {
		t.Fatal("inverted range returned data")
	}
}

// Once the file has wrapped, the oldest samples must be gone and the retained
// window must still read back contiguously across the wrap point.
func TestHistoryWrapAround(t *testing.T) {
	// A sub-second window at 1 Sps sizes below the 1000-pair floor, which is a
	// cheap capacity to wrap.
	hw, key := newTestHistory(t, HistoryConfig{WindowSec: 0.36}, 1)
	hf := hw.files[key]
	if hf.capacity != histMinCapacity {
		t.Fatalf("capacity = %d, want the %d floor", hf.capacity, histMinCapacity)
	}

	// 2.5 fills, in batches that do not align with the capacity so the wrap
	// lands mid-batch.
	total := 2500
	ts, vs := ramp(0, 1, total)
	for i := 0; i < total; i += 333 {
		end := i + 333
		if end > total {
			end = total
		}
		hw.write(key, ts[i:end], vs[i:end])
	}

	if hf.count != histMinCapacity {
		t.Fatalf("count = %d, want a full %d", hf.count, histMinCapacity)
	}
	wantOldest := float64(total - histMinCapacity)
	if hf.tOldest != wantOldest {
		t.Fatalf("tOldest = %v, want %v", hf.tOldest, wantOldest)
	}
	if hf.tNewest != float64(total-1) {
		t.Fatalf("tNewest = %v, want %v", hf.tNewest, float64(total-1))
	}

	rt, rv := hw.readRange(key, wantOldest, float64(total-1), 10000)
	if len(rt) != histMinCapacity {
		t.Fatalf("read %d points, want the full %d", len(rt), histMinCapacity)
	}
	for i := range rt {
		want := wantOldest + float64(i)
		if rt[i] != want || rv[i] != want {
			t.Fatalf("point %d = (%v, %v), want (%v, %v)", i, rt[i], rv[i], want, want)
		}
	}

	// The evicted samples must not come back.
	if et, _ := hw.readRange(key, 0, wantOldest-1, 10000); len(et) != 0 {
		t.Fatalf("read %d evicted points", len(et))
	}
}

// A single batch larger than the file keeps its tail, not its head.
func TestHistoryOversizedBatchKeepsTail(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{WindowSec: 0.36}, 1)
	ts, vs := ramp(0, 1, 3000)
	hw.write(key, ts, vs)

	hf := hw.files[key]
	if hf.count != histMinCapacity {
		t.Fatalf("count = %d, want %d", hf.count, histMinCapacity)
	}
	if hf.tNewest != 2999 {
		t.Fatalf("tNewest = %v, want 2999", hf.tNewest)
	}
	rt, _ := hw.readRange(key, 2000, 2999, 10000)
	if len(rt) != histMinCapacity || rt[0] != 2000 {
		t.Fatalf("retained window starts at %v with %d points, want 2000 / %d",
			rt[0], len(rt), histMinCapacity)
	}
}

func TestHistoryDecimation(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{Decimation: 4}, 100)
	// Two batches, so the decimation phase must carry across the call boundary
	// rather than restarting.
	ts, vs := ramp(0, 0.01, 100)
	hw.write(key, ts[:37], vs[:37])
	hw.write(key, ts[37:], vs[37:])

	rt, _ := hw.readRange(key, -1, 1e9, 10000)
	if len(rt) != 25 {
		t.Fatalf("kept %d of 100 points at decimation 4, want 25", len(rt))
	}
	for i := 1; i < len(rt); i++ {
		if d := rt[i] - rt[i-1]; math.Abs(d-0.04) > 1e-9 {
			t.Fatalf("spacing at %d = %v, want 0.04", i, d)
		}
	}
}

// The input slices are shared with the zoom ring and the trigger, so decimation
// must not touch them.
func TestHistoryWriteDoesNotMutateInput(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{Decimation: 3}, 100)
	ts, vs := ramp(0, 0.01, 30)
	tCopy := append([]float64(nil), ts...)
	vCopy := append([]float64(nil), vs...)
	hw.write(key, ts, vs)
	for i := range ts {
		if ts[i] != tCopy[i] || vs[i] != vCopy[i] {
			t.Fatalf("write mutated input at %d", i)
		}
	}
}

// Reopening the same directory must pick the file back up with its contents,
// which is the whole point of persisting the header.
func TestHistoryReopenPreservesData(t *testing.T) {
	dir := t.TempDir()
	cfg := HistoryConfig{Directory: dir, WindowSec: 0.36}
	sigs := []udpsprotocol.SignalInfo{{Name: "sig", TypeCode: 8, SamplingRate: 1}}

	hw, err := newHistoryWriter(cfg)
	if err != nil {
		t.Fatalf("newHistoryWriter: %v", err)
	}
	hw.onSourceConfigured("src", sigs)
	ts, vs := ramp(0, 1, 400)
	hw.write("src:sig", ts, vs)
	hw.close()

	hw2, err := newHistoryWriter(cfg)
	if err != nil {
		t.Fatalf("reopen: %v", err)
	}
	defer hw2.close()
	hw2.onSourceConfigured("src", sigs)

	hf := hw2.files["src:sig"]
	if hf.count != 400 || hf.head != 400 {
		t.Fatalf("reopened count=%d head=%d, want 400/400", hf.count, hf.head)
	}
	rt, rv := hw2.readRange("src:sig", 100, 199, 10000)
	if len(rt) != 100 || rt[0] != 100 || rv[0] != 100 {
		t.Fatalf("reopened read = %d points starting (%v, %v)", len(rt), rt[0], rv[0])
	}

	// Appending after the reopen must continue where the file left off.
	ts2, vs2 := ramp(400, 1, 50)
	hw2.write("src:sig", ts2, vs2)
	if hf.tNewest != 449 {
		t.Fatalf("tNewest after append = %v, want 449", hf.tNewest)
	}
}

// A file sized for a different rate cannot be reused, so it must be recreated
// rather than reopened with a mismatched capacity.
func TestHistoryReopenWithDifferentCapacityRecreates(t *testing.T) {
	dir := t.TempDir()
	cfg := HistoryConfig{Directory: dir, WindowSec: 3600}

	hw, _ := newHistoryWriter(cfg)
	hw.onSourceConfigured("src", []udpsprotocol.SignalInfo{
		{Name: "sig", TypeCode: 8, SamplingRate: 10},
	})
	firstCap := hw.files["src:sig"].capacity
	hw.write("src:sig", []float64{1, 2}, []float64{1, 2})
	hw.close()

	hw2, _ := newHistoryWriter(cfg)
	defer hw2.close()
	hw2.onSourceConfigured("src", []udpsprotocol.SignalInfo{
		{Name: "sig", TypeCode: 8, SamplingRate: 100}, // 10× the rate
	})
	hf := hw2.files["src:sig"]
	if hf.capacity == firstCap {
		t.Fatalf("capacity unchanged at %d despite a 10x rate change", firstCap)
	}
	if hf.count != 0 {
		t.Fatalf("recreated file kept %d samples", hf.count)
	}
}

// A corrupt header must not be trusted: the file gets rebuilt instead.
func TestHistoryCorruptHeaderRecreates(t *testing.T) {
	dir := t.TempDir()
	cfg := HistoryConfig{Directory: dir, WindowSec: 0.36}
	sigs := []udpsprotocol.SignalInfo{{Name: "sig", TypeCode: 8, SamplingRate: 1}}

	hw, _ := newHistoryWriter(cfg)
	hw.onSourceConfigured("src", sigs)
	hw.write("src:sig", []float64{1, 2, 3}, []float64{1, 2, 3})
	hw.close()

	path := filepath.Join(dir, "src", "sig.shist")
	f, err := os.OpenFile(path, os.O_RDWR, 0o644)
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	if _, err := f.WriteAt([]byte("XXXX"), 0); err != nil { // clobber the magic
		t.Fatalf("clobber: %v", err)
	}
	f.Close()

	hw2, _ := newHistoryWriter(cfg)
	defer hw2.close()
	hw2.onSourceConfigured("src", sigs)
	if got := hw2.files["src:sig"].count; got != 0 {
		t.Fatalf("count = %d, want a recreated empty file", got)
	}
}

// Raising the budget from the UI has to buy resolution: same duration, a
// narrower min/max bucket. Lowering it again must not overrun the new budget.
func TestSetBudgetRebucketsAtTheSameDuration(t *testing.T) {
	// 100 s of 100 kSps is 10 M samples, well past either budget.
	hw, key := newTestHistory(t, HistoryConfig{
		WindowSec: 100, MaxPointsPerSignal: 100_000,
	}, 1e5)

	before := hw.files[key]
	if before.bucket <= 1 {
		t.Fatalf("bucket = %d, want the signal enveloped to fit the budget", before.bucket)
	}

	if got := hw.setBudget(1_000_000); got != 1_000_000 {
		t.Fatalf("setBudget = %d, want 1000000", got)
	}
	after := hw.files[key]
	if after == before {
		t.Fatal("the file was not re-created")
	}
	if after.bucket >= before.bucket {
		t.Fatalf("bucket %d → %d, want a finer envelope for a 10× budget",
			before.bucket, after.bucket)
	}
	if after.capacity > 1_000_000 {
		t.Fatalf("capacity = %d, over the 1 MPts budget", after.capacity)
	}
	// The point of the envelope: the duration is covered whatever the budget.
	if cov := float64(after.capacity) * float64(after.bucket) / 2 / 1e5; cov < 99 {
		t.Fatalf("coverage = %.1f s, want ~100 s", cov)
	}

	if got := hw.setBudget(100_000); got != 100_000 {
		t.Fatalf("setBudget back = %d, want 100000", got)
	}
	if c := hw.files[key].capacity; c > 100_000 {
		t.Fatalf("capacity = %d, over the restored 100 kPts budget", c)
	}
}

// A budget that leaves a signal's geometry alone must leave its archive alone
// too — re-creating files nobody asked to resize would throw away history.
func TestSetBudgetKeepsUnaffectedFiles(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{
		WindowSec: 1, MaxPointsPerSignal: 16 << 20,
	}, 1000)
	ts, vs := ramp(0, 0.001, 100)
	hw.write(key, ts, vs)

	hw.setBudget(8 << 20) // still far more than the 1000 points this signal needs
	hf := hw.files[key]
	if hf.bucket != 1 {
		t.Fatalf("bucket = %d, want the slow signal still archived verbatim", hf.bucket)
	}
	if hf.count != 100 {
		t.Fatalf("count = %d, want the 100 archived samples kept", hf.count)
	}
}

// Time-reference signals are the clock for the others, so archiving them would
// just waste disk.
func TestHistorySkipsTimeSignals(t *testing.T) {
	dir := t.TempDir()
	hw, _ := newHistoryWriter(HistoryConfig{Directory: dir})
	defer hw.close()
	hw.onSourceConfigured("src", []udpsprotocol.SignalInfo{
		{Name: "TimeArray", TypeCode: histTypeCodeUint64, SamplingRate: 1000},
		{Name: "data", TypeCode: 8, SamplingRate: 1000},
	})
	if _, ok := hw.files["src:TimeArray"]; ok {
		t.Fatal("uint64 time signal was archived")
	}
	if _, ok := hw.files["src:data"]; !ok {
		t.Fatal("data signal was not archived")
	}
}

// A second CONFIG for the same source must not throw away the history already
// collected for signals it re-declares.
func TestHistoryReconfigureKeepsExistingFile(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{WindowSec: 0.36}, 1)
	hw.write(key, []float64{1, 2, 3}, []float64{1, 2, 3})
	before := hw.files[key]

	hw.onSourceConfigured("src", []udpsprotocol.SignalInfo{
		{Name: "sig", TypeCode: 8, SamplingRate: 1},
		{Name: "sig2", TypeCode: 8, SamplingRate: 1},
	})
	if hw.files[key] != before {
		t.Fatal("re-CONFIG replaced the existing signal file")
	}
	if before.count != 3 {
		t.Fatalf("count = %d, want the 3 already written", before.count)
	}
	if _, ok := hw.files["src:sig2"]; !ok {
		t.Fatal("newly declared signal was not opened")
	}
}

// The C++ UDPStreamer declares samplingRate=0, so sizing the file on the spot
// would use a guess that is three orders of magnitude out at 1 MSps.
func TestHistoryDefersSignalsWithoutDeclaredRate(t *testing.T) {
	dir := t.TempDir()
	hw, _ := newHistoryWriter(HistoryConfig{Directory: dir})
	defer hw.close()
	hw.onSourceConfigured("src", []udpsprotocol.SignalInfo{
		{Name: "fast", TypeCode: 8, SamplingRate: 0},
		{Name: "known", TypeCode: 8, SamplingRate: 100},
	})

	if _, ok := hw.files["src:fast"]; ok {
		t.Fatal("undeclared-rate signal was sized before its rate was measured")
	}
	if got := hw.pendingKeys(); len(got) != 1 || got[0] != "src:fast" {
		t.Fatalf("pendingKeys = %v, want [src:fast]", got)
	}
	if _, ok := hw.files["src:known"]; !ok {
		t.Fatal("declared-rate signal was deferred")
	}
	// Data for a deferred signal is dropped, not misfiled.
	hw.write("src:fast", []float64{1}, []float64{1})

	// A repeated CONFIG must not queue it twice.
	hw.onSourceConfigured("src", []udpsprotocol.SignalInfo{
		{Name: "fast", TypeCode: 8, SamplingRate: 0},
	})
	if got := hw.pendingKeys(); len(got) != 1 {
		t.Fatalf("pendingKeys = %v after re-CONFIG, want one entry", got)
	}

	if !hw.openPending("src:fast", 100000) {
		t.Fatal("openPending refused a measured rate")
	}
	hf, ok := hw.files["src:fast"]
	if !ok {
		t.Fatal("file not opened after the rate was measured")
	}
	// The default window × 100 kSps, enveloped if it does not fit the budget.
	wantCap, wantBucket := histCapacityFor(defaultLiveWindowSec, 100000, 1, histDefaultMaxPoints)
	if hf.capacity != wantCap || hf.bucket != wantBucket {
		t.Fatalf("capacity/bucket = %d/%d, want %d/%d", hf.capacity, hf.bucket, wantCap, wantBucket)
	}
	if len(hw.pendingKeys()) != 0 {
		t.Fatal("signal still pending after being opened")
	}
	if hw.openPending("src:fast", 100000) {
		t.Fatal("openPending reopened an already-open signal")
	}
}

func TestOpenPendingHistoryFilesUsesMeasuredRate(t *testing.T) {
	h := NewHub()
	if err := h.EnableHistory(HistoryConfig{Directory: t.TempDir(), WindowSec: 3.6}); err != nil {
		t.Fatalf("EnableHistory: %v", err)
	}
	defer h.CloseHistory()
	h.hist.onSourceConfigured("s1", []udpsprotocol.SignalInfo{
		{Name: "sig", TypeCode: 8, SamplingRate: 0},
	})

	rb := newSigRing(200000)
	h.rings["s1:sig"] = rb

	// Too little data to measure a rate from: the sweep must wait rather than
	// size the file from a burst.
	fillRing(rb, 0, 100000, 100) // 1 ms of data
	h.openPendingHistoryFiles(100)
	if len(h.hist.pendingKeys()) != 1 {
		t.Fatal("sweep sized the file from a sub-millisecond sample")
	}

	fillRing(rb, 0, 100000, 100000) // 1 s at 100 kSps
	h.openPendingHistoryFiles(200)
	hf, ok := h.hist.files["s1:sig"]
	if !ok {
		t.Fatal("file not opened once the rate was measurable")
	}
	// 3.6 s at ~100 kSps, plus headroom, ≈ 450 000 pairs; a fixed 1 kHz guess
	// would have produced the 1000-sample floor instead.
	if hf.capacity < 400_000 || hf.capacity > 500_000 {
		t.Fatalf("capacity = %d, want ~450000 from the measured 100 kSps", hf.capacity)
	}
}

func TestOpenPendingHistoryFilesIsThrottled(t *testing.T) {
	h := NewHub()
	if err := h.EnableHistory(HistoryConfig{Directory: t.TempDir()}); err != nil {
		t.Fatalf("EnableHistory: %v", err)
	}
	defer h.CloseHistory()
	h.hist.onSourceConfigured("s1", []udpsprotocol.SignalInfo{
		{Name: "sig", TypeCode: 8, SamplingRate: 0},
	})

	h.openPendingHistoryFiles(100) // no ring yet: nothing to measure
	rb := newSigRing(20000)
	fillRing(rb, 0, 1000, 20000)
	h.rings["s1:sig"] = rb

	h.openPendingHistoryFiles(100.5)
	if len(h.hist.files) != 0 {
		t.Fatal("sweep ran inside the throttle window")
	}
	h.openPendingHistoryFiles(200)
	if len(h.hist.files) != 1 {
		t.Fatal("sweep did not run after the throttle window elapsed")
	}
}

// A hub without history must tolerate the sweep, since Run() calls it every tick.
func TestOpenPendingHistoryFilesNoopWithoutHistory(t *testing.T) {
	h := NewHub()
	h.openPendingHistoryFiles(100)
}

func TestHistoryInfoShape(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{WindowSec: 0.36}, 1)
	// Reported before any data arrives, so clients can enable their history UI.
	inf := hw.info()
	if e, ok := inf[key]; !ok || e.Count != 0 || e.Capacity != histMinCapacity {
		t.Fatalf("pre-data info = %+v (present=%v)", inf[key], ok)
	}

	ts, vs := ramp(5, 1, 10)
	hw.write(key, ts, vs)
	e := hw.info()[key]
	if e.Count != 10 || e.T0 != 5 || e.T1 != 14 {
		t.Fatalf("info = %+v, want count=10 t0=5 t1=14", e)
	}
}

func TestHistoryHeaderIsPersistedOnFlush(t *testing.T) {
	dir := t.TempDir()
	hw, key := newTestHistory(t, HistoryConfig{Directory: dir, WindowSec: 0.36, Decimation: 2}, 1)
	ts, vs := ramp(0, 1, 20)
	hw.write(key, ts, vs)
	hw.flushHeaders()

	hdr, err := os.ReadFile(filepath.Join(dir, "src", "sig.shist"))
	if err != nil {
		t.Fatalf("read: %v", err)
	}
	if string(hdr[0:4]) != "SHR1" {
		t.Fatalf("magic = %q", hdr[0:4])
	}
	if v := binary.LittleEndian.Uint32(hdr[4:]); v != histVersion {
		t.Fatalf("version = %d, want %d", v, histVersion)
	}
	if c := binary.LittleEndian.Uint32(hdr[8:]); c != histMinCapacity {
		t.Fatalf("capacity = %d, want %d", c, histMinCapacity)
	}
	if h := binary.LittleEndian.Uint32(hdr[12:]); h != 10 {
		t.Fatalf("head = %d, want 10 (20 samples, decimation 2)", h)
	}
	if n := binary.LittleEndian.Uint32(hdr[16:]); n != 10 {
		t.Fatalf("count = %d, want 10", n)
	}
	if d := binary.LittleEndian.Uint32(hdr[20:]); d != 2 {
		t.Fatalf("decimation = %d, want 2", d)
	}
	if got := math.Float64frombits(binary.LittleEndian.Uint64(hdr[32:])); got != 19 {
		t.Fatalf("tNewest = %v, want 19", got)
	}
	// The data region must be pre-allocated in full, not grown as it fills.
	if want := int64(histHeaderSize) + histMinCapacity*histPairSize; int64(len(hdr)) != want {
		t.Fatalf("file size = %d, want the pre-allocated %d", len(hdr), want)
	}
}

func TestSanitizeHistName(t *testing.T) {
	cases := map[string]string{
		"Signal_1":       "Signal_1",
		"GAM.Out[0]":     "GAM.Out[0]",
		"a/b":            "a_b",
		"../../etc/pass": ".._.._etc_pass",
		"":               "_",
		".":              "_",
		"..":             "_",
		"with space":     "with_space",
		"nul\x00byte":    "nul_byte",
	}
	for in, want := range cases {
		if got := sanitizeHistName(in); got != want {
			t.Errorf("sanitizeHistName(%q) = %q, want %q", in, got, want)
		}
	}
}

// A producer-supplied name must never place a file outside the history dir.
func TestHistoryNameCannotEscapeDirectory(t *testing.T) {
	dir := t.TempDir()
	hw, _ := newHistoryWriter(HistoryConfig{Directory: dir})
	defer hw.close()
	hw.onSourceConfigured("../evil", []udpsprotocol.SignalInfo{
		{Name: "../../pwned", TypeCode: 8, SamplingRate: 1},
	})
	found := false
	err := filepath.Walk(dir, func(p string, info os.FileInfo, err error) error {
		if err == nil && !info.IsDir() {
			found = true
		}
		return err
	})
	if err != nil {
		t.Fatalf("walk: %v", err)
	}
	if !found {
		t.Fatal("no file created inside the history directory")
	}
	if _, err := os.Stat(filepath.Join(dir, "..", "..", "pwned.shist")); err == nil {
		t.Fatal("a file escaped the history directory")
	}
}

func TestHistCapacityFor(t *testing.T) {
	cases := []struct {
		window     float64
		rate       float64
		decim      int
		maxPts     int
		want       uint32
		wantBucket int
	}{
		// window × rate / decimation, plus the 1.25 headroom.
		{600, 1000, 1, 0, 750_000, 1},
		{600, 1000, 10, 0, 75_000, 1},
		{10, 100, 1, 0, 1250, 1},
		{600, 0.001, 1, 0, histMinCapacity, 1}, // absurdly slow → the floor
		// Absurdly fast: bounded by histMaxCapacity, and the window is bought with
		// a correspondingly absurd bucket rather than by storing less of it.
		{600, 1e9, 1, 0, 1_073_729_421, 1397},
		{math.NaN(), 1000, 1, 0, histMinCapacity, 1},
		{600, math.NaN(), 1, 0, histMinCapacity, 1},
		// A budget envelopes a fast signal without touching a slow one, and the
		// window is kept either way.
		{600, 1e6, 1, 16 << 20, 16_666_667, 90},
		{600, 1000, 1, 16 << 20, 750_000, 1},
	}
	for _, c := range cases {
		got, bucket := histCapacityFor(c.window, c.rate, c.decim, c.maxPts)
		if got != c.want || bucket != c.wantBucket {
			t.Errorf("histCapacityFor(%v, %v, %d, %d) = %d/%d, want %d/%d",
				c.window, c.rate, c.decim, c.maxPts, got, bucket, c.want, c.wantBucket)
		}
	}
}

func TestHistoryConfigDefaults(t *testing.T) {
	c := HistoryConfig{}.withDefaults()
	if c.WindowSec != defaultLiveWindowSec || c.Decimation != 1 || c.FlushIntervalSec != 5 || c.MinDiskFreeMB != 500 {
		t.Fatalf("defaults = %+v", c)
	}
	// A negative value is the explicit "no disk guard", so it must survive
	// defaulting rather than being turned back into 500.
	if got := (HistoryConfig{MinDiskFreeMB: -1}).withDefaults().MinDiskFreeMB; got != -1 {
		t.Fatalf("MinDiskFreeMB = %d, want the -1 that disables the guard", got)
	}
}

func TestHistoryWritePausedWhenDiskLow(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{}, 100)
	hw.diskLow = true
	hw.write(key, []float64{1, 2, 3}, []float64{1, 2, 3})
	if hw.files[key].count != 0 {
		t.Fatalf("count = %d, want 0 while the disk guard is tripped", hw.files[key].count)
	}
	hw.diskLow = false
	hw.write(key, []float64{1, 2, 3}, []float64{1, 2, 3})
	if hw.files[key].count != 3 {
		t.Fatalf("count = %d, want 3 once writing resumes", hw.files[key].count)
	}
}

func TestHistoryReadRangeRespectsMaxOut(t *testing.T) {
	// A window wide enough that the whole ramp is still on disk when it is read.
	hw, key := newTestHistory(t, HistoryConfig{WindowSec: 50}, 100)
	ts, vs := ramp(0, 0.01, 5000)
	hw.write(key, ts, vs)
	rt, rv := hw.readRange(key, -1, 1e9, 100)
	if len(rt) != 100 || len(rv) != 100 {
		t.Fatalf("read %d/%d points, want the 100 cap", len(rt), len(rv))
	}
}

func TestHistoryReadRangeSpansWholeRange(t *testing.T) {
	// A capped read must thin the range out, not return its first maxOut
	// samples: a client asking for 100 points over 50 s and getting the first
	// second of it draws a flat line and falls back to its coarse copy.
	hw, key := newTestHistory(t, HistoryConfig{WindowSec: 50}, 100)
	ts, vs := ramp(0, 0.01, 5000)
	hw.write(key, ts, vs)

	rt, _ := hw.readRange(key, 0, 49.99, 100)
	if len(rt) == 0 {
		t.Fatal("no points read")
	}
	if got := rt[len(rt)-1] - rt[0]; got < 0.95*49.99 {
		t.Fatalf("read spans %.2f s of the 49.99 s asked; a capped read must "+
			"cover the whole range", got)
	}
}

func TestHistoryReadRangeUncappedIsExact(t *testing.T) {
	// Below the cap every sample in the range comes back, so a zoom deep enough
	// to fit is served at full resolution.
	hw, key := newTestHistory(t, HistoryConfig{WindowSec: 50}, 100)
	ts, vs := ramp(0, 0.01, 5000)
	hw.write(key, ts, vs)

	rt, rv := hw.readRange(key, 1, 1.99, 1000)
	if len(rt) != 100 {
		t.Fatalf("read %d points, want the 100 samples in [1, 1.99]", len(rt))
	}
	if rv[0] != 100 || rv[len(rv)-1] != 199 {
		t.Fatalf("values %.0f..%.0f, want 100..199", rv[0], rv[len(rv)-1])
	}
}

// The capture copy is what makes a trigger window zoomable long after the
// circular archive has wrapped over it.
func TestCaptureRangeOutlivesTheArchive(t *testing.T) {
	hw, key := newTestHistory(t, HistoryConfig{}, 0.001) // floor capacity: 1000
	hf := hw.files[key]
	if hf.capacity != histMinCapacity {
		t.Fatalf("capacity = %d, want the %d floor", hf.capacity, histMinCapacity)
	}
	ts, vs := ramp(0, 1, 1000) // t = 0..999, exactly full
	hw.write(key, ts, vs)

	hw.captureRange(500, 600)

	// Wrap the archive right over the captured window.
	ts2, vs2 := ramp(1000, 1, 1000)
	hw.write(key, ts2, vs2)
	if hf.tOldest != 1000 || hf.tNewest != 1999 {
		t.Fatalf("archive holds [%v, %v], want [1000, 1999]: capturing must not "+
			"stop or divert the archive", hf.tOldest, hf.tNewest)
	}

	rt, rv := hw.readRange(key, 500, 600, 1000)
	if len(rt) != 101 {
		t.Fatalf("read %d captured samples in [500, 600], want 101", len(rt))
	}
	if rv[0] != 500 || rv[len(rv)-1] != 600 {
		t.Fatalf("captured values %.0f..%.0f, want 500..600", rv[0], rv[len(rv)-1])
	}

	// A range the capture does not hold is still answered by the archive.
	if at, _ := hw.readRange(key, 1500, 1600, 1000); len(at) != 101 {
		t.Fatalf("read %d archived samples in [1500, 1600], want 101", len(at))
	}

	// The next capture replaces the last one, and only then.
	hw.captureRange(1500, 1600)
	if ct, _ := hw.readRange(key, 500, 600, 1000); len(ct) != 0 {
		t.Fatalf("read %d samples of a replaced capture, want 0", len(ct))
	}
}

// Delivering a capture copies its window out of the archive, and the archive
// keeps rolling so the next capture's pre-trigger window is there when it fires.
func TestTriggerCaptureCopiesWindowToDisk(t *testing.T) {
	h := NewHub()
	if err := h.EnableHistory(HistoryConfig{
		Directory: t.TempDir(), WindowSec: 36, MinDiskFreeMB: -1,
	}); err != nil {
		t.Fatalf("EnableHistory: %v", err)
	}
	t.Cleanup(h.CloseHistory)
	h.hist.onSourceConfigured("s1", []udpsprotocol.SignalInfo{
		{Name: "sig", TypeCode: 8, SamplingRate: 1000},
	})

	h.rings["s1:sig"] = newSigRing(10000)
	h.trigger.SetConfig(trigConfig{signalKey: "s1:sig", edge: "rising", threshold: 0,
		windowSec: 1, prePercent: 20, mode: "single"})
	h.trigger.Arm()
	// Cross the threshold, then cover the post-trigger window so the capture
	// comes due on the next tick.
	h.ingest("s1:sig", 1, []float64{5.0, 5.001}, []float64{-1, 1})
	h.ingest("s1:sig", 1, []float64{6.0}, []float64{1})

	h.triggerTick()
	if h.trigger.State() != trigTriggered {
		t.Fatalf("state = %q, want triggered", h.trigger.State())
	}
	cf := h.hist.captures["s1:sig"]
	if cf == nil {
		t.Fatal("capture delivered but its window was not copied to disk")
	}
	// The window is [trigTime-0.2, trigTime+0.8] around the 5.001 crossing, so
	// the sample at 6.0 falls outside it.
	if cf.count != 2 || cf.tOldest != 5.0 || cf.tNewest != 5.001 {
		t.Fatalf("capture holds %d samples in [%v, %v], want 2 in [5, 5.001]",
			cf.count, cf.tOldest, cf.tNewest)
	}

	// Copying the window leaves the archive rolling, so the next capture's
	// pre-trigger window — written before its trigger fires — is there for it.
	h.ingest("s1:sig", 1, []float64{7.0}, []float64{1})
	if got := h.hist.files["s1:sig"].count; got != 4 {
		t.Fatalf("archived %d samples, want 4: capturing must not stop writing", got)
	}

	// Rearming does not discard the capture: it stays on screen until the next
	// trigger replaces it.
	h.trigger.Arm()
	h.triggerTick()
	if h.hist.captures["s1:sig"] != cf {
		t.Fatal("rearming discarded the capture the client is still showing")
	}
}

func TestHistSearch(t *testing.T) {
	vals := []float64{0, 1, 2, 3, 4, 5}
	at := func(i uint32) float64 { return vals[i] }
	if got := histSearch(0, 6, func(i uint32) bool { return at(i) < 3 }); got != 3 {
		t.Fatalf("lower bound = %d, want 3", got)
	}
	if got := histSearch(0, 6, func(i uint32) bool { return at(i) <= 3 }); got != 4 {
		t.Fatalf("upper bound = %d, want 4", got)
	}
	if got := histSearch(0, 6, func(i uint32) bool { return at(i) < -1 }); got != 0 {
		t.Fatalf("all-false = %d, want 0", got)
	}
	if got := histSearch(0, 6, func(i uint32) bool { return at(i) < 100 }); got != 6 {
		t.Fatalf("all-true = %d, want 6", got)
	}
}