///|
/// Parameters for finite-window content-defined chunking.
pub(all) struct CdcConfig {
  min_size : Int
  average_size : Int
  max_size : Int
  window_size : Int
} derive(Debug, Eq)

///|
pub(all) struct CdcSummary {
  bytes : Int
  chunks : Int
  min_chunk : Int
  max_chunk : Int
  average_chunk : Double
} derive(Debug)

///|
/// Stateful finite-window CDC processor for streaming or batched input.
///
/// The rolling window remains continuous across emitted chunks, matching the
/// boundary semantics of `content_defined_chunks` while avoiding the need to
/// retain the complete input in memory.
pub(all) struct CdcStream {
  config : CdcConfig
  pending : Array[Int]
  ring : Array[Int]
  mut ring_start : Int
  mut rolling_value : Int
  mut next_start : Int
  highest_power : Int
} derive(Debug)

///|
pub fn CdcConfig::new(
  min_size : Int,
  average_size : Int,
  max_size : Int,
  window_size : Int,
) -> CdcConfig {
  let safe_min = max_int(1, min_size)
  let safe_average = max_int(safe_min, average_size)
  let safe_max = max_int(safe_average, max_size)
  {
    min_size: safe_min,
    average_size: safe_average,
    max_size: safe_max,
    window_size: max_int(1, min_int(window_size, safe_min)),
  }
}

///|
pub fn CdcStream::new(config : CdcConfig) -> CdcStream {
  let highest_power = modular_power(257, config.window_size - 1, 1000000007)
  {
    config,
    pending: [],
    ring: [],
    ring_start: 0,
    rolling_value: 0,
    next_start: 0,
    highest_power,
  }
}

///|
fn CdcStream::push_byte(self : CdcStream, byte : Int) -> Unit {
  let normalized = normalize_byte(byte)
  self.pending.push(normalized)
  if self.ring.length() < self.config.window_size {
    self.ring.push(normalized)
    if self.ring.length() == self.config.window_size {
      self.rolling_value = polynomial_hash_value(self.ring)
    }
  } else {
    let removed = self.ring[self.ring_start] + 1
    self.ring[self.ring_start] = normalized
    self.ring_start = (self.ring_start + 1) % self.config.window_size
    let modulus = 1000000007L
    let without_old = (
        self.rolling_value.to_int64() -
        removed.to_int64() * self.highest_power.to_int64() % modulus +
        modulus
      ) %
      modulus
    self.rolling_value = ((without_old * 257L + (normalized + 1).to_int64()) %
    modulus).to_int()
  }
}

///|
fn CdcStream::should_cut(self : CdcStream) -> Bool {
  let length = self.pending.length()
  let has_window = self.ring.length() == self.config.window_size
  let content_boundary = length >= self.config.min_size &&
    has_window &&
    self.rolling_value % self.config.average_size == 0
  content_boundary || length >= self.config.max_size
}

///|
fn CdcStream::emit_pending(self : CdcStream) -> ChunkFingerprint {
  let chunk = {
    start: self.next_start,
    length: self.pending.length(),
    hash: polynomial_hash_value(self.pending),
  }
  self.next_start = self.next_start + self.pending.length()
  self.pending.clear()
  chunk
}

///|
/// Feeds any number of bytes and returns only fully decided chunks.
pub fn CdcStream::push(
  self : CdcStream,
  bytes : Array[Int],
) -> Array[ChunkFingerprint] {
  let completed : Array[ChunkFingerprint] = []
  for byte in bytes {
    self.push_byte(byte)
    if self.should_cut() {
      completed.push(self.emit_pending())
    }
  }
  completed
}

///|
/// Finishes a stream and returns its final partial chunk, if present.
pub fn CdcStream::finish(self : CdcStream) -> Array[ChunkFingerprint] {
  let completed : Array[ChunkFingerprint] = []
  if self.pending.length() > 0 {
    completed.push(self.emit_pending())
  }
  completed
}

///|
pub fn CdcStream::pending_bytes(self : CdcStream) -> Int {
  self.pending.length()
}

///|
/// Splits data at content-derived boundaries.
///
/// Boundary decisions use a finite rolling window, allowing chunk boundaries
/// to resynchronize after local insertions or deletions.
pub fn content_defined_chunks(
  bytes : Array[Int],
  config : CdcConfig,
) -> Array[ChunkFingerprint] {
  let result : Array[ChunkFingerprint] = []
  if bytes.length() == 0 {
    return result
  }
  let rolling = window_hashes(bytes, config.window_size)
  let mut start = 0
  for end = 1; end <= bytes.length(); end = end + 1 {
    let length = end - start
    let enough_context = end >= config.window_size
    let fingerprint = if enough_context {
      rolling[end - config.window_size].value
    } else {
      0
    }
    let content_boundary = length >= config.min_size &&
      enough_context &&
      fingerprint % config.average_size == 0
    let forced_boundary = length >= config.max_size
    if content_boundary || forced_boundary || end == bytes.length() {
      result.push(fingerprint_range(bytes, start, end))
      start = end
    }
  }
  result
}

///|
fn fingerprint_range(
  bytes : Array[Int],
  start : Int,
  end : Int,
) -> ChunkFingerprint {
  let chunk : Array[Int] = []
  for i = start; i < end; i = i + 1 {
    chunk.push(bytes[i])
  }
  { start, length: end - start, hash: polynomial_hash_value(chunk) }
}

///|
pub fn summarize_cdc(chunks : Array[ChunkFingerprint]) -> CdcSummary {
  if chunks.length() == 0 {
    return {
      bytes: 0,
      chunks: 0,
      min_chunk: 0,
      max_chunk: 0,
      average_chunk: 0.0,
    }
  }
  let mut bytes = 0
  let mut min_chunk = chunks[0].length
  let mut max_chunk = chunks[0].length
  for chunk in chunks {
    bytes = bytes + chunk.length
    min_chunk = min_int(min_chunk, chunk.length)
    max_chunk = max_int(max_chunk, chunk.length)
  }
  {
    bytes,
    chunks: chunks.length(),
    min_chunk,
    max_chunk,
    average_chunk: bytes.to_double() / chunks.length().to_double(),
  }
}

///|
pub fn CdcSummary::to_json(self : CdcSummary) -> String {
  "{\"bytes\":\{self.bytes},\"chunks\":\{self.chunks},\"min_chunk\":\{self.min_chunk},\"max_chunk\":\{self.max_chunk},\"average_chunk\":\{self.average_chunk}}"
}

///|
fn min_int(a : Int, b : Int) -> Int {
  if a < b {
    a
  } else {
    b
  }
}

///|
fn max_int(a : Int, b : Int) -> Int {
  if a > b {
    a
  } else {
    b
  }
}