/// Content-defined chunking.

pub(all) struct ChunkerStats {
  input_bytes : Int
  chunk_count : Int
  smallest : Int
  largest : Int
  average : Int
  boundaries : Array[Int]
} derive(Show)

pub fn ChunkerStats::empty() -> ChunkerStats {
  { input_bytes: 0, chunk_count: 0, smallest: 0, largest: 0, average: 0, boundaries: [] }
}

pub fn ChunkerStats::input(self : ChunkerStats) -> Int { self.input_bytes }
pub fn ChunkerStats::count(self : ChunkerStats) -> Int { self.chunk_count }
pub fn ChunkerStats::min_size(self : ChunkerStats) -> Int { self.smallest }
pub fn ChunkerStats::max_size(self : ChunkerStats) -> Int { self.largest }
pub fn ChunkerStats::mean(self : ChunkerStats) -> Int { self.average }
pub fn ChunkerStats::boundary_offsets(self : ChunkerStats) -> Array[Int] { self.boundaries.copy() }

fn boundary_mask(average : Int) -> UInt {
  let mut power = 1
  let mut bits = 0
  while power < average {
    power = power * 2
    bits = bits + 1
  }
  if bits <= 1 { 1U } else { (1U << (bits - 1)) - 1U }
}

fn gear_value(index : Int, seed : UInt) -> UInt {
  let mut x = seed ^ index.reinterpret_as_uint()
  x = x ^ (x >> 16)
  x = x * 0x7feb352dU
  x = x ^ (x >> 15)
  x = x * 0x846ca68bU
  x ^ (x >> 16)
}

fn update_roll(hash : UInt, byte : Byte, position : Int, seed : UInt) -> UInt {
  let table_value = gear_value(byte.to_int(), seed)
  let mixed = (hash << 1) | (hash >> 31) ^ table_value
  mixed + position.reinterpret_as_uint() * 0x9e3779b9U
}

fn should_cut(hash : UInt, size : Int, cfg : ChunkerConfig, mask : UInt) -> Bool {
  size >= cfg.minimum() && ((hash & mask) == 0U || size >= cfg.maximum())
}

fn take_slice(source : Array[Byte], start : Int, end : Int) -> Array[Byte] {
  let out = []
  for i in start.. Array[Chunk] {
  let _ = cfg.validate()
  let chunks = []
  let mut start = 0
  let mut hash = cfg.seed
  let mask = boundary_mask(cfg.average())
  let mut ordinal = 0
  for position, byte in source {
    hash = update_roll(hash, byte, position, cfg.seed)
    let size = position + 1 - start
    if should_cut(hash, size, cfg, mask) {
      let data = take_slice(source, start, position + 1)
      chunks.push(Chunk::new(ordinal, Span::new(start, position + 1), data))
      ordinal = ordinal + 1
      start = position + 1
      hash = cfg.seed ^ ordinal.reinterpret_as_uint()
    }
  }
  if start < source.length() {
    chunks.push(Chunk::new(ordinal, Span::new(start, source.length()), take_slice(source, start, source.length())))
  }
  chunks
}

/// Incremental chunker for readers that arrive in arbitrary input fragments.
pub(all) struct StreamingChunker {
  config : ChunkerConfig
  mut buffer : Array[Byte]
  mut total_seen : Int
  mut next_ordinal : Int
  mut rolling : UInt
  mut boundaries : Array[Int]
}

pub fn StreamingChunker::new(config : ChunkerConfig) -> StreamingChunker {
  {
    config,
    buffer: [],
    total_seen: 0,
    next_ordinal: 0,
    rolling: config.seed,
    boundaries: [],
  }
}

pub fn StreamingChunker::bytes_seen(self : StreamingChunker) -> Int { self.total_seen }
pub fn StreamingChunker::pending_size(self : StreamingChunker) -> Int { self.buffer.length() }
pub fn StreamingChunker::config(self : StreamingChunker) -> ChunkerConfig { self.config }

fn StreamingChunker::emit_if_ready(self : StreamingChunker, final_chunk : Bool) -> Array[Chunk] {
  let output = []
  let mask = boundary_mask(self.config.average())
  let mut cut = -1
  if final_chunk && self.buffer.length() > 0 {
    cut = self.buffer.length()
  } else if self.buffer.length() >= self.config.minimum() {
    for i in 0.. 0 {
    let data = take_slice(self.buffer, 0, cut)
    let start = self.total_seen - self.buffer.length()
    output.push(Chunk::new(self.next_ordinal, Span::new(start, start + cut), data))
    self.next_ordinal = self.next_ordinal + 1
    self.boundaries.push(start + cut)
    self.buffer = take_slice(self.buffer, cut, self.buffer.length())
  }
  output
}

/// Feed one fragment. It may return zero or more complete chunks.
pub fn StreamingChunker::push(self : StreamingChunker, fragment : Array[Byte]) -> Array[Chunk] {
  for b in fragment { self.buffer.push(b) }
  self.total_seen = self.total_seen + fragment.length()
  let output = []
  while self.buffer.length() >= self.config.maximum() {
    let emitted = self.emit_if_ready(false)
    if emitted.length() == 0 {
      let data = take_slice(self.buffer, 0, self.config.maximum())
      let start = self.total_seen - self.buffer.length()
      output.push(Chunk::new(self.next_ordinal, Span::new(start, start + data.length()), data))
      self.next_ordinal = self.next_ordinal + 1
      self.buffer = take_slice(self.buffer, data.length(), self.buffer.length())
    } else {
      for c in emitted { output.push(c) }
    }
  }
  output
}

/// Finish the stream and return the final partial chunk.
pub fn StreamingChunker::finish(self : StreamingChunker) -> Array[Chunk] {
  let output = []
  while self.buffer.length() > 0 {
    let emitted = self.emit_if_ready(true)
    if emitted.length() == 0 { break }
    for c in emitted { output.push(c) }
  }
  output
}

pub fn StreamingChunker::finish_all(self : StreamingChunker) -> Array[Chunk] {
  let out = self.finish()
  out
}

pub fn chunk_stats(chunks : Array[Chunk], input_bytes : Int) -> ChunkerStats {
  if chunks.length() == 0 {
    return { ..ChunkerStats::empty(), input_bytes }
  }
  let mut smallest = chunks[0].size()
  let mut largest = chunks[0].size()
  let mut total = 0
  let boundaries = []
  for chunk in chunks {
    let size = chunk.size()
    if size < smallest { smallest = size }
    if size > largest { largest = size }
    total = total + size
    boundaries.push(chunk.range().end)
  }
  { input_bytes, chunk_count: chunks.length(), smallest, largest, average: total / chunks.length(), boundaries }
}

pub fn chunk_histogram(chunks : Array[Chunk], buckets : Int) -> Array[Int] {
  let count = if buckets <= 0 { 1 } else { buckets }
  let out = Array::make(count, 0)
  for chunk in chunks {
    let slot = chunk.size() % count
    out[slot] = out[slot] + 1
  }
  out
}

pub fn chunk_boundaries(chunks : Array[Chunk]) -> Array[Int] {
  chunks.map(c => c.range().end)
}

pub fn chunk_total_size(chunks : Array[Chunk]) -> Int {
  let mut total = 0
  for c in chunks { total = total + c.size() }
  total
}