///|
/// Bounds for recursive legacy multipart/mixed expansion.
pub(all) struct MixedLimits {
  max_depth : Int
  max_parts_per_container : Int
  max_container_bytes : Int64
  max_part_bytes : Int64
} derive(Debug, Eq)

///|
pub fn default_mixed_limits() -> MixedLimits {
  {
    max_depth: 4,
    max_parts_per_container: 64,
    max_container_bytes: 8388608L,
    max_part_bytes: 4194304L,
  }
}

///|
/// One event after recursively expanding legacy multipart/mixed containers.
/// Container envelopes are suppressed; leaf events retain their original
/// protocol metadata, depth, and immediate container part index.
pub(all) struct MixedEvent {
  depth : Int
  parent_part_index : Int?
  event : @model.MultipartEvent
} derive(Debug, Eq)

///|
/// One owned leaf part returned by the compatibility batch helper.
pub(all) struct MixedPart {
  depth : Int
  parent_part_index : Int?
  meta : @model.PartMeta
  body : Bytes
} derive(Debug, Eq)

///|
pub(all) suberror MixedError {
  InvalidLimits(reason~ : String)
  DepthExceeded(limit~ : Int, observed~ : Int)
  ContainerTooLarge(limit~ : Int64, observed~ : Int64)
  InvalidContainer(reason~ : String)
  UnexpectedEvent(reason~ : String)
  Protocol(error~ : @model.MultipartError)
} derive(Debug, Eq)

///|
pub struct ActiveMixed {
  part_index : Int
  parser : @parser.MultipartParser
  child : MixedStream
  mut observed_bytes : Int64
}

///|
pub enum ActivePart {
  Leaf(Int)
  Container(ActiveMixed)
}

///|
pub struct MixedStream {
  limits : MixedLimits
  depth : Int
  parent_part_index : Int?
  emit_message_end : Bool
  mut active : ActivePart?
  mut completed : Bool
}

///|
fn validate_mixed_limits(limits : MixedLimits) -> Unit raise MixedError {
  if limits.max_depth <= 0 || limits.max_parts_per_container <= 0 {
    raise InvalidLimits(reason="mixed depth and part limits must be positive")
  }
  if limits.max_container_bytes <= 0L || limits.max_part_bytes <= 0L {
    raise InvalidLimits(reason="mixed byte limits must be positive")
  }
}

///|
fn nested_parser_limits(limits : MixedLimits) -> @model.Limits {
  {
    ..@model.default_limits(),
    max_total_bytes: limits.max_container_bytes,
    max_parts: limits.max_parts_per_container,
    max_part_bytes: limits.max_part_bytes,
  }
}

///|
fn mixed_boundary(content_type : String) -> String raise MixedError {
  let media_type = @syntax.parse_multipart_content_type(content_type) catch {
    error => raise InvalidContainer(reason=@model.redact_error(error).code)
  }
  if media_type.media_type != "multipart/mixed" {
    raise InvalidContainer(reason="expected multipart/mixed content type")
  }
  media_type.boundary
}

///|
fn MixedStream::new_level(
  limits : MixedLimits,
  depth : Int,
  parent_part_index : Int?,
  emit_message_end : Bool,
) -> MixedStream {
  {
    limits,
    depth,
    parent_part_index,
    emit_message_end,
    active: None,
    completed: false,
  }
}

///|
/// Create an incremental transformer for top-level multipart parser events.
pub fn MixedStream::new(
  limits? : MixedLimits = default_mixed_limits(),
) -> MixedStream raise MixedError {
  validate_mixed_limits(limits)
  MixedStream::new_level(limits, 0, None, true)
}

///|
fn MixedStream::append_leaf(
  self : MixedStream,
  event : @model.MultipartEvent,
  output : Array[MixedEvent],
) -> Unit {
  output.push({
    depth: self.depth,
    parent_part_index: self.parent_part_index,
    event,
  })
}

///|
fn MixedStream::begin(
  self : MixedStream,
  meta : @model.PartMeta,
  output : Array[MixedEvent],
) -> Unit raise MixedError {
  if self.completed || self.active is Some(_) {
    raise UnexpectedEvent(
      reason="PartBegin is not valid in the current mixed state",
    )
  }
  match meta.content_type {
    Some(content_type) if content_type.to_lower().has_prefix("multipart/mixed") => {
      let child_depth = self.depth + 1
      if child_depth > self.limits.max_depth {
        raise DepthExceeded(limit=self.limits.max_depth, observed=child_depth)
      }
      let boundary = mixed_boundary(content_type)
      let parser = @parser.MultipartParser::new_mixed(
        boundary,
        meta.name,
        nested_parser_limits(self.limits),
      ) catch {
        error => raise Protocol(error~)
      }
      let child = MixedStream::new_level(
        self.limits,
        child_depth,
        Some(meta.index),
        false,
      )
      self.active = Some(
        Container({ part_index: meta.index, parser, child, observed_bytes: 0L }),
      )
    }
    _ => {
      self.active = Some(Leaf(meta.index))
      self.append_leaf(PartBegin(meta), output)
    }
  }
}

///|
fn MixedStream::data(
  self : MixedStream,
  bytes : Bytes,
  output : Array[MixedEvent],
) -> Unit raise MixedError {
  match self.active {
    Some(Leaf(_)) => self.append_leaf(PartData(bytes), output)
    Some(Container(container)) => {
      let observed = container.observed_bytes + bytes.length().to_int64()
      if observed > self.limits.max_container_bytes {
        raise ContainerTooLarge(
          limit=self.limits.max_container_bytes,
          observed~,
        )
      }
      container.observed_bytes = observed
      let child_events = container.parser.feed(bytes) catch {
        error => raise Protocol(error~)
      }
      output.append(container.child.feed(child_events))
    }
    None => raise UnexpectedEvent(reason="PartData without an active part")
  }
}

///|
fn MixedStream::end(
  self : MixedStream,
  index : Int,
  output : Array[MixedEvent],
) -> Unit raise MixedError {
  let active = match self.active {
    Some(active) => active
    None => raise UnexpectedEvent(reason="PartEnd without an active part")
  }
  match active {
    Leaf(active_index) => {
      if active_index != index {
        raise UnexpectedEvent(reason="PartEnd index does not match active leaf")
      }
      self.append_leaf(PartEnd(index), output)
    }
    Container(container) => {
      if container.part_index != index {
        raise UnexpectedEvent(
          reason="PartEnd index does not match active container",
        )
      }
      let child_events = container.parser.finish() catch {
        error => raise Protocol(error~)
      }
      output.append(container.child.feed(child_events))
    }
  }
  self.active = None
}

///|
/// Transform one batch of top-level parser events. Feed this method after each
/// `MultipartParser::feed` result; nested leaf data can be emitted before the
/// outer container has ended.
pub fn MixedStream::feed(
  self : MixedStream,
  events : Array[@model.MultipartEvent],
) -> Array[MixedEvent] raise MixedError {
  let output : Array[MixedEvent] = []
  for event in events {
    match event {
      PartBegin(meta) => self.begin(meta, output)
      PartData(bytes) => self.data(bytes, output)
      PartEnd(index) => self.end(index, output)
      MessageEnd => {
        if self.active is Some(_) {
          raise UnexpectedEvent(
            reason="MessageEnd before the active part ended",
          )
        }
        self.completed = true
        if self.emit_message_end {
          self.append_leaf(MessageEnd, output)
        }
      }
    }
  }
  output
}

///|
/// Require that the event stream has completed without a partial container.
pub fn MixedStream::finish(self : MixedStream) -> Unit raise MixedError {
  if self.active is Some(_) || !self.completed {
    raise UnexpectedEvent(reason="mixed stream ended before MessageEnd")
  }
}

///|
fn collect_expanded_parts(
  events : Array[MixedEvent],
) -> Array[MixedPart] raise MixedError {
  let output : Array[MixedPart] = []
  let mut active : (Int, Int, Int?, @model.PartMeta, @buffer.Buffer)? = None
  for mixed in events {
    match mixed.event {
      PartBegin(meta) => {
        if active is Some(_) {
          raise UnexpectedEvent(reason="overlapping leaf parts in mixed output")
        }
        active = Some(
          (
            meta.index,
            mixed.depth,
            mixed.parent_part_index,
            meta,
            Buffer(size_hint=256),
          ),
        )
      }
      PartData(bytes) =>
        match active {
          Some((_, _, _, _, body)) => body.write_bytes(bytes)
          None => raise UnexpectedEvent(reason="leaf data without a leaf begin")
        }
      PartEnd(index) => {
        let (active_index, depth, parent_part_index, meta, body) = match
          active {
          Some(active) => active
          None => raise UnexpectedEvent(reason="leaf end without a leaf begin")
        }
        if index != active_index {
          raise UnexpectedEvent(
            reason="leaf end index does not match leaf begin",
          )
        }
        output.push({ depth, parent_part_index, meta, body: body.to_bytes() })
        active = None
      }
      MessageEnd => ()
    }
  }
  if active is Some(_) {
    raise UnexpectedEvent(reason="mixed output ended with an unfinished leaf")
  }
  output
}

///|
/// Compatibility helper for callers that already own a complete container.
/// New upload paths should use `MixedStream` directly to avoid buffering the
/// nested body before parsing it.
pub fn expand_mixed(
  content_type : String,
  body : Bytes,
  inherited_name : String,
  limits? : MixedLimits = default_mixed_limits(),
) -> Array[MixedPart] raise MixedError {
  if inherited_name.is_empty() {
    raise InvalidContainer(
      reason="multipart/mixed requires an inherited field name",
    )
  }
  let stream = MixedStream::new(limits~)
  let events = stream.feed([
    PartBegin({
      index: 0,
      name: inherited_name,
      filename: None,
      content_type: Some(content_type),
      headers: [],
    }),
    PartData(body),
    PartEnd(0),
    MessageEnd,
  ])
  stream.finish()
  collect_expanded_parts(events)
}