///|
/// Result of memory-budgeted indexing. `flushed_segments` counts immutable
/// segments emitted before the final flush; `peak_estimated_bytes` is a
/// conservative input-side estimate rather than allocator RSS.
pub struct BudgetedWriterStats {
  memory_budget_bytes : Int
  flushed_segments : Int
  indexed_documents : Int
  peak_estimated_bytes : Int
}

///|
pub fn BudgetedWriterStats::memory_budget_bytes(
  self : BudgetedWriterStats,
) -> Int {
  self.memory_budget_bytes
}

///|
pub fn BudgetedWriterStats::flushed_segments(self : BudgetedWriterStats) -> Int {
  self.flushed_segments
}

///|
pub fn BudgetedWriterStats::indexed_documents(
  self : BudgetedWriterStats,
) -> Int {
  self.indexed_documents
}

///|
pub fn BudgetedWriterStats::peak_estimated_bytes(
  self : BudgetedWriterStats,
) -> Int {
  self.peak_estimated_bytes
}

///|
fn estimated_value_bytes(value : FieldValue) -> Int {
  match value {
    @core.Text(text) | @core.Keyword(text) => @utf8.encode(text).length() + 16
    @core.Bytes(bytes) => bytes.length() + 16
    @core.I64(_) | @core.U64(_) | @core.F64(_) | @core.Date(_) => 16
    @core.Bool(_) => 8
  }
}

///|
/// Estimates the mutable indexing footprint contributed by a document. The
/// multiplier accounts for term text, postings, positions, offsets and stored
/// fields and intentionally errs high so automatic flush happens early.
pub fn estimate_document_memory(document : Document) -> Int {
  let mut bytes = 64
  for index in 0.. bytes += @utf8.encode(text).length() * 3 + 48
      None => ()
    }
  }
  for index in 0.. bytes += estimated_value_bytes(value)
      None => ()
    }
  }
  bytes
}

///|
/// Schema-aware Segment writer that automatically seals segments at a memory
/// budget. It is deterministic on every MoonBit backend and is the unit that
/// native executors can shard across independent indexing lanes.
pub struct BudgetedSegmentWriter {
  schema : Schema
  tokenizers : TokenizerManager
  memory_budget_bytes : Int
  mut current : SegmentWriter[WhitespaceAnalyzer]
  mut current_estimated_bytes : Int
  mut current_documents : Int
  segments : Array[Segment]
  mut indexed_documents : Int
  mut peak_estimated_bytes : Int
}

///|
pub fn BudgetedSegmentWriter::new(
  schema : Schema,
  tokenizers : TokenizerManager,
  memory_budget_bytes : Int,
) -> BudgetedSegmentWriter raise AnalysisError {
  guard memory_budget_bytes > 0 else {
    abort("index memory budget must be positive")
  }
  let snapshot = tokenizers.snapshot()
  {
    schema,
    tokenizers: snapshot,
    memory_budget_bytes,
    current: SegmentWriter::with_schema_and_tokenizers(schema, snapshot),
    current_estimated_bytes: 0,
    current_documents: 0,
    segments: [],
    indexed_documents: 0,
    peak_estimated_bytes: 0,
  }
}

///|
fn BudgetedSegmentWriter::seal_current(
  self : BudgetedSegmentWriter,
) -> Unit raise AnalysisError {
  if self.current_documents == 0 {
    return
  }
  self.segments.push(self.current.finish())
  self.current = SegmentWriter::with_schema_and_tokenizers(
    self.schema,
    self.tokenizers,
  )
  self.current_estimated_bytes = 0
  self.current_documents = 0
}

///|
/// Adds a document and flushes before it when the next document would exceed
/// the budget. A single oversized document is accepted into its own segment.
pub fn BudgetedSegmentWriter::add_document(
  self : BudgetedSegmentWriter,
  document : Document,
) -> DocId raise AnalysisError {
  let estimated = estimate_document_memory(document)
  if self.current_documents > 0 &&
    self.current_estimated_bytes + estimated > self.memory_budget_bytes {
    self.seal_current()
  }
  let doc_id = self.current.add_document(document)
  self.current_documents += 1
  self.indexed_documents += 1
  self.current_estimated_bytes += estimated
  if self.current_estimated_bytes > self.peak_estimated_bytes {
    self.peak_estimated_bytes = self.current_estimated_bytes
  }
  if self.current_estimated_bytes >= self.memory_budget_bytes {
    self.seal_current()
  }
  doc_id
}

///|
pub fn BudgetedSegmentWriter::flush(
  self : BudgetedSegmentWriter,
) -> Unit raise AnalysisError {
  self.seal_current()
}

///|
pub fn BudgetedSegmentWriter::finish(
  self : BudgetedSegmentWriter,
) -> ReadOnlyArray[Segment] raise AnalysisError {
  self.seal_current()
  ReadOnlyArray::from_array(self.segments.copy())
}

///|
pub fn BudgetedSegmentWriter::stats(
  self : BudgetedSegmentWriter,
) -> BudgetedWriterStats {
  {
    memory_budget_bytes: self.memory_budget_bytes,
    flushed_segments: self.segments.length(),
    indexed_documents: self.indexed_documents,
    peak_estimated_bytes: self.peak_estimated_bytes,
  }
}

///|
/// Portable capability report for segment-sharded indexing. MoonSearch keeps
/// the sharding contract backend-neutral; the current built-in executor is
/// cooperative and therefore reports effective parallelism 1.
pub struct IndexingExecutionPlan {
  requested_parallelism : Int
  effective_parallelism : Int
  degraded : Bool
}

///|
pub fn IndexingExecutionPlan::portable(
  requested_parallelism : Int,
) -> IndexingExecutionPlan {
  guard requested_parallelism > 0 else {
    abort("requested indexing parallelism must be positive")
  }
  {
    requested_parallelism,
    effective_parallelism: 1,
    degraded: requested_parallelism > 1,
  }
}

///|
pub fn IndexingExecutionPlan::requested_parallelism(
  self : IndexingExecutionPlan,
) -> Int {
  self.requested_parallelism
}

///|
pub fn IndexingExecutionPlan::effective_parallelism(
  self : IndexingExecutionPlan,
) -> Int {
  self.effective_parallelism
}

///|
pub fn IndexingExecutionPlan::is_degraded(self : IndexingExecutionPlan) -> Bool {
  self.degraded
}