// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#pragma once

#include "channel.hpp"
#include "config.hpp"
#include "event_series.hpp"
#include "trigger.hpp"

#include <plot/data/ring_buffer.hpp>

#include <algorithm>
#include <array>
#include <atomic>
#include <unordered_map>
#include <cstdint>
#include <memory>
#include <mutex>
#include <string>

#include "math_channel.hpp"
#include <vector>

namespace udpscope {

/// `SignalInfo::time_mode` values. They equal the wire protocol's `UDPS_TIME_*`
/// (`udps_source.cpp` static_asserts it), but the model must not name a source's own header.
constexpr std::uint8_t kTimePacket = 0;
constexpr std::uint8_t kTimeFullArray = 1;
constexpr std::uint8_t kTimeFirstSample = 2;
constexpr std::uint8_t kTimeLastSample = 3;

/// One streamed signal, as the CONFIG packet described it.
struct SignalInfo {
  std::string name;
  /// True for the entries the model *synthesised* for a math channel: the stream never sends them,
  /// the producer fills their samples by evaluating the graph. The flag is what lets a CONFIG
  /// update tell them from the stream's own signals and put them back afterwards.
  bool math = false;
  std::string unit;
  std::uint8_t type_code = 0;
  std::uint8_t quant_type = 0;
  std::uint32_t rows = 1;
  std::uint32_t cols = 1;
  double range_min = 0.0;
  double range_max = 1.0;
  std::uint8_t time_mode = 0;
  double sampling_rate = 0.0;
  std::uint32_t time_signal_idx = 0xFFFFFFFFu;

  std::uint32_t elements() const noexcept {
    const std::uint32_t n = rows * cols;
    return n == 0 ? 1u : n;
  }
};

/// One plottable signal element: a signal with a single element, or one element
/// of a vector/matrix signal. The unit of enable/disable is the *signal*; the
/// unit of acquisition and plotting is the channel.
struct ChannelInfo {
  std::size_t signal = 0;
  bool enabled = false;   ///< whether its signal is enabled
  double gain = 1.0;      ///< calibration, applied as sample * gain + offset
  double offset = 0.0;

  std::string name;       ///< the signal's name
  /// True for a channel the producer *derives* rather than reads. The ingest must skip it: it has
  /// no packet signal, so reading one gives zero -- and pushing that zero into a ring whose time
  /// ring advances once per slot is what made a math channel alternate between its value and 0, and
  /// then drift off the time ring entirely once the buffer was full.
  bool math = false;

  std::string unit;   ///< the signal's unit, or a math channel's own

  /// Samples of this channel per DATA packet (`elements()` of the signal). A
  /// UDPStreamer signal is one channel: its columns are consecutive samples of the
  /// same signal, not separate channels.
  std::uint32_t elements = 1;

  /// Sample interval of this channel's own cadence, in seconds. 0 when unknown.
  double dt = 0.0;
};

/// How one buffer's storage is divided between the shared time base and the
/// channels (spec item 2: "split in N+1", the +1 being the time base).
///
/// The split is by *time span*, not by bytes: every channel's ring and the time
/// ring cover the same amount of stream time, which is what makes a shared time
/// base meaningful. When no channel is decimated (the common case) the two rules
/// coincide and every slice is the same size.
struct Partition {
  u64 time_capacity = 0;               ///< time-base slots retained
  /// Base ticks per stored slot: the decimation factor the requested window forced. 1
  /// until the window needs more samples than the budget can hold. The ingest path
  /// *averages* each group of this many samples rather than keeping one of them (see
  /// `Acquisition::on_data()`), so this is a sample rate reduction with an anti-alias
  /// filter in front of it, not a stride.
  u64 time_stride = 1;
  std::vector<u64> stride;             ///< time slots per channel sample (>= 1)
  std::vector<u64> value_capacity;     ///< samples retained per channel
  double base_dt = 1.0;                ///< seconds per base tick
  double span_s = 0.0;                 ///< stream seconds one buffer covers

  /// True when a second f64 ring parallel to the clock was planned and paid for in the
  /// byte budget. Relative mode only -- in stream mode `time()` already *is* the absolute
  /// clock and a copy would cost ~22% of the user's sample budget for nothing.
  bool abs_time = false;

  /// The samples-per-signal the user asked for, as given (not rounded). `time_capacity` is what
  /// the plan actually holds: the request rounded down to a power of two, then halved further
  /// while the rings did not fit the memory limit. These two are readings, not storage -- they are
  /// not in `same_storage_as()` -- so a re-plan that only moves them does not drop history.
  u64 requested_samples = 0;
  /// True when the memory limit, and not the power-of-two rounding, made `time_capacity` smaller than
  /// the request: the one case the UI has to explain.
  bool memory_limited = false;

  /// The bytes the rings of this plan occupy: the time ring (twice in relative mode) plus every
  /// value ring. This is the figure to quote and to budget against; a count of the user's
  /// request is not, because the rings round and the limit can clip.
  u64 bytes() const noexcept {
    u64 total = time_capacity * sizeof(f64) * (abs_time ? 2u : 1u);
    for (u64 c : value_capacity) total += c * sizeof(f32);
    return total;
  }

  /// Seconds between two time-base slots (the sample interval of every channel).
  double slot_dt() const noexcept { return base_dt * static_cast<double>(time_stride); }
  /// The rate samples are actually stored at, in samples per second.
  double stored_rate() const noexcept {
    const double dt = slot_dt();
    return dt > 0.0 ? 1.0 / dt : 0.0;
  }
  /// The anti-alias filter's corner, in Hz: half the stored rate, which is the new
  /// Nyquist. The boxcar the ingest path averages with has its first null at the stored
  /// rate itself, so everything that would fold back is attenuated.
  double antialias_hz() const noexcept { return 0.5 * stored_rate(); }

  /// Two plans that describe the **same storage**: the same rings, of the same lengths, at
  /// the same decimation, over the same clock. This is the test `rebuild_locked()` uses to
  /// decide whether it can clear the buffer it has instead of allocating another one.
  ///
  /// `base_dt` is compared although it buys no byte, because a reused store keeps the
  /// `Partition` *it* was built with and hands `slot_dt()` to every `ChannelBuffer` as the
  /// nominal grid `dx()` reports -- and `dx()` is what picks the LOD level. Two channel sets
  /// can want the same capacities off a different base tick, and reusing across that would
  /// leave every channel describing a grid the samples are not on. It is an exact compare of
  /// a value that is copied, never recomputed, from the fastest channel's `dt`.
  /// `span_s` needs no compare of its own: it is the product of the three fields above.
  bool same_storage_as(const Partition& o) const noexcept {
    return time_capacity == o.time_capacity && time_stride == o.time_stride &&
           abs_time == o.abs_time && base_dt == o.base_dt && stride == o.stride &&
           value_capacity == o.value_capacity;
  }
};

/// The ring length a request of `requested_samples` per signal buys: the largest power of two not
/// above it (at least 128). The rings are powers of two, so the user's number is a ceiling.
u64 buffer_ticks(u64 requested_samples);

/// What `ticks` slots cost for `channels` unit-stride channels: the f64 time ring (twice with the
/// absolute clock) plus one f32 ring per channel. The planner's own byte formula, for a caller
/// that has no `Partition` yet (the source form, the probes).
u64 planned_bytes(u64 ticks, std::size_t channels, bool with_abs_time);

/// Plans a partition for `channel_dt` cadences: `samples_per_signal` slots for the time ring and for
/// every channel, rounded down to a power of two, then halved while the rings (`planned_bytes()`)
/// exceed `max_bytes` -- the memory limit. Covers at least `target_span_s` seconds of stream,
/// decimating (with the ingest filter) by whatever power of two that takes. Throws
/// std::runtime_error when even 128 slots do not fit `max_bytes`. `with_abs_time` adds a second f64
/// per time slot to the budget and sets `Partition::abs_time`. `Partition::memory_limited` says
/// whether the limit clipped the request.
Partition plan_partition(const std::vector<double>& channel_dt, u64 samples_per_signal,
                         u64 max_bytes, double target_span_s, bool with_abs_time = false);

/// One of the two identical sample buffers: a shared time-base ring plus one
/// value ring and one `ChannelBuffer` per channel.
class BufferStore {
 public:
  BufferStore(const Partition& part, f64 span_s);

  /// Re-bases every channel's x (see `ChannelBuffer::set_epoch`). Used to make the
  /// axis read seconds from the trigger event.
  void set_epoch(f64 epoch) noexcept;

  /// The shared clock, in **f64**. The values are f32 and the clock is not: a stamp is
  /// `stream_time - epoch` and the epoch is fixed at the first sample ever seen, so the
  /// magnitude grows without bound over a run's life while the interval that has to be
  /// resolved stays the sample period. f32 has 24 bits of mantissa, so its ULP at 32 s is
  /// 3.8 us -- four consecutive samples of a 1 MS/s stream then share one stamp, and at
  /// 600 s uptime 61 of them do. A value's precision is a property of the signal and 24
  /// bits is plenty; a *timestamp's* is a property of how long the scope has been running.
  plot::data::RingBuffer<f64>& time() noexcept { return *time_; }
  const plot::data::RingBuffer<f64>& time() const noexcept { return *time_; }
  /// The stream's **absolute** clock, parallel to `time()` and on the same slots, or
  /// `nullptr` when the store was planned without it.
  ///
  /// There is exactly one time ring per store, shared by every channel, and relative mode
  /// fills it with counter-derived stamps -- so without this ring, selecting relative mode
  /// would silently drop absolute time from an export, the one output `todo.md` item 2 says
  /// must keep it. Allocated in relative mode only; read only by the export path.
  plot::data::RingBuffer<f64>* abs_time() noexcept { return abs_time_.get(); }
  const plot::data::RingBuffer<f64>* abs_time() const noexcept { return abs_time_.get(); }
  plot::data::RingBuffer<f32>& values(std::size_t channel) { return *values_[channel]; }
  ChannelBuffer& channel(std::size_t channel) { return *channels_[channel]; }
  /// Shared handle to a channel view: the caller keeps the storage alive even if
  /// the model rebuilds.
  std::shared_ptr<ChannelBuffer> shared_channel(std::size_t channel) {
    return channels_[channel];
  }
  const Partition& partition() const noexcept { return part_; }

  /// Drops every sample and leaves a store indistinguishable from a freshly built one with
  /// the same `Partition` -- the state `rebuild_locked()` wants when the plan did not change.
  ///
  /// Why it exists: constructing a store *is* its cost. A ring is a
  /// `vector<std::atomic<T>>`, which is value-initialised, so building one touches every
  /// byte of the planned buffer (`--buffer-samples`) -- about 300 ms per GB, nearly all of it first-touch
  /// page faults. Clearing is a count store per ring. Most structural changes (a
  /// calibration, a re-enable of what was just disabled, going back to live) re-plan to the
  /// *same* partition, and paying a second full allocation for them is what made a large
  /// buffer take ten seconds to start and an event take seconds to leave.
  ///
  /// It must leave **nothing** of the old history behind, in either copy: the rings are
  /// emptied and every `ChannelBuffer` is `invalidate()`d, which drops the published window
  /// and the pyramid built over it. The epoch goes back to 0 for the same reason -- a fresh
  /// store has none, and a reused one keeping the old event's epoch would re-base the next
  /// axis against a trigger that no longer exists.
  ///
  /// Caller's thread: the consumer, under `ScopeModel::mutex()` -- the same lock the
  /// producer's pushes hold. `RingBuffer::clear()` is not safe against a concurrent push.
  void reset() noexcept;

  /// Consumer: publishes every channel's new samples.
  void acquire_all();

 private:
  Partition part_;
  std::shared_ptr<plot::data::RingBuffer<f64>> time_;
  std::shared_ptr<plot::data::RingBuffer<f64>> abs_time_;
  std::vector<std::shared_ptr<plot::data::RingBuffer<f32>>> values_;
  std::vector<std::shared_ptr<ChannelBuffer>> channels_;
};

/// One reserved, invalidated span: samples the counter says exist and the stream has not
/// delivered. Its stored slots hold NaN, so a reader needs no special case -- the table is
/// what lets a *late* packet find the slots it belongs to, and what the band draws.
struct Hole {
  f64 x_begin = 0.0;          ///< stream seconds of the hole's first slot
  f64 x_end = 0.0;            ///< stream seconds of the hole's last slot
  /// Absolute **time-ring** slot of the first reserved slot, and the count of them. The
  /// time ring is the one clock every channel shares, so this is the only slot address
  /// that is unambiguous: per-channel value rings run at `time_stride * stride[i]`, which
  /// differs between channels. A consumer that needs a *channel's* slots derives them from
  /// `tick_begin` and that channel's stride rather than reusing these.
  u64 slot_begin = 0;
  u64 slot_count = 0;
  u64 tick_begin = 0;         ///< absolute base tick of the hole's first raw sample
  std::uint32_t counter_first = 0;  ///< counter of the first missing packet
  std::uint32_t packets = 0;  ///< packets the hole is worth
  /// How many of those packets have since arrived late and been written in. The band's two tints
  /// are this field against `packets`: a hole still missing data is shaded solidly and a fully
  /// repaired one faintly, so a reader can tell "there is no data here" from "there was none when
  /// this window was first drawn".
  ///
  /// Counted in *packets* and not in slots, because a packet is the unit the reservation was made
  /// in: the hole is worth exactly `packets` of them, each late arrival delivers exactly one, and
  /// the client refuses a duplicate -- so this count is exact. A slot count would not be: the
  /// channels have different strides, so "how many slots did this hole want" has one answer per
  /// channel and the repaired test would have to pick one.
  std::uint32_t filled_packets = 0;

  /// Every packet this hole was reserved for has arrived. A hole worth no packets is never
  /// repaired -- there is nothing for a late packet to deliver -- which keeps a degenerate
  /// reservation shaded rather than silently downgraded.
  bool repaired() const noexcept { return packets > 0 && filled_packets >= packets; }
};

/// Live connection and counter state, readable from the UI thread.
struct StreamStats {
  std::atomic<std::uint64_t> packets{0};
  std::atomic<std::uint64_t> frames{0};
  std::atomic<std::uint64_t> gaps{0};
  std::atomic<std::uint64_t> reconnects{0};
  std::atomic<bool> connected{false};
  std::atomic<bool> have_config{false};
  std::atomic<std::uint64_t> holes{0};       ///< reserved invalidated spans
  std::atomic<std::uint64_t> hole_slots{0};  ///< stored slots those spans cover
  /// Out-of-order packets put back in order inside the reorder window: no hole was ever
  /// seen for these.
  std::atomic<std::uint64_t> resequenced{0};
  /// Out-of-order packets delivered behind the sequence, past the window.
  std::atomic<std::uint64_t> late{0};
  /// Times the client's reorder ring was full when a packet needed a slot, so the gap was
  /// committed early. Nothing is dropped; the window was just shorter than asked for, and a
  /// non-zero reading is what says the ring is undersized for the stream.
  std::atomic<std::uint64_t> overflows{0};
  /// Packets the client still refused outright (a duplicate, or any late packet while
  /// resequencing is disabled). Counted since the beginning and displayed nowhere until
  /// now, which is why an inversion had to be inferred from `packets` and `frames`
  /// diverging.
  std::atomic<std::uint64_t> stale{0};
  /// Stored slots a late packet repaired. Declared here rather than with the backfill that
  /// writes it, because the display segment below is built once and reads it from the start:
  /// a reading of 0 beside a non-zero `holes` is exactly the honest answer until the
  /// backfill lands.
  std::atomic<std::uint64_t> backfilled{0};
  /// Reservations refused because the counter's claim was not corroborated by the arrival
  /// cadence, or because the hole it would book is bigger than the ring can hold. A counter
  /// is a claim about the wire and the arrival clock is the measurement of it: a stream whose
  /// counter does not count packets (a sample counter, say) reports a forward jump on every
  /// datagram, and believing it booked a NaN hole per packet -- a band over the whole window
  /// filled with samples that were never missing, which the backfill cannot repair because no
  /// packet was ever lost. Counted here because it is otherwise *invisible*: the status bar's
  /// `gaps` climbs either way, and only this field says the gap was refused.
  std::atomic<std::uint64_t> gaps_refused{0};
};

/// What the plot is showing, which is one decision made once per frame: the bind, the
/// export and the status wording all have to agree, and the producer can latch between
/// any two of them.
///
/// `Capturing` is the live-updating capture: an edge has been accepted (so the pre
/// window was full) and the post window is still filling, and the plot follows the live
/// ring on an edge-relative axis until the snapshot lands. It is a deliberate departure
/// from SPECS:37 ("not refreshed till next event") -- see README.
enum class DisplayMode {
  Live,       ///< no trigger: the live ring, stream clock
  Capturing,  ///< an edge was accepted: the live ring, edge-relative clock
  Held,       ///< the frozen snapshot of the last event
  Waiting,    ///< armed, nothing captured yet: blank
  /// The user paused: the acquisition thread keeps writing the live ring, and the plot draws
  /// the view that was frozen when the pause began -- a snapshot, exactly as `Held` is, so
  /// resuming costs no data.
  Paused,
};

/// The application's data model: the stream description, the two sample buffers,
/// the trigger configuration and the counters. Structural changes (a CONFIG
/// update, an enable/disable, a calibration) rebuild the buffers, so all mutating
/// access is serialised on one mutex; the acquisition thread takes it once per
/// DATA frame.
class ScopeModel {
 public:
  explicit ScopeModel(const Config& cfg);

  const Config& config() const noexcept { return cfg_; }

  // --- stream description ---------------------------------------------------
  /// Takes a new CONFIG: rebuilds the signal and channel lists and the two
  /// buffers. Clears the trigger's channel selection if the stream changed.
  void set_signals(std::vector<SignalInfo> signals);

  /// The user's math channels, as configured. Read it under the lock, like the other accessors
  /// here -- the UI copies what it needs while it holds it.
  std::vector<MathChannel> math_channels() const { return math_; }
  /// Replaces the math channels. A *structural* change, exactly like a CONFIG or an
  /// enable/disable: the enabled ones are re-appended to the signal list, their channels are
  /// rebuilt and their runtimes re-prepared for the new stored rate. **Takes the lock itself** --
  /// do not call it holding `mutex()`, which is not recursive.
  void set_math_channels(std::vector<MathChannel> channels);

  /// One stored sample's worth of derived values, called by the producer right after it has pushed
  /// the streamed channels. `sample[c]` is channel `c`'s value for this slot; the math channels are
  /// evaluated from those, in order, so a math channel may use another one.
  void push_math_sample(const std::vector<f32>& sample);

  /// Applies the graphs to one stored slot. `write_mask`, when given, names the derived channels whose
  /// rings may be written -- `nullptr` writes all of them, which is what the producer does; a recompute
  /// passes the channels a graph change actually affects, so the others keep the samples they have.
  void math_apply_locked(const std::vector<f32>& sample, const std::vector<bool>* write_mask, bool timed);

  /// Recomputes the derived channels in `affected` from the samples the store already holds: their rings are
  /// rebuilt from scratch, and the channels a chart change does *not* touch are left exactly as they were.
  void recompute_math_locked(const std::vector<bool>& affected);

  /// How many slots the store currently holds. A file view reads it to tell a structural change that
  /// dropped the buffer (0) from one that could not have (see `reload_file_view()` in ui.cpp).
  std::uint64_t stored_slots();

  /// Pushes one stored slot of a **file** load -- the values, the shared clock and the absolute
  /// clock -- and then the derived channels, in the producer's own order and under the model's own
  /// lock. The loader cannot simply hold the lock and push into `live()` itself: every setter it
  /// needs (`set_signals`, `set_signal_enabled`, ...) takes that same non-recursive lock, so the
  /// outer one deadlocks the first call.
  void push_loaded_slot(const std::vector<f32>& values, f64 t, f64 abs_t);
  /// What the derived channels cost the *producer*: every stored slot that went through
  /// `push_math_sample()`, and the nanoseconds taken by the `math_timed()` of them that were
  /// measured. The producer owns the budget between two DATA packets, so this cost is invisible in
  /// the frame mean and surfaces only as dropped packets -- which is why it is counted rather than
  /// inferred. It is *sampled*, so that the instrumentation cannot itself cause what it is looking
  /// for: read `math_cost_ns() / math_timed()` as the per-slot figure, scaled by `math_slots()`.
  std::uint64_t math_slots() const noexcept {
    return math_slots_.load(std::memory_order_relaxed);
  }
  std::uint64_t math_timed() const noexcept {
    return math_timed_.load(std::memory_order_relaxed);
  }
  std::uint64_t math_cost_ns() const noexcept {
    return math_cost_ns_.load(std::memory_order_relaxed);
  }
  /// What freezing a view costs, and *which thread it costs it to*. The two publishes are stalls on
  /// opposite sides of `mutex()` and so are counted apart: `publish_event()` runs on the **producer**
  /// with the lock already held, so its cost lands on the UI as a long tick (the documented `tick`
  /// max); `publish_pause_view()` runs on the **UI** thread and takes the lock, so its cost lands on
  /// the producer and can only surface as `gaps`. Neither is attributable in a frame mean, which is
  /// why they are counted rather than inferred. Each counter is written by one thread and read by the
  /// dump, so a relaxed store of a value already in hand is enough.
  ///
  /// This is the number to judge a second live ring against: freezing a whole `BufferStore` would
  /// make both figures zero, since a `ChannelBuffer` maintains its pyramid incrementally and there
  /// would be nothing to copy -- but it costs a second budget's worth of memory and a second
  /// per-sample write on the producer, so it is worth paying only if the worst figures here are
  /// large enough to matter.
  std::uint64_t publish_event_count() const noexcept {
    return publish_event_count_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_event_us_total() const noexcept {
    return publish_event_us_total_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_event_us_worst() const noexcept {
    return publish_event_us_worst_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_pause_count() const noexcept {
    return publish_pause_count_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_pause_us_total() const noexcept {
    return publish_pause_us_total_.load(std::memory_order_relaxed);
  }
  std::uint64_t publish_pause_us_worst() const noexcept {
    return publish_pause_us_worst_.load(std::memory_order_relaxed);
  }
  const std::vector<SignalInfo>& signals() const noexcept { return signals_; }
  const std::vector<ChannelInfo>& channels() const noexcept { return channels_; }

  bool signal_enabled(std::size_t signal) const;
  /// True when the signal is the stream's time base, i.e. when some other signal
  /// declares it via `time_signal_idx`. A signal used as a time base builds the x
  /// axis and is not a plottable channel.
  static bool is_time_signal(const std::vector<SignalInfo>& signals,
                             std::size_t signal);
  void set_signal_enabled(std::size_t signal, bool on);
  /// The same switch thrown for several signals as **one** structural change.
  ///
  /// Enabling a signal changes the channel count, so it re-plans the partition and the new
  /// plan really is a different buffer -- there is no reusing the old store for it (see
  /// `BufferStore::reset()`). A caller that enables n signals one at a time therefore pays n
  /// full allocations to reach a state one allocation describes, and every one of them drops
  /// the history the one before it started collecting. That is what a restored session, a
  /// loaded waveform and an event group all do, and at a 1 GB buffer it is seconds of a
  /// blank window: the plots are bound to a store that is replaced again before it holds a
  /// screenful. Indices the model has no signal for are ignored.
  void set_signals_enabled(const std::vector<std::size_t>& signals, bool on);
  /// Stores the calibration for a signal's *name* and applies it to that signal's
  /// live channels. Keyed by name, not index: a CONFIG renumbers and an
  /// enable/disable rebuilds the channel list, and `rebuild_channels()` re-applies
  /// from here -- so a calibration set before a signal was enabled, or surviving a
  /// CONFIG, is not silently dropped.
  void set_calibration(std::size_t signal, double gain, double offset);

  /// The calibration that *will* be applied to that signal, as `{gain, offset}` -- the identity
  /// `{1, 0}` for a signal nobody calibrated. The read counterpart of `set_calibration()`, and read
  /// from the model's own record rather than from the channels: a signal that is not acquired has
  /// no channel to read it from, and the record is what a dialog must show.
  std::pair<double, double> calibration(std::size_t signal);

  /// Renames the unit a streamed signal is shown, read and exported in. Stored by the signal's *name*
  /// like the calibration, and re-applied to every later CONFIG, so a rename survives a reconnect.
  /// An empty `unit` removes the override and restores the unit the stream declared. A math channel's
  /// unit belongs to its graph and is ignored here. Bumps `generation()` when it changed something,
  /// which is what makes the panels re-read the signal list.
  void set_signal_unit(std::size_t signal, const std::string& unit);
  /// The unit the stream declared for that signal, whatever the user renamed it to.
  std::string declared_unit(std::size_t signal);
  /// The renames in force, `{name, unit}`, for the session file.
  std::vector<std::pair<std::string, std::string>> unit_overrides();

  std::size_t channel_count() const noexcept { return channels_.size(); }
  const Partition& partition() const noexcept { return part_; }

  /// The channel's sample interval (0 when the stream did not say).
  double channel_dt(std::size_t channel) const;

  // --- buffers --------------------------------------------------------------
  /// The one sample store: the producer always writes it, and the plot draws from
  /// it whenever it is not showing a frozen event. Always-filling is what
  /// guarantees a full pre-trigger history on the *first* latch after the trigger
  /// is armed.
  BufferStore& live();

  /// Records a reserved hole. Caller holds `mutex()` (the producer already does). The table
  /// is capped at `kMaxHoles`; past that the oldest is dropped, because an unbounded list of
  /// gaps is a leak and the oldest gap is the one furthest from being backfilled.
  static constexpr std::size_t kMaxHoles = 64;
  void note_hole(const Hole& h);
  const std::vector<Hole>& holes() const noexcept { return holes_; }
  /// The producer's view of the same table. `Acquisition::backfill_from()` marks a hole's packet
  /// filled as it repairs it, and that mark is what the band's fainter tint reads. Caller holds
  /// `mutex()` -- the same requirement `note_hole()` above states, and the producer already does.
  std::vector<Hole>& holes() noexcept { return holes_; }
  /// Drops holes whose time-ring slots the ring has already discarded.
  void forget_holes_before(u64 slot);

  /// Builds the captured event from the slice of the live buffer covering
  /// `[edge - pre_s, edge + post_s]` in stream seconds, and publishes it atomically.
  /// The stamps are re-based so the edge sits at x = 0. The plot binds to this
  /// immutable snapshot -- it is the "frozen buffer" -- while the live ring keeps
  /// filling behind it.
  ///
  /// The cut is by time, not by a sample count: the window the trigger panel asks
  /// for is in seconds and the time ring stores seconds, so any count in between
  /// needs a cadence estimate -- which is exactly what used to size the capture
  /// wrong.
  void publish_event(double pre_s, double post_s, double edge);

  /// The live buffer's absolute index range covering `[lo_s, hi_s]` in stream seconds,
  /// clamped to what the time ring and *every* value ring still hold. One range for the
  /// time ring and all the channels, which is what keeps a cut index-aligned by
  /// construction; the bounds are compared in the ring's own type, which is now the
  /// `f64` they arrive in, so a bound that lands on a sample includes it exactly.
  ///
  /// Caller holds `mutex()` (the acquisition thread already does when it latches).
  void live_window_locked(double lo_s, double hi_s, u64& first, u64& last);
  /// The frozen view taken when the user paused, or null while running. A pause is a
  /// snapshot of the *visible window* in exactly the sense a trigger event is -- an immutable
  /// series the plot binds to -- and both are anchored the same way: the stamps are the
  /// coordinates the axis was already reading, with `x_offset` recording which instant that
  /// x = 0 was. So the axis does not move when the pause begins.
  std::shared_ptr<const EventSnapshot> pause_view() const noexcept {
    return pause_view_.load(std::memory_order_acquire);
  }
  /// Snapshots the live ring over `[lo_s, hi_s]` (stream seconds, i.e. the display window put
  /// back on the stream's clock) as the pause view, rebased onto `x_offset` so its stamps are
  /// the x the axis is already showing. The acquisition thread keeps writing, so this is a
  /// copy: what is frozen is the picture, not the data.
  void publish_pause_view(double lo_s, double hi_s, double x_offset);
  void clear_pause_view() noexcept { pause_view_.store(nullptr, std::memory_order_release); }

  /// Whether the user paused acquisition. Lives here rather than on `Acquisition` because the
  /// bind, the measurement cut and the export all have to agree about what is on screen.
  bool paused() const noexcept { return paused_.load(std::memory_order_relaxed); }
  void set_paused(bool on) noexcept { paused_.store(on, std::memory_order_relaxed); }

  /// The snapshot the plot should draw from, if any: the pause's frozen view while acquisition
  /// is paused, else the trigger's held event. One accessor, so the bind, the measurements and
  /// the export cannot disagree about which picture is on screen.
  std::shared_ptr<const EventSnapshot> display_snapshot() const noexcept {
    if (paused()) return pause_view();
    return display_mode() == DisplayMode::Held ? event() : nullptr;
  }

  /// The event on screen, or null before the first latch.
  std::shared_ptr<const EventSnapshot> event() const noexcept {
    return event_.load(std::memory_order_acquire);
  }
  /// How many snapshots `publish_event()` has produced. Not interchangeable with
  /// `trigger_latches()`: a Disarm during a capture publishes what the capture got to
  /// *without* an event having fired, so this counts "there is a new picture to frame"
  /// while the latch count keeps answering "how many events fired".
  std::uint64_t event_seq() const noexcept { return event_seq_.load(std::memory_order_acquire); }

  // --- persistence ----------------------------------------------------------
  /// The acquisitions retained for the persistence display, newest first: `persist_event(1)` is
  /// the event one latch back, `(2)` the one before that. Null for an age nobody kept.
  ///
  /// Read **without a lock**, like `event()`: the producer publishes these in `publish_event()`
  /// with the same atomic-`shared_ptr` discipline, and a snapshot is immutable and shared, so a
  /// reader holds it alive for as long as it draws it. What the *ring* would have given is not
  /// needed afterwards -- a retained snapshot is a whole window copied out of it.
  std::shared_ptr<const EventSnapshot> persist_event(std::size_t age) const noexcept {
    if (age < 1 || age > persist_.size()) return nullptr;
    return persist_[age - 1].load(std::memory_order_acquire);
  }
  /// How many acquisitions are retained right now: what the depth asks for, or fewer -- never more
  /// -- when the retention budget cut the set short. A reading, so it is counted as the slots that
  /// hold a snapshot rather than derived from the depth.
  std::size_t persist_count() const noexcept {
    return persist_count_.load(std::memory_order_relaxed);
  }
  /// What those snapshots hold in bytes, and how many acquisitions the *budget* -- as against the
  /// depth -- has dropped. Both are producer-written and read as readings.
  std::size_t persist_bytes() const noexcept {
    return persist_bytes_.load(std::memory_order_relaxed);
  }
  std::uint64_t persist_dropped() const noexcept {
    return persist_dropped_.load(std::memory_order_relaxed);
  }
  /// The bytes persistence may retain, which is the rings' own planned byte total
  /// (`Partition::bytes()`): the snapshots are copies *of* what the rings hold, so holding as much
  /// again as the buffer does is the honest ceiling for a display of it -- and it is the bound that
  /// cannot be driven into an OOM by a long window on a fast stream, whatever depth the panel asks
  /// for. Zero until a stream has planned a buffer. Read under `mutex()`, like `partition()`.
  std::size_t persist_budget_bytes() const noexcept {
    return static_cast<std::size_t>(part_.bytes());
  }

  /// Sets how many acquisitions the persistence display keeps, clamped to `kMaxPersistDepth`, and
  /// prunes the retained set to it **at once**: lowering the depth is a user asking to stop paying
  /// for the layers, and the next latch -- when the retention would otherwise catch up -- may never
  /// come on a trigger that has stopped firing.
  ///
  /// Takes `mutex()` itself; do not call it holding the lock.
  void set_persist_depth(int depth);

  /// How many acquisitions the persistence display keeps: the depth, clamped to the cap. Reading
  /// `trigger()` needs the lock, which every caller of this already holds.
  int persist_depth_locked() const noexcept {
    return std::clamp(trigger_.persist_depth, 1, kMaxPersistDepth);
  }

  /// Publishes the live buffer's new samples to its `ChannelBuffer`s.
  void acquire_all();

  // --- trigger --------------------------------------------------------------
  /// The trigger configuration. Mutating it means holding `mutex()`: the producer
  /// reads it once per frame (and writes `armed` itself when a single-shot latch
  /// disarms), so an unguarded field write can be lost or read half-updated. The
  /// panel's callbacks take the lock for the fields they touch.
  TriggerConfig& trigger() noexcept { return trigger_; }
  const TriggerConfig& trigger() const noexcept { return trigger_; }

  /// Trigger mode on/off, as both the frame tick and the producer read it: the one
  /// trigger field the tick needs and the only one written from the UI, so it has
  /// an atomic mirror and a setter that keeps the copy under the lock in step.
  bool trigger_enabled() const noexcept {
    return trigger_enabled_.load(std::memory_order_relaxed);
  }
  void set_trigger_enabled(bool on);

  /// Whether the engine is mid-capture (an edge was seen -- so the pre window was
  /// full -- and the post-trigger window is still filling). This is what puts the
  /// display in `DisplayMode::Capturing`: the plot follows the live ring, on an axis
  /// pinned to the coming event's window, until the snapshot replaces it.
  ///
  /// Release/acquire, not relaxed: `capture_edge()` is written before this is set and
  /// has to be visible to a consumer that sees the flag.
  bool trigger_pending() const noexcept {
    return trigger_pending_.load(std::memory_order_acquire);
  }
  void set_trigger_pending(bool on) noexcept {
    trigger_pending_.store(on, std::memory_order_release);
  }

  /// The stream time of the edge of the capture `trigger_pending()` refers to. Written
  /// by the producer before it raises the flag, so a consumer that sees the flag sees
  /// this too; stale once the flag clears.
  f64 capture_edge() const noexcept { return capture_edge_.load(std::memory_order_relaxed); }
  void set_capture_edge(f64 e) noexcept {
    capture_edge_.store(e, std::memory_order_relaxed);
  }

  /// What the plot should bind to this frame. Read it *once* per frame and pass the
  /// answer around: calling it twice can straddle a latch, and then the bind and the
  /// axis disagree about which clock they are on.
  DisplayMode display_mode() const noexcept {
    // A pause is the most recent instruction, so it wins over the trigger's own state: what the
    // user last asked for is a still picture.
    if (paused()) return DisplayMode::Paused;
    if (!trigger_enabled()) return DisplayMode::Live;
    if (trigger_pending()) return DisplayMode::Capturing;
    return event() != nullptr ? DisplayMode::Held : DisplayMode::Waiting;
  }

  /// The offset the **live** channels report their x with: `display_x = stream_t + epoch`.
  /// Minus the newest stored sample's stamp while streaming, so the axis reads
  /// `[-window, 0]` and the newest data is at 0; minus the trigger instant while a capture
  /// is filling, so the axis reads seconds from the event. It is 0 while a *snapshot* is on
  /// screen (a held event, a paused view), because then the live channels are bound to
  /// nothing and the snapshot's own `x_offset` is what says which clock x = 0 is --
  /// `stream_at_x0()` is the accessor that does not have to know which of the two it is.
  f64 display_epoch() const noexcept { return display_epoch_.load(std::memory_order_relaxed); }
  void set_display_epoch(f64 e) noexcept;

  /// The stream time of the display's x = 0, so `stream_t = display_x + stream_at_x0()` and
  /// `display_x = stream_t - stream_at_x0()` in **every** display mode.
  ///
  /// One accessor for the two places the offset can live: a snapshot carries its own
  /// (`EventSnapshot::x_offset`, frozen with the picture), and a live view carries it as
  /// minus the display epoch. Everything that has to cross between the two clocks goes
  /// through here -- the export's absolute column, and the pause handler putting a display
  /// window back on the ring to cut it -- rather than picking the branch itself and getting
  /// one of the four modes wrong.
  f64 stream_at_x0() const noexcept {
    if (std::shared_ptr<const EventSnapshot> snap = display_snapshot()) return snap->x_offset;
    return -display_epoch();
  }

  // --- stream state ---------------------------------------------------------
  StreamStats& stats() noexcept { return stats_; }
  const StreamStats& stats() const noexcept { return stats_; }
  /// Seconds since the epoch of the streamed time, or 0 before the first frame.
  f64 epoch() const noexcept { return epoch_; }
  void set_epoch_if_unset(f64 e) noexcept;

  /// Bumped whenever the signal or channel lists change, so the UI knows to
  /// rebuild the panels it derives from them.
  unsigned generation() const noexcept { return generation_.load(std::memory_order_acquire); }

  /// The length of the event the trigger latches, in seconds: the time window the
  /// toolbar selects. The pre-trigger percentage is a share of *this*, not of the
  /// whole ring, which is what makes the control mean something on a long buffer.
  void set_event_span(double seconds) noexcept {
    event_span_.store(seconds > 0.0 ? seconds : 1.0, std::memory_order_relaxed);
  }
  double event_span() const noexcept { return event_span_.load(std::memory_order_relaxed); }

  /// The window the buffer has to cover, which is the display window. Re-plans the
  /// partition -- and so drops the history -- only when the decimation factor it implies
  /// actually changes, so walking the window inside one decade costs nothing. Takes
  /// `mutex()` itself; do not call it holding the lock.
  /// Returns true when the buffer was rebuilt.
  bool set_target_span(double seconds);

  /// The memory limit the planner honours, in bytes. Starts as `Config::memory_limit_mb` and is the
  /// one planner input that changes while the application runs (the preferences). Atomic so the
  /// status and the probes can read it without the lock.
  std::uint64_t memory_limit_bytes() const noexcept {
    return memory_limit_bytes_.load(std::memory_order_relaxed);
  }
  /// Changes the memory limit and re-plans. The buffer is rebuilt -- and the history dropped -- only
  /// when the new limit changes the *storage* (`Partition::same_storage_as()`): a limit that still
  /// leaves room for the plan updates `Partition::memory_limited` in place and costs nothing.
  /// Returns true when the buffer was rebuilt. Takes `mutex()` itself; do not call it holding the
  /// lock. A limit too small for even 128 slots per channel is refused (returns false, keeps the old
  /// limit) and said on stderr.
  /// With `replan` false the limit is only stored, for the next plan: a file or a replayed event
  /// lives in the buffer, and re-planning it would clear what the user is looking at.
  bool set_memory_limit_bytes(std::uint64_t bytes, bool replan = true);
  double target_span() const noexcept { return target_span_.load(std::memory_order_relaxed); }

  /// Trigger state for the UI, published by the acquisition thread. Atomics
  /// rather than fields of `TriggerConfig` because the producer writes the arm
  /// state when a single-shot trigger disarms itself, and the frame tick reads it
  /// without holding the model mutex.
  std::atomic<bool>& trigger_armed() noexcept { return trigger_armed_; }
  /// Incremented every time an event is latched, so the UI can tell "waiting for
  /// an event" from "an event is on screen".
  std::atomic<std::uint64_t>& trigger_latches() noexcept { return trigger_latches_; }
  /// The stream time of the edge behind the event on screen, and whether there is
  /// one. Written by the acquisition thread on a latch and cleared by the UI when
  /// the event stops meaning anything (arming again, or leaving trigger mode).
  std::atomic<double>& trigger_time() noexcept { return trigger_time_; }
  std::atomic<bool>& trigger_time_valid() noexcept { return trigger_time_valid_; }

  /// Serialises a structural change against the acquisition thread.
  std::mutex& mutex() noexcept { return mutex_; }

  /// Rebuilds buffers from the current channel list. Caller holds `mutex()`.
  void rebuild_locked();

  /// Counts buffer re-plans: every `rebuild_locked()` bumps it. The producer caches the value
  /// it last saw and resets the state tied to the old store when it moves (see
  /// `Acquisition::on_data`). Not an atomic on purpose: `rebuild_locked()` only runs under
  /// `mutex()` and the producer holds that lock for all of `on_data`, so a plain read under the
  /// lock is correct, and a reconnect with an identical CONFIG is a re-plan this sees and the
  /// stride/channel-count comparison does not.
  /// A restarted client is a **new stream**, and the retained history belongs to the old one: its clock may
  /// sit behind the stamps the previous stream wrote, and the producer's monotonic guard would then pin
  /// every later packet flat at the old value -- the `ring=[x..x]` a display that never moves shows.
  /// `reset()` is what the data-path invariants require of a rebuild (nothing of the old history behind),
  /// and `plan_generation_` moving is what tells the producer to drop the state tied to the old store.
  void reset_stream() {
    std::lock_guard<std::mutex> lock(mutex_);
    // Null before the first CONFIG has planned anything: `Acquisition::start()` calls this, and at start-up
    // there is no store yet. Dereferencing it there was a SIGSEGV with no output at all.
    if (store_ != nullptr) store_->reset();
    ++plan_generation_;
    ++generation_;
  }

  unsigned plan_generation() const noexcept { return plan_generation_; }

 private:
  void rebuild_channels();
  /// One signal's enable bookkeeping -- `enabled_` and the math channel's own flag -- with no
  /// re-plan. Returns whether the state actually moved. The two halves of
  /// `set_signal_enabled()` are split so that a batch can do n of these and one of the
  /// rebuild below.
  bool enable_locked(std::size_t signal, bool on);
  /// The re-plan an enable needs: carry the trigger's channel across the renumbering,
  /// rebuild the channels and the buffers, and publish the new generation. Caller holds
  /// `mutex()` and has already moved `enabled_`.
  void rebuild_enabled_locked();
  /// Re-appends one synthetic signal per math channel, after the stream's own. Called under the
  /// lock by both `set_signals()` (a CONFIG) and `set_math_channels()` (an edit).
  void append_math_signals_locked();
  /// Rebuilds the math runtimes for the *current* plan: the biquad coefficients follow the stored
  /// sample interval, and their state is reset with them. Called at the end of `rebuild_locked()`,
  /// so a window change that moves the decimation re-prepares the filters exactly as an edit does.
  void prepare_math_locked();

  /// Plans the partition for the current channels at `seconds` of window with the current sample
  /// budget and memory limit. The one place the planner's inputs are assembled, so the re-plan on a
  /// window change, a structural change and a memory-limit change cannot disagree. Throws what
  /// `plan_partition()` throws. Caller holds `mutex()`.
  Partition plan_locked(double seconds) const;

  Config cfg_;
  std::atomic<std::uint64_t> memory_limit_bytes_{0};
  std::vector<SignalInfo> signals_;
  std::vector<ChannelInfo> channels_;
  // --- math channels --------------------------------------------------------
  std::vector<MathChannel> math_;
  std::vector<MathRuntime> math_rt_;
  /// The channel index each math channel was given by `rebuild_channels()`, -1 while it has none
  /// (disabled, or before the first rebuild).
  std::vector<int> math_channel_;
  /// The names of the synthetic signal entries, so they can be lifted out of `signals_`/`enabled_`
  /// before the stream's own are re-appended -- a CONFIG must not leave a stale derived channel
  /// behind, and must not drop a live one either.
  std::vector<std::string> math_signal_names_;
  /// The producer's scratch for one slot's channel values, derived ones appended as they are
  /// computed. A member rather than a local because `push_math_sample()` runs once per *stored
  /// slot* -- up to the wire rate -- and a local `std::vector` is a heap allocation and free per
  /// slot. `assign()` reuses the capacity, so after the first slot it is a memcpy.
  std::vector<f32> math_values_;
  /// The counters' own copies, written and read by the producer and nobody else, so each publish
  /// below is a relaxed **store** of a value already in hand rather than a `fetch_add`. The
  /// read-modify-write is a locked instruction and this is the per-slot path: one `lock xadd` costs
  /// about what a two-node graph's whole arithmetic does, so counting the slots cost more than the
  /// work it was counting.
  std::uint64_t math_slot_count_ = 0;
  std::uint64_t math_timed_count_ = 0;
  std::uint64_t math_cost_count_ = 0;
  /// When the slot being timed began.
  std::uint64_t math_cost_t0_ = 0;
  std::atomic<std::uint64_t> math_slots_{0};
  std::atomic<std::uint64_t> math_timed_{0};
  std::atomic<std::uint64_t> math_cost_ns_{0};
  /// See `publish_event_count()`. Single-writer each: the first three are the producer's, the last
  /// three the UI thread's.
  std::atomic<std::uint64_t> publish_event_count_{0};
  std::atomic<std::uint64_t> publish_event_us_total_{0};
  std::atomic<std::uint64_t> publish_event_us_worst_{0};
  std::atomic<std::uint64_t> publish_pause_count_{0};
  std::atomic<std::uint64_t> publish_pause_us_total_{0};
  std::atomic<std::uint64_t> publish_pause_us_worst_{0};
  /// Fold one publish's elapsed microseconds into a counter triple. Called by the one thread that
  /// owns the triple, so the load/store pairs need no read-modify-write.
  static void note_publish(std::atomic<std::uint64_t>& count, std::atomic<std::uint64_t>& total,
                           std::atomic<std::uint64_t>& worst, std::uint64_t us) noexcept {
    count.store(count.load(std::memory_order_relaxed) + 1, std::memory_order_relaxed);
    total.store(total.load(std::memory_order_relaxed) + us, std::memory_order_relaxed);
    if (us > worst.load(std::memory_order_relaxed)) worst.store(us, std::memory_order_relaxed);
  }
  std::vector<bool> enabled_;

  Partition part_;
  std::unique_ptr<BufferStore> store_;
  std::vector<Hole> holes_;
  /// See `plan_generation()`. Written and read under `mutex()` only.
  unsigned plan_generation_ = 0;
  TriggerConfig trigger_{};
  StreamStats stats_{};
  f64 epoch_ = 0.0;
  std::atomic<double> event_span_{1.0};
  /// The window the buffer is planned to cover. Read by `rebuild_locked()`, which a
  /// CONFIG can reach from the producer thread, so it is an atomic rather than a plain
  /// field.
  std::atomic<double> target_span_{0.0};
  std::atomic<bool> trigger_pending_{false};
  std::atomic<double> capture_edge_{0.0};
  std::atomic<std::shared_ptr<const EventSnapshot>> event_;  ///< the event on screen
  std::atomic<std::shared_ptr<const EventSnapshot>> pause_view_;  ///< the view frozen at pause
  /// The retention behind `persist_event()`: one slot per *previous* acquisition, newest first, so
  /// slot 0 is the event one latch back. `kMaxPersistDepth - 1` of them, because the acquisition on
  /// screen is `event_` itself and is not retained a second time. Written by the producer in
  /// `publish_event()` and by `set_persist_depth()`, both under `mutex()`; read without it.
  std::array<std::atomic<std::shared_ptr<const EventSnapshot>>, kMaxPersistDepth - 1> persist_{};
  /// The readings `persist_count()`/`persist_bytes()`/`persist_dropped()` answer with, maintained
  /// where the slots are written.
  std::atomic<std::size_t> persist_count_{0};
  std::atomic<std::size_t> persist_bytes_{0};
  std::atomic<std::uint64_t> persist_dropped_{0};
  /// Drops the retained snapshots past `keep` (0 empties the set) and recounts the readings.
  /// Caller holds `mutex()`.
  void prune_persist_locked(std::size_t keep);
  std::atomic<bool> paused_{false};
  /// Per-signal-name calibration (see `set_calibration`).
  std::unordered_map<std::string, std::pair<double, double>> calibration_;
  /// Per-signal-name unit renames (see `set_signal_unit`) and the units the stream declared.
  std::unordered_map<std::string, std::string> unit_override_;
  std::unordered_map<std::string, std::string> declared_unit_;
  std::atomic<std::uint64_t> event_seq_{0};
  std::atomic<double> display_epoch_{0.0};
  std::atomic<bool> trigger_armed_{false};
  std::atomic<bool> trigger_enabled_{false};
  std::atomic<std::uint64_t> trigger_latches_{0};
  std::atomic<double> trigger_time_{0.0};
  std::atomic<bool> trigger_time_valid_{false};
  std::atomic<unsigned> generation_{0};
  std::mutex mutex_;
};

}  // namespace udpscope