#include <plot/data/stream_series.hpp>

#include <algorithm>
#include <cstddef>

namespace plot::data {
namespace {

/// Samples copied per ring read on the ingest path, so appending the samples
/// produced between two frames costs no allocation.
constexpr u64 kReadChunk = 256;

}  // namespace

StreamSeries::StreamSeries(u64 capacity, f64 x0, f64 dx)
    : ring_(capacity), x0_(x0), dx_(dx) {
  // Size the pyramid's level table for a full window up front: the ring never
  // holds more than capacity() samples, so appending never grows it.
  pyramid_.reserve(capacity);
}

void StreamSeries::push(f32 y) { ring_.push(y); }

void StreamSeries::push(const f32* ys, u64 n) { ring_.push(ys, n); }

u64 StreamSeries::produced() const { return ring_.produced(); }

void StreamSeries::acquire() {
  // One atomic read: every absolute index below is relative to this single count,
  // so the producer may keep pushing without the snapshot shifting under us.
  const u64 p = ring_.produced();
  const u64 cap = ring_.capacity();
  const u64 base = p > cap ? p - cap : 0;   // oldest index the ring still holds

  // Slide the pyramid's window: the buckets that covered samples the ring has
  // overwritten are dropped, and the rest stay exactly as they were built --
  // their boundaries are absolute, so nothing has to be recomputed. This is also
  // where a producer that lapped the consumer by more than the ring holds is
  // handled: the pyramid restarts at the new base.
  pyramid_.drop_before(base);

  // Append exactly the samples produced since the last acquire(). Normally that
  // is one frame's worth; after a lapped producer it is the whole window.
  f32 chunk[kReadChunk];
  for (u64 a = pyramid_.window_end(); a < p;) {
    const u64 take = std::min<u64>(kReadChunk, p - a);
    ring_.copy_from(a, take, chunk);
    for (u64 k = 0; k < take; ++k) pyramid_.append(chunk[k]);
    a += take;
  }

  // Exact bounds for the published window, in the same O(log n + kBucket) as any
  // other exact query over it: the bucket-aligned interior comes from the
  // pyramid and the two partial edge spans are scanned through copy_raw(). An
  // incrementally extended bound cannot work here -- it would have to know
  // whether the extreme it is holding has just slid out of the window.
  y_bounds_ = reduce(0, size());
}

u64 StreamSeries::size() const { return pyramid_.size(); }

f64 StreamSeries::x0() const { return x0_ + f64(pyramid_.window_base()) * dx_; }

f64 StreamSeries::dx() const { return dx_; }

Range StreamSeries::x_bounds() const {
  const f64 n = f64(size());
  return n == 0.0 ? Range{x0(), x0()} : Range{x0(), x0() + (n - 1.0) * dx_};
}

Range StreamSeries::y_bounds() const { return y_bounds_; }

const MinMaxPyramid& StreamSeries::pyramid() const { return pyramid_; }

void StreamSeries::gather(u32 level, u64 begin, u64 end, std::vector<f32>& out_xy) const {
  gather_impl(level, begin, end, out_xy);
}

void StreamSeries::copy_raw(u64 begin, u64 end, f32* out) const {
  const u64 want = end > begin ? end - begin : 0;
  if (want == 0) return;
  const u64 n = size();
  const u64 b = std::min(begin, n);
  const u64 e = std::min(std::max(end, b), n);
  // Retained index b is absolute window_base() + b: fixed for the whole copy, so
  // a pushing producer cannot shift the window midway (the pyramid is written by
  // this thread alone, so its window cannot move under the read either).
  ring_.copy_from(pyramid_.window_base() + b, e - b, out);
  std::fill(out + static_cast<std::ptrdiff_t>(e - b),
            out + static_cast<std::ptrdiff_t>(want), 0.0f);
}

}  // namespace plot::data