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

#include "acquisition.hpp"

#include <udps_client.h>

#include <algorithm>
#include <cmath>
#include <cstring>
#include <stdexcept>
#include <string>

namespace udpscope {
namespace {

SignalInfo to_info(const udps_signal_t& s) {
  SignalInfo info;
  info.name = s.name;
  info.unit = s.unit;
  info.type_code = s.type_code;
  info.quant_type = s.quant_type;
  info.rows = s.num_rows;
  info.cols = s.num_cols;
  info.range_min = s.range_min;
  info.range_max = s.range_max;
  info.time_mode = s.time_mode;
  info.sampling_rate = s.sampling_rate;
  info.time_signal_idx = s.time_signal_idx;
  return info;
}

}  // namespace

Acquisition::Acquisition(ScopeModel& model, const Config& cfg) : model_(model), cfg_(cfg) {}

Acquisition::~Acquisition() { stop(); }

void Acquisition::start() {
  udps_client_config_t c;
  udps_client_config_init(&c);
  c.server_addr = cfg_.server.c_str();
  c.server_port = cfg_.port;
  c.multicast_group = cfg_.multicast.empty() ? nullptr : cfg_.multicast.c_str();
  c.interface_addr = cfg_.interface_addr.empty() ? nullptr : cfg_.interface_addr.c_str();
  c.recv_buffer_bytes = static_cast<std::uint32_t>(cfg_.recv_buffer_bytes);
  c.silence_timeout_s = 2.0;
  c.reconnect_delay_s = 0.5;
  c.keepalive_interval_s = 1.0;

  client_ = udps_client_create(&c);
  if (client_ == nullptr) {
    throw std::runtime_error("could not create the UDPS client for " + cfg_.server + ":" +
                             std::to_string(cfg_.port));
  }
  udps_client_set_callbacks(client_, &Acquisition::config_trampoline, &Acquisition::data_trampoline,
                            &Acquisition::event_trampoline, this);

  started_ = std::chrono::steady_clock::now();
  running_.store(true, std::memory_order_release);
  thread_ = std::thread([this] { run(); });
}

void Acquisition::stop() {
  if (!running_.exchange(false, std::memory_order_acq_rel)) {
    if (client_ != nullptr) {
      udps_client_destroy(client_);
      client_ = nullptr;
    }
    return;
  }
  if (thread_.joinable()) thread_.join();
  if (client_ != nullptr) {
    udps_client_destroy(client_);
    client_ = nullptr;
  }
}

void Acquisition::run() {
  while (running_.load(std::memory_order_acquire)) {
    udps_client_poll(client_, 50);
    udps_stats_t s;
    udps_client_stats(client_, &s);
    StreamStats& out = model_.stats();
    out.packets.store(s.packets_received, std::memory_order_relaxed);
    out.frames.store(s.frames_delivered, std::memory_order_relaxed);
    out.gaps.store(s.counter_gaps, std::memory_order_relaxed);
    out.reconnects.store(s.reconnects, std::memory_order_relaxed);
    out.connected.store(udps_client_is_connected(client_) != 0, std::memory_order_relaxed);
  }
}

void Acquisition::on_config(const udps_signal_t* signals, std::uint32_t count,
                            std::uint8_t publish_mode) {
  std::vector<SignalInfo> info;
  info.reserve(count);
  for (std::uint32_t i = 0; i < count; ++i) info.push_back(to_info(signals[i]));

  // One line per CONFIG on stderr: what the stream says it is. This is the first
  // thing to look at when a signal refuses to plot.
  static const char* const kModes[] = {"strict", "accumulate", "decimate"};
  std::fprintf(stderr, "[udpscope] CONFIG: %u signals, publish=%s\n", count,
               publish_mode < 3 ? kModes[publish_mode] : "?");
  for (const SignalInfo& sig : info) {
    std::fprintf(stderr,
                 "[udpscope]   '%s' %ux%u (%u elem) rate=%g Hz time=%s idx=%u unit='%s'\n",
                 sig.name.c_str(), sig.rows, sig.cols, sig.elements(), sig.sampling_rate,
                 sig.time_mode == UDPS_TIME_PACKET       ? "packet"
                 : sig.time_mode == UDPS_TIME_FULL_ARRAY ? "full-array"
                 : sig.time_mode == UDPS_TIME_FIRST_SAMPLE ? "first-sample"
                                                           : "last-sample",
                 sig.time_signal_idx, sig.unit.c_str());
  }
  std::fflush(stderr);
  model_.stats().have_config.store(true, std::memory_order_relaxed);
  model_.set_signals(std::move(info));
}

void Acquisition::on_event(int event, const char* detail) {
  (void)detail;
  if (event == UDPS_EVENT_CONNECTED) {
    model_.stats().connected.store(true, std::memory_order_relaxed);
  } else if (event == UDPS_EVENT_DISCONNECTED) {
    model_.stats().connected.store(false, std::memory_order_relaxed);
  }
}

std::uint32_t Acquisition::build_stamps(const udps_frame_t* frame, int time_signal) {
  // The frame's span in base ticks: a scalar signal carries `num_samples` values
  // (one per batched cycle), a block signal (rows*cols > 1) carries its whole
  // element block. The largest of those is how many base ticks this frame spans --
  // in `strict` mode `num_samples` is 1 while a 1x1000 block carries 1000 samples.
  std::uint32_t n = frame->num_samples;
  for (std::uint32_t i = 0; i < frame->num_signals; ++i) {
    n = std::max(n, udps_signal_num_elements(&frame->signals[i]));
  }
  stamps_.assign(n, 0.0);

  double period = model_.partition().base_dt;
  if (!(period > 0.0)) period = 1.0e-3;
  // The cadence of the samples actually stored: the partition's slot interval (base
  // tick times the clock's decimation), as *declared* by the CONFIG.
  const double slot = model_.partition().slot_dt();
  double cadence = slot > 0.0 ? slot : period;
  // A declared rate is a claim the stream may not keep (a CONFIG declaring 1 MS/s
  // while delivering a hundred samples every fifth of a second is the case that
  // exposed this): stamps spaced by the declaration then span microseconds inside a
  // frame and jump by the frame interval between frames, which reads as a staircase
  // in the x column and, in an export, as data that is "downsampled" while the
  // values are fine. When the arrivals disagree with the declaration, the arrivals
  // are the truth -- the mean spacing of the samples that actually arrived.
  if (have_arrival_interval_ && n > 0) {
    const double measured = arrival_interval_ / double(n);
    const bool agrees = cadence > 0.0 && measured >= 0.75 * cadence && measured <= 1.25 * cadence;
    if (measured > 0.0 && std::isfinite(measured) && !agrees) {
      cadence = measured;
    }
  }

  bool used_stream_time = false;
  // Once the stream has been judged to have no usable time base, it keeps the
  // arrival clock for the rest of its life (until a CONFIG says otherwise): letting
  // the decision flip per frame is what interleaved two timelines in one ring.
  if (time_source_ != 2 && time_signal >= 0 &&
      time_signal < static_cast<int>(frame->num_signals)) {
    const udps_signal_t& ts = frame->signals[time_signal];
    // How the client indexes a signal's values depends on its shape
    // (`udps_frame_value`): a scalar carries one value per batched cycle, so the
    // stamp of sample k is (sample=k, elem=0); a multi-element signal carries one
    // *sample* of `elements` values per packet, so the stamp of sample k is
    // (sample=0, elem=k). Passing the scalar form for both is why FULL_ARRAY time
    // signals -- the only mode that carries a stamp per element -- read as
    // element 0 for every sample and were then judged degenerate.
    const bool scalar_time = udps_signal_num_elements(&ts) == 1;
    auto value_of = [&](std::uint32_t k) {
      return scalar_time
                 ? udps_frame_value(frame, static_cast<std::uint32_t>(time_signal), k, 0)
                 : udps_frame_value(frame, static_cast<std::uint32_t>(time_signal), 0, k);
    };
    switch (ts.time_mode) {
      case UDPS_TIME_FULL_ARRAY:
        for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = value_of(k);
        used_stream_time = true;
        break;
      case UDPS_TIME_FIRST_SAMPLE: {
        // Only element 0 is meaningful here: the producer stamps the batch's first
        // sample and says nothing about the rest, so the cadence has to space them --
        // exactly as LAST_SAMPLE below dates its batch backwards from its anchor.
        // Deriving a step from `value_of(n - 1)` instead cannot work: a scalar signal
        // in Accumulate mode still occupies `num_samples` slots on the wire, and a
        // producer fills them by repeating the anchor, so the step came out as zero
        // and every sample of the frame landed on one x. The degeneracy guard below
        // then threw the whole stream onto the arrival clock.
        const double anchor = value_of(0);
        for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = anchor + double(k) * cadence;
        used_stream_time = true;
        break;
      }
      case UDPS_TIME_LAST_SAMPLE: {
        const double anchor = value_of(n > 0 ? n - 1 : 0);
        for (std::uint32_t k = 0; k < n; ++k)
          stamps_[k] = anchor - double(n - 1 - k) * cadence;
        used_stream_time = true;
        break;
      }
      default:
        break;
    }
  }

  // A stream-time signal may be declared in a time unit rather than seconds ('ns' is
  // the common one). The scale has to be known *here*, before the checks below, because
  // they weigh the stamps' own spacing against the declared cadence, which is in
  // seconds: on a 'ns' stream an unscaled comparison is off by a factor of 1e9, so
  // every packet would read as an implausible jump.
  double unit_scale = 1.0;
  if (time_signal >= 0 && time_signal < static_cast<int>(model_.signals().size())) {
    const std::string& unit = model_.signals()[static_cast<std::size_t>(time_signal)].unit;
    if (unit == "ns") unit_scale = 1e-9;
    else if (unit == "us" || unit == "\u00b5s" || unit == "\xce\xbc" "s") unit_scale = 1e-6;
    else if (unit == "ms") unit_scale = 1e-3;
  }

  if (used_stream_time && n > 0) {
    // One sample's worth of the time signal's *own* unit, which is what a synthesised
    // continuation has to be spaced by and what a jump is measured against.
    const double step = cadence / unit_scale;
    // How far a packet's stamps may run ahead of the previous packet's last stamp
    // before the jump is read as a fault in the producer's clock rather than a gap in
    // the stream. Counted in packets, because that is what a loss burst costs: 256
    // packet spans in one go is an outage, not loss. Letting a real glitch through is
    // unrecoverable -- the monotonic clamp below then pins every later packet flat at
    // the glitched value, so the whole stream freezes at one x.
    constexpr double kGlitchPackets = 256.0;
    // How many packets in a row must fail before the declared time signal is abandoned
    // for the arrival clock. A glitch is one packet; a signal that is not a time base
    // at all fails every one of them.
    constexpr unsigned kFailureLimit = 4;

    // A time signal that does not actually vary -- a vector time signal, whose
    // values the client reads one element at a time, or a producer that leaves it
    // zero -- would put every sample at one x and collapse the axis to a point.
    // The arrival estimate is a worse clock but a usable one; a degenerate one is
    // not a time base at all.
    const char* fault = nullptr;
    double lo = stamps_[0];
    double hi = stamps_[0];
    if (!std::isfinite(stamps_[0])) fault = "carries a stamp that is not finite";
    for (std::uint32_t k = 1; fault == nullptr && k < n; ++k) {
      if (!std::isfinite(stamps_[k])) {
        fault = "carries a stamp that is not finite";
        break;
      }
      lo = std::min(lo, stamps_[k]);
      hi = std::max(hi, stamps_[k]);
    }
    if (fault == nullptr && !(hi > lo)) fault = "carries no span (every stamp is equal)";
    if (fault == nullptr && have_stream_anchor_) {
      // The stamps must also ADVANCE across packets, and by a plausible amount. A
      // producer that restarts its stamps every packet (or a signal mislabelled as
      // time) keeps handing back the same values; the max-clamp would then pin every
      // sample at one x, and the trigger's hold-off clock with it -- the engine waited
      // forever and no second event ever fired.
      if (!(stamps_[n - 1] > last_stream_anchor_)) {
        fault = "does not advance across packets";
      } else if (stamps_[0] - last_stream_anchor_ > kGlitchPackets * double(n) * step) {
        fault = "jumped implausibly far ahead of the previous packet";
      }
    }
    if (fault == nullptr) {
      stamp_failures_ = 0;
      last_stream_anchor_ = stamps_[n - 1];
      have_stream_anchor_ = true;
    } else if (have_stream_anchor_ && ++stamp_failures_ < kFailureLimit) {
      // A short run of bad packets is bridged rather than believed: the stamps are
      // replaced by a continuation of the accepted timeline at the declared cadence,
      // so neither the monotonic mark nor the anchor is poisoned, and the next good
      // packet -- which carries on from where the true timeline was -- is accepted
      // normally. Believing one glitching packet instead is what condemned the *next*,
      // perfectly good one and threw the whole stream onto the arrival clock.
      const double from = last_stream_anchor_;
      for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = from + double(k + 1) * step;
      last_stream_anchor_ = stamps_[n - 1];
      static bool bridged = false;
      if (!bridged) {
        bridged = true;
        std::fprintf(stderr,
                     "[udpscope] warning: the stream's time signal %s (%g..%g against a "
                     "previous %g); bridging that packet at the declared cadence and "
                     "keeping the stream clock.\n",
                     fault, lo, hi, from);
      }
    } else {
      used_stream_time = false;  // not a time base; fall back to arrival
      static bool warned = false;
      if (!warned) {
        warned = true;
        std::fprintf(stderr,
                     "[udpscope] warning: the stream's time signal %s (%g..%g); timing "
                     "samples by arrival instead. Check the time signal's mode and index "
                     "in the CONFIG dump above.\n",
                     fault, lo, hi);
      }
    }
  }

  if (!used_stream_time) {
    // Arrival-anchored: the packet is assumed to have arrived as its last element
    // was produced, and the earlier ones are dated backwards. A gap (`lost`)
    // shows up as extra elapsed time between arrivals, which is exactly the
    // spacing the samples had.
    const double end = frame->recv_time;
    if (!have_arrival_epoch_) {
      arrival_epoch_ = end;
      have_arrival_epoch_ = true;
    } else {
      const double elapsed = end - arrival_prev_;
      if (elapsed > 0.0 && std::isfinite(elapsed)) {
        // Smoothed: the socket clock jitters by a millisecond or so per frame, and
        // an unsmoothed step would make every frame's span disagree slightly.
        arrival_interval_ = have_arrival_interval_ ? 0.8 * arrival_interval_ + 0.2 * elapsed
                                                   : elapsed;
        have_arrival_interval_ = true;
      }
    }
    arrival_prev_ = end;
    const double base = end - arrival_epoch_;
    // The spacing between two *stored* samples is the partition's own cadence --
    // the exact interval the ingest filter decimates to, which is `slot_dt()` and
    // not the base tick. Deriving it from the arrival gap divided by
    // `n * (lost + 1)` collapsed the step to ~zero whenever the client reported
    // losses, and a whole frame then shared one f32 instant: the x column became a
    // staircase (many rows at one time, then a jump) while the values were fine --
    // which is what an export of it looks like, and what the plot's crosshair and
    // trigger times were reading. The arrival clock still anchors the frame's LAST
    // sample, so a lost packet or a stalled producer still shows up as a jump
    // *between* frames.
    // The frame's stamps start at the later of two anchors: where the arrival clock
    // puts the frame's first sample, and one cadence past the previous frame's last.
    // The second is what keeps the timeline strictly increasing -- anchoring the
    // frame's *end* at the arrival let a cadence estimate even slightly long reach
    // back past the previous end, and the monotonic clamp then flattened the overlap
    // into equal timestamps (a consumer plotting against them draws a vertical line
    // there, and an export reads as if sampled twice at one instant). A gap still
    // shows: when the arrivals run ahead, the first anchor wins and the frame jumps.
    const double base_rel = base - model_.epoch();
    const double start = std::max(base_rel - double(n - 1) * cadence, last_stamp_ + cadence);
    for (std::uint32_t k = 0; k < n; ++k) stamps_[k] = start + double(k) * cadence + model_.epoch();
  }

  // A change of time source invalidates the running mark: the stream-time branch
  // stamps in the time signal's own values and the arrival branch in seconds since
  // the first packet, and the two timelines share no zero. Leaving the mark at the
  // old source's last value made every arriving frame clamp flat onto it -- which is
  // the 2.5 s plateau of one timestamp that a run reports when its declared time
  // signal turns out unusable and the app falls back mid-stream.
  if (time_source_ == 0) {
    time_source_ = used_stream_time ? 1 : 2;
  } else if (time_source_ == 1 && !used_stream_time) {
    // The declared time signal failed its checks for good: keep the arrival clock from
    // here on, anchored so that it continues from the monotonic mark. The two timelines
    // then share a zero *by construction*, which is the whole point of deriving the
    // epoch from `last_stamp_` -- so the recorded history is still on one timeline and
    // is kept. It used to be wiped here, on the reasoning that the old stamps belonged
    // to a dead clock; they do not, and the wipe is what made a single glitching packet
    // look like the stream restarting from zero.
    time_source_ = 2;
    const double end = frame->recv_time;
    arrival_epoch_ = end - (last_stamp_ + model_.epoch());
    arrival_prev_ = end;
  }

  // The declared unit, applied last so the x axis is in seconds either way. It applies
  // ONLY to stream time: the arrival fallback already produces seconds off the socket
  // clock -- scaling those by the time signal's unit shrank the whole time base by 1e9
  // on 'ns' streams, which froze the trigger's hold-off clock and pinned the axis at ~0.
  if (used_stream_time && unit_scale != 1.0) {
    for (std::uint32_t k = 0; k < n; ++k) stamps_[k] *= unit_scale;
  }

  return n;
}

void Acquisition::on_data(const udps_frame_t* frame) {

  std::lock_guard<std::mutex> lock(model_.mutex());
  if (!running_.load(std::memory_order_relaxed)) return;

  const Partition& part = model_.partition();
  const std::vector<ChannelInfo>& channels = model_.channels();
  if (part.time_capacity == 0 || channels.empty()) return;

  // The time reference, in order of preference: the signal the first channel's
  // CONFIG entry names via `time_signal_idx`, or -- when the stream instead
  // carries its time base as an explicit full-array signal -- that signal.
  int time_signal = -1;
  {
    const SignalInfo& sig = model_.signals()[channels[0].signal];
    if (sig.time_signal_idx != 0xFFFFFFFFu && sig.time_signal_idx < model_.signals().size()) {
      time_signal = static_cast<int>(sig.time_signal_idx);
    }
  }
  if (time_signal < 0) {
    // No channel names a time signal: look for the standalone timestamp signal
    // (full-array, more than one element, no declared rate).
    for (std::size_t i = 0; i < model_.signals().size(); ++i) {
      const SignalInfo& sig = model_.signals()[i];
      if (sig.time_mode == UDPS_TIME_FULL_ARRAY && sig.elements() > 1 &&
          sig.sampling_rate == 0.0) {
        time_signal = static_cast<int>(i);
        break;
      }
    }
  }
  const std::uint32_t n = build_stamps(frame, time_signal);
  if (n == 0) return;

  if (model_.epoch() == 0.0) {
    model_.set_epoch_if_unset(stamps_[0]);
    std::fprintf(stderr, "[udpscope] time base: epoch=%.3f, first frame recv=%.3f, n=%u\n",
                 model_.epoch(), frame->recv_time, n);
    std::fflush(stderr);
  }
  const double epoch = model_.epoch();

  TriggerConfig& tc = model_.trigger();
  // One ring, always filling, whatever the trigger is doing: the pre-trigger
  // history has to already be there when an edge arrives, and the first latch after
  // arming would otherwise open on however little had accumulated since. What
  // freezes in trigger mode is the *view*, not acquisition -- the latch copies the
  // event window out into an immutable `EventSnapshot`.
  BufferStore& store = model_.live();
  const u64 g = part.time_stride;

  // The filter's groups are `g` long, so a re-plan has to start a fresh group rather than
  // finish the one it was in with the wrong length.
  if (acc_stride_ != g || acc_channels_ != channels.size()) {
    acc_stride_ = g;
    acc_channels_ = channels.size();
    value_acc_.assign(channels.size(), 0.0);
    value_acc_n_.assign(channels.size(), 0);
    math_sample_.assign(channels.size(), std::nanf(""));
    time_acc_ = 0.0;
    time_acc_n_ = 0;
  }

  // Calibration is read per frame; it only changes under the same lock.
  TriggerEngine& engine = engine_;
  const int trig_channel = tc.channel;
  // The post-trigger window straight through in seconds. There used to be a measured,
  // smoothed samples-per-second estimate here, converting the window into a sample
  // count for the engine; the estimate is what sized captures wrong. Its floor
  // (`inst >= 1.0`) turned any stream whose time signal advances in large steps into
  // 1 S/s, the count came out as 1, and the engine latched on the edge sample itself
  // -- an event with no post-trigger data, which is exactly the "post window is
  // empty" defect. The engine is fed stream time per sample anyway, so it can count
  // the window itself and no conversion has to exist.
  //
  // The oldest buffered stamp goes with it, so the engine can refuse an edge the
  // history does not reach back far enough behind. Read from the ring rather than
  // timed from the arm: the ring fills with the trigger off too, so the pre window is
  // usually already there and a timer would delay the first event for nothing.
  const double event_span = model_.event_span();
  const double post_want = tc.post_span_s(event_span);
  const double oldest_s =
      store.time().size() > 0 ? double(store.time()[0]) : last_stamp_;
  // The channel's *name* goes with it, because that is what identifies it across a
  // rebuild of the channel list (see `TriggerEngine::configure`).
  static const std::string kNoTriggerChannel;
  const std::string& trig_name =
      trig_channel >= 0 && trig_channel < static_cast<int>(channels.size())
          ? channels[static_cast<std::size_t>(trig_channel)].name
          : kNoTriggerChannel;
  engine.configure(tc, trig_name, post_want, event_span - post_want, oldest_s);
  // The engine's armed state follows the panel's, both ways: without the disarm
  // half, Disarm left the engine firing (its own `armed_` was only ever set), and
  // re-entering trigger mode kept the stale prev/edge/hold-off state.
  if (tc.enabled && tc.armed) {
    if (!engine.armed()) {
      engine.arm(last_stamp_);
      model_.trigger_armed().store(true, std::memory_order_relaxed);
    }
  } else if (engine.armed()) {
    // Disarm cuts a capture in progress where it got to and keeps it, instead of
    // discarding it. `engine.disarm()` clears `pending_`, which drops the display out of
    // `Capturing`; with nothing published the view fell back to the *previous* event, or
    // to blank if this was the first -- the capture the user was watching simply
    // vanished. Publishing first means the same snapshot path freezes it, so what is on
    // screen at the moment of the Disarm stays on screen.
    //
    // The full post window is requested and `live_window_locked()` clips it to what the
    // rings hold, so the snapshot ends at the newest sample and reports its own shorter
    // `post_s` -- the same way a first latch whose pre window was short reports its own
    // `pre_s`. No event fired, so `trigger_latches()` is deliberately *not* bumped; the
    // UI notices the new picture through `event_seq()`.
    if (engine.pending()) {
      const double edge = engine.last_trigger_time();
      model_.publish_event(event_span - post_want, post_want, edge);
      model_.trigger_time().store(edge, std::memory_order_relaxed);
      model_.trigger_time_valid().store(true, std::memory_order_relaxed);
      auto snap = model_.event();
      std::fprintf(stderr,
                   "[udpscope] trigger: disarmed mid-capture, edge at %.3f s kept as "
                   "[-%.4f..%.4f] n=%llu (post window asked %.4f s, filled %.4f s)\n",
                   edge, snap ? snap->pre_s : 0.0, snap ? snap->post_s : 0.0,
                   static_cast<unsigned long long>(snap ? snap->stamps.size() : 0), post_want,
                   engine.post_elapsed());
      std::fflush(stderr);
    }
    engine.disarm();
    model_.trigger_armed().store(false, std::memory_order_relaxed);
  }

  double last_stamp_used = last_stamp_;

  // The latch is handled on the sample it happens on, and with the edge the engine
  // latched *on* -- after that sample's channels have all been pushed (see the loop).
  //
  // It used to set a flag and publish once after the loop, reading the edge back from
  // the engine. But the loop keeps feeding the engine after a latch, so any further
  // edge in the rest of the packet -- the hold-off is 10 ms and a packet can be a long
  // block of samples -- overwrote that edge before it was read. The capture was then
  // cut around an edge whose post window had not been collected yet, so the event came
  // out all pre-trigger with the marker and x = 0 at the right-hand end of the data.
  // Publishing here also means the ring provably holds the post window: the latch
  // condition is that stream time has passed `edge + post`, and every sample up to
  // this one has already been pushed.
  const double span = std::max(1.0e-9, model_.event_span());
  const double post_s = tc.post_span_s(span);
  const auto latch = [&](double edge) {
    model_.publish_event(span - post_s, post_s, edge);
    model_.trigger_latches().fetch_add(1, std::memory_order_relaxed);
    model_.trigger_time().store(edge, std::memory_order_relaxed);
    model_.trigger_time_valid().store(true, std::memory_order_relaxed);
    // The captured window is reported from the snapshot itself, not from the request:
    // `want` versus `got` is what distinguishes "the engine did not wait" from "the
    // cut lost the data", which is otherwise guesswork from the picture alone.
    auto snap = model_.event();
    std::fprintf(stderr,
                 "[udpscope] trigger: edge at %.3f s, latched at %.3f s of stream time "
                 "(want [-%.4f..%.4f] got [-%.4f..%.4f] n=%llu, frame recv=%.3f, "
                 "ticks_so_far=%llu, wall=%.3f s)\n",
                 edge, last_stamp_, span - post_s, post_s, snap ? snap->pre_s : 0.0,
                 snap ? snap->post_s : 0.0,
                 static_cast<unsigned long long>(snap ? snap->stamps.size() : 0),
                 frame->recv_time, static_cast<unsigned long long>(tick_),
                 std::chrono::duration<double>(std::chrono::steady_clock::now() - started_)
                     .count());
    std::fflush(stderr);
    if (tc.mode == TriggerMode::Single) tc.armed = false;
  };

  for (std::uint32_t k = 0; k < n; ++k) {
    const u64 tick = tick_++;
    const double rel = std::max(stamps_[k] - epoch, last_stamp_);
    last_stamp_used = rel;
    last_stamp_ = rel;

    // The stored stamp is the *mean* of the group, not its last member: that is where the
    // boxcar's group delay puts the averaged value, and stamping it at the end of the
    // group would shift every trace half a stored sample late.
    time_acc_ += rel;
    ++time_acc_n_;
    if (tick % g == g - 1) {
      store.time().push(time_acc_ / static_cast<double>(time_acc_n_));
      time_acc_ = 0.0;
      time_acc_n_ = 0;
    }

    // The latch belongs to this sample, but it cannot be published from inside the
    // channel loop: `publish_event()` clamps the slice to every value ring's
    // `produced()`, and the channels *after* the trigger channel have not pushed this
    // sample yet, so the newest sample -- the one the post window was waiting for --
    // would be clamped away. One sample is invisible in a 20 000-sample capture (it is
    // the `n=19999` and the `0.9999` instead of `1.0000` in the probe's output), and at
    // one sample per second it is the entire post side: `got [-0.9997..0.0000]`, which
    // reads exactly like the "post window is empty" defect.
    bool latch_here = false;
    double latch_edge = 0.0;

    for (std::size_t i = 0; i < channels.size(); ++i) {
      const ChannelInfo& ch = channels[i];
      // A derived channel is filled by `push_math_sample()` below, once per stored slot, from the
      // values the streamed ones just pushed. Reading it from the packet would give zero (it has no
      // signal) and pushing that would advance its value ring twice per time slot -- which is an
      // alternation between the value and 0 while the buffer is filling, and a complete loss of the
      // pairing once both rings wrap.
      if (ch.math) continue;
      // A signal's columns are consecutive samples of the same channel: a scalar
      // (one element) carries one value per batched cycle, a block signal carries
      // the cycle's sample at element `k`. Either way element 0 is the channel.
      const double raw =
          ch.elements > 1
              ? udps_frame_value(frame, static_cast<std::uint32_t>(ch.signal), 0,
                                 std::min<std::uint32_t>(k, ch.elements - 1))
              : udps_frame_value(frame, static_cast<std::uint32_t>(ch.signal), k, 0);
      // Every sample is read and accumulated even when only one in `period` is stored:
      // that is the whole point of the filter. Skipping the read -- which is what the
      // old stride did -- is what made decimation alias.
      value_acc_[i] += raw * ch.gain + ch.offset;
      ++value_acc_n_[i];

      const u64 period = g * part.stride[i];
      if (tick % period != period - 1) continue;
      const f32 value =
          static_cast<f32>(value_acc_[i] / static_cast<double>(value_acc_n_[i]));
      value_acc_[i] = 0.0;
      value_acc_n_[i] = 0;
      store.values(i).push(value);
      // This slot's value, for the math channels below. Every channel is pushed on the same slot
      // (per-channel strides are 1 by design), so the vector is complete by the time the loop ends.
      if (i < math_sample_.size()) math_sample_[i] = value;
      // The trigger sees the *stored* signal, not the wire signal. An edge found in a
      // sample that is not in the buffer would be cut from data that does not contain it,
      // and with the filter in front of it a spike above the stored Nyquist is exactly
      // the thing that must not latch -- it is not in the record the user would get.
      if (static_cast<int>(i) == trig_channel) {
        if (engine.process(value, rel)) {
          latch_here = true;
          latch_edge = engine.last_trigger_time();
        }
      }
    }
    // The derived channels: once per *stored* slot, and **before** the latch publishes.
    // `publish_event()` clamps its snapshot to every value ring's `produced()`, so a derived
    // channel pushed after it would be missing its newest sample from the capture -- the same
    // one-sample clamp the deferred latch above exists for.
    if (tick % g == g - 1) model_.push_math_sample(math_sample_);
    if (latch_here) latch(latch_edge);
  }
  last_stamp_ = last_stamp_used;

  model_.trigger_armed().store(engine.armed(), std::memory_order_relaxed);
  // The edge goes out before the flag that makes the UI read it (release/acquire on
  // `trigger_pending`), so a tick that sees a capture open sees which edge it is for. A
  // capture that opens and latches inside this one frame never reports pending, and that
  // is right: the display goes straight to the frozen event.
  if (engine.pending()) model_.set_capture_edge(engine.last_trigger_time());
  model_.set_trigger_pending(engine.pending());
  static std::int64_t next_engine_log = 0;
  const std::int64_t now_e = static_cast<std::int64_t>(frame->recv_time * 1.0e6);
  if (now_e >= next_engine_log) {
    next_engine_log = now_e + 1000000;
    // `elapsed/post_want` is the capture's progress in the only unit that matters now
    // (stream seconds); `accum` is how many samples that window has collected, which
    // is what says whether a capture is thin because the window is short or because
    // the stream is slow.
    std::fprintf(stderr,
                 "[udpscope] trig engine: ch=%d armed=%d pending=%d elapsed=%.4f/%.4f s "
                 "accum=%llu history=%.4f/%.4f s pre_pct=%.0f v=%.4g now_t=%.3f "
                 "holdoff_until=%.3f\n",
                 trig_channel, int(engine.armed()), int(engine.pending()),
                 engine.post_elapsed(), engine.post_seconds(),
                 static_cast<unsigned long long>(engine.post_accum()), engine.history(),
                 engine.pre_seconds(), tc.pretrigger_pct, engine.last_value(),
                 engine.last_t(), engine.holdoff_until());
    std::fflush(stderr);
  }
}

// Static trampolines: the client's C ABI cannot call a member function directly.
void Acquisition::config_trampoline(const udps_signal_t* signals, std::uint32_t count,
                                    std::uint8_t publish_mode, void* user) {
  static_cast<Acquisition*>(user)->on_config(signals, count, publish_mode);
}

void Acquisition::data_trampoline(const udps_frame_t* frame, void* user) {
  static_cast<Acquisition*>(user)->on_data(frame);
}

void Acquisition::event_trampoline(udps_event_t event, const char* detail, void* user) {
  static_cast<Acquisition*>(user)->on_event(static_cast<int>(event), detail);
}

}  // namespace udpscope