///|
/// Redacted callback-level measurements for a host storage sink. The values
/// make downstream buffering visible without recording field names, filenames,
/// payload bytes, or sink-specific paths.
pub(all) struct SinkMetrics {
  begin_calls : Int
  data_calls : Int64
  end_calls : Int
  commit_calls : Int
  abort_calls : Int
  payload_bytes : Int64
  max_chunk_bytes : Int
} derive(Debug, Eq)

///|
/// A sink decorator that records callback volume and peak application chunk
/// size while forwarding behavior unchanged to an application-owned sink.
pub struct MeasuredSink {
  delegate : UploadSink
  mut begin_calls : Int
  mut data_calls : Int64
  mut end_calls : Int
  mut commit_calls : Int
  mut abort_calls : Int
  mut payload_bytes : Int64
  mut max_chunk_bytes : Int
}

///|
pub fn MeasuredSink::new(delegate : UploadSink) -> MeasuredSink {
  {
    delegate,
    begin_calls: 0,
    data_calls: 0L,
    end_calls: 0,
    commit_calls: 0,
    abort_calls: 0,
    payload_bytes: 0L,
    max_chunk_bytes: 0,
  }
}

///|
pub fn MeasuredSink::sink(self : MeasuredSink) -> UploadSink {
  UploadSink::new(
    part => {
      self.begin_calls += 1
      (self.delegate.on_begin)(part)
    },
    (part, bytes) => {
      self.data_calls += 1L
      self.payload_bytes += bytes.length().to_int64()
      self.max_chunk_bytes = Int::max(self.max_chunk_bytes, bytes.length())
      (self.delegate.on_data)(part, bytes)
    },
    part => {
      self.end_calls += 1
      (self.delegate.on_end)(part)
    },
    receipt => {
      self.commit_calls += 1
      (self.delegate.on_commit)(receipt)
    },
    error => {
      self.abort_calls += 1
      (self.delegate.on_abort)(error)
    },
  )
}

///|
pub fn MeasuredSink::metrics(self : MeasuredSink) -> SinkMetrics {
  {
    begin_calls: self.begin_calls,
    data_calls: self.data_calls,
    end_calls: self.end_calls,
    commit_calls: self.commit_calls,
    abort_calls: self.abort_calls,
    payload_bytes: self.payload_bytes,
    max_chunk_bytes: self.max_chunk_bytes,
  }
}