///|
pub struct MultipartParser {
  priv boundary : String
  priv mixed_parent_name : String?
  priv limits : @model.Limits
  priv mut phase_ : @model.ParserPhase
  priv mut bytes_seen_ : Int64
  priv mut next_part_index : Int
  priv mut current_part_index : Int
  priv mut current_part_bytes : Int64
  priv mut parts_completed : Int
  priv mut header_bytes : Int64
  priv mut body_bytes : Int64
  priv mut events_emitted : Int64
  priv mut max_pending_body_bytes : Int
  priv mut max_matcher_retained_bytes : Int
  priv mut max_buffered_bytes : Int
  priv header_buffer : @buffer.Buffer
  priv mut header_line_bytes : Int
  priv mut header_saw_cr : Bool
  priv body_pending : @buffer.Buffer
  priv mut matcher : @boundary.Matcher
}

///|
const BODY_EVENT_BUFFER_BYTES : Int = 8192

///|
fn MultipartParser::invalid_state(
  self : MultipartParser,
  operation : String,
) -> @model.MultipartError {
  InvalidState(operation~, phase=self.phase_)
}

///|
fn[T] MultipartParser::raise_failed(
  self : MultipartParser,
  error : @model.MultipartError,
) -> T raise @model.MultipartError {
  self.phase_ = Failed
  raise error
}

///|
pub fn MultipartParser::new(
  boundary : String,
  limits : @model.Limits,
) -> MultipartParser raise @model.MultipartError {
  MultipartParser::new_with_parent(boundary, None, limits)
}

///|
fn MultipartParser::new_with_parent(
  boundary : String,
  mixed_parent_name : String?,
  limits : @model.Limits,
) -> MultipartParser raise @model.MultipartError {
  @syntax.validate_boundary(boundary)
  @model.validate_limits(limits)
  {
    boundary,
    mixed_parent_name,
    limits,
    phase_: Preamble,
    bytes_seen_: 0L,
    next_part_index: 0,
    current_part_index: -1,
    current_part_bytes: 0L,
    parts_completed: 0,
    header_bytes: 0L,
    body_bytes: 0L,
    events_emitted: 0L,
    max_pending_body_bytes: 0,
    max_matcher_retained_bytes: 0,
    max_buffered_bytes: 0,
    header_buffer: Buffer(size_hint=256),
    header_line_bytes: 0,
    header_saw_cr: false,
    body_pending: Buffer(size_hint=boundary.length() + 8),
    matcher: @boundary.Matcher::new(boundary),
  }
}

///|
pub fn MultipartParser::new_mixed(
  boundary : String,
  parent_name : String,
  limits : @model.Limits,
) -> MultipartParser raise @model.MultipartError {
  if parent_name.is_empty() {
    raise InvalidContentType(
      reason="multipart/mixed parent name must not be empty",
    )
  }
  MultipartParser::new_with_parent(boundary, Some(parent_name), limits)
}

///|
fn MultipartParser::reset_headers(self : MultipartParser) -> Unit {
  self.header_buffer.reset()
  self.header_line_bytes = 0
  self.header_saw_cr = false
}

///|
fn MultipartParser::enter_headers(self : MultipartParser) -> Unit {
  self.phase_ = Headers
  self.reset_headers()
}

///|
fn MultipartParser::reset_body_matcher(self : MultipartParser) -> Unit {
  self.body_pending.reset()
  self.matcher = @boundary.Matcher::new(self.boundary)
  self.current_part_bytes = 0L
}

///|
fn MultipartParser::observe_buffer_high_water(self : MultipartParser) -> Unit {
  let matcher_retained = self.matcher.stats().max_retained_bytes
  self.max_matcher_retained_bytes = Int::max(
    self.max_matcher_retained_bytes,
    matcher_retained,
  )
  self.max_buffered_bytes = Int::max(
    self.max_buffered_bytes,
    self.body_pending.length() + matcher_retained,
  )
}

///|
fn MultipartParser::enter_body(self : MultipartParser) -> Unit {
  self.phase_ = Body
  self.reset_body_matcher()
}

///|
fn MultipartParser::emit_body(
  self : MultipartParser,
  bytes : Bytes,
  events : Array[@model.MultipartEvent],
) -> Unit raise @model.MultipartError {
  if bytes.is_empty() {
    return
  }
  let observed = self.current_part_bytes + bytes.length().to_int64()
  if observed > self.limits.max_part_bytes {
    self.raise_failed(
      LimitExceeded(kind=PartBytes, limit=self.limits.max_part_bytes, observed~),
    )
  }
  self.current_part_bytes = observed
  self.body_bytes += bytes.length().to_int64()
  events.push(PartData(bytes))
}

///|
fn MultipartParser::body_event_chunk_size(self : MultipartParser) -> Int {
  if self.limits.max_part_bytes >= BODY_EVENT_BUFFER_BYTES.to_int64() {
    BODY_EVENT_BUFFER_BYTES
  } else {
    Int::max(1, self.limits.max_part_bytes.to_int())
  }
}

///|
fn MultipartParser::flush_body_prefix(
  self : MultipartParser,
  count : Int,
  events : Array[@model.MultipartEvent],
) -> Unit raise @model.MultipartError {
  if count <= 0 {
    return
  }
  let bytes = self.body_pending.to_bytes()
  let emitted = bytes.view(end=count).to_owned()
  let rest : Bytes = if count >= bytes.length() {
    b""
  } else {
    bytes.view(start=count).to_owned()
  }
  self.body_pending.reset()
  self.body_pending.write_bytes(rest)
  self.emit_body(emitted, events)
}

///|
fn MultipartParser::begin_part(
  self : MultipartParser,
  events : Array[@model.MultipartEvent],
) -> Unit raise @model.MultipartError {
  let raw_with_terminator = self.header_buffer.to_bytes()
  let raw = raw_with_terminator
    .view(end=raw_with_terminator.length() - 4)
    .to_owned()
  let part_index = self.next_part_index
  let meta = try {
    match self.mixed_parent_name {
      Some(parent_name) =>
        @syntax.parse_mixed_part_headers(raw, part_index, parent_name)
      None => @syntax.parse_part_headers(raw, part_index)
    }
  } catch {
    error => self.raise_failed(error)
  }
  let header_count = meta.headers.length()
  if header_count > self.limits.max_headers_per_part {
    self.raise_failed(
      LimitExceeded(
        kind=HeadersPerPart,
        limit=self.limits.max_headers_per_part.to_int64(),
        observed=header_count.to_int64(),
      ),
    )
  }
  let observed_parts = self.next_part_index + 1
  if observed_parts > self.limits.max_parts {
    self.raise_failed(
      LimitExceeded(
        kind=Parts,
        limit=self.limits.max_parts.to_int64(),
        observed=observed_parts.to_int64(),
      ),
    )
  }
  self.next_part_index = observed_parts
  self.current_part_index = part_index
  self.header_bytes += raw.length().to_int64()
  events.push(PartBegin(meta))
  self.enter_body()
}

///|
fn MultipartParser::header_is_complete(self : MultipartParser) -> Bool {
  let bytes = self.header_buffer.to_bytes()
  let length = bytes.length()
  length >= 4 &&
  bytes.at(length - 4) == b'\r' &&
  bytes.at(length - 3) == b'\n' &&
  bytes.at(length - 2) == b'\r' &&
  bytes.at(length - 1) == b'\n'
}

///|
fn MultipartParser::consume_header(
  self : MultipartParser,
  byte : Byte,
  events : Array[@model.MultipartEvent],
) -> Unit raise @model.MultipartError {
  self.header_buffer.write_byte(byte)
  if self.header_buffer.length() > self.limits.max_header_bytes + 4 {
    self.raise_failed(
      LimitExceeded(
        kind=HeaderBytes,
        limit=self.limits.max_header_bytes.to_int64(),
        observed=(self.header_buffer.length() - 4).to_int64(),
      ),
    )
  }
  if self.header_saw_cr {
    if byte != b'\n' {
      self.raise_failed(
        MalformedHeader(
          part_index=self.next_part_index,
          reason="header line must end with CRLF",
        ),
      )
    }
    self.header_saw_cr = false
    self.header_line_bytes = 0
  } else if byte == b'\r' {
    self.header_saw_cr = true
  } else if byte == b'\n' {
    self.raise_failed(
      MalformedHeader(
        part_index=self.next_part_index,
        reason="bare LF in header block",
      ),
    )
  } else {
    self.header_line_bytes += 1
    if self.header_line_bytes > self.limits.max_header_line_bytes {
      self.raise_failed(
        LimitExceeded(
          kind=HeaderLineBytes,
          limit=self.limits.max_header_line_bytes.to_int64(),
          observed=self.header_line_bytes.to_int64(),
        ),
      )
    }
  }
  if self.header_is_complete() {
    self.begin_part(events)
  }
}

///|
fn MultipartParser::consume_preamble(
  self : MultipartParser,
  byte : Byte,
) -> Unit {
  let single : Bytes = [byte]
  let matches = self.matcher.scan(single)
  self.observe_buffer_high_water()
  if matches.length() > 0 {
    match matches[0].kind {
      NextPart => self.enter_headers()
      Closing => self.phase_ = Epilogue
    }
  }
}

///|
fn MultipartParser::consume_body(
  self : MultipartParser,
  byte : Byte,
  events : Array[@model.MultipartEvent],
) -> Unit raise @model.MultipartError {
  self.body_pending.write_byte(byte)
  self.max_pending_body_bytes = Int::max(
    self.max_pending_body_bytes,
    self.body_pending.length(),
  )
  let single : Bytes = [byte]
  let matches = self.matcher.scan(single)
  self.observe_buffer_high_water()
  if matches.length() == 0 {
    let keep = self.matcher.retained_prefix_len()
    let safe_bytes = self.body_pending.length() - keep
    let event_chunk_size = self.body_event_chunk_size()
    if safe_bytes >= event_chunk_size {
      self.flush_body_prefix(event_chunk_size, events)
    }
    return
  }
  let found = matches[0]
  let bytes = self.body_pending.to_bytes()
  let delimiter_length = (found.end_ - found.start).to_int()
  let body_length = bytes.length() - delimiter_length
  let false_initial = found.start == 0L &&
    bytes.length() >= 2 &&
    bytes.at(0) == b'-' &&
    bytes.at(1) == b'-'
  if false_initial {
    self.flush_body_prefix(bytes.length(), events)
    self.matcher = @boundary.Matcher::new(self.boundary)
    return
  }
  self.flush_body_prefix(body_length, events)
  self.body_pending.reset()
  events.push(PartEnd(self.current_part_index))
  self.parts_completed += 1
  self.current_part_index = -1
  match found.kind {
    NextPart => self.enter_headers()
    Closing => self.phase_ = Epilogue
  }
}

///|
fn MultipartParser::consume_byte(
  self : MultipartParser,
  byte : Byte,
  events : Array[@model.MultipartEvent],
) -> Unit raise @model.MultipartError {
  match self.phase_ {
    Preamble => self.consume_preamble(byte)
    Headers => self.consume_header(byte, events)
    Body => self.consume_body(byte, events)
    Epilogue => ()
    Done | Failed => self.raise_failed(self.invalid_state("feed"))
  }
}

///|
pub fn MultipartParser::feed(
  self : MultipartParser,
  chunk : Bytes,
) -> Array[@model.MultipartEvent] raise @model.MultipartError {
  if self.phase_ == Done || self.phase_ == Failed {
    raise self.invalid_state("feed")
  }
  let observed = self.bytes_seen_ + chunk.length().to_int64()
  if observed > self.limits.max_total_bytes {
    self.bytes_seen_ = observed
    self.raise_failed(
      LimitExceeded(
        kind=TotalBytes,
        limit=self.limits.max_total_bytes,
        observed~,
      ),
    )
  }
  let events : Array[@model.MultipartEvent] = []
  for byte in chunk {
    self.bytes_seen_ += 1L
    self.consume_byte(byte, events)
  }
  self.events_emitted += events.length().to_int64()
  events
}

///|
pub fn MultipartParser::finish(
  self : MultipartParser,
) -> Array[@model.MultipartEvent] raise @model.MultipartError {
  match self.phase_ {
    Epilogue => {
      self.phase_ = Done
      self.events_emitted += 1L
      [MessageEnd]
    }
    Done | Failed => raise self.invalid_state("finish")
    phase => self.raise_failed(UnexpectedEof(phase~, offset=self.bytes_seen_))
  }
}

///|
pub fn MultipartParser::phase(self : MultipartParser) -> @model.ParserPhase {
  self.phase_
}

///|
pub fn MultipartParser::bytes_seen(self : MultipartParser) -> Int64 {
  self.bytes_seen_
}

///|
pub fn MultipartParser::stats(self : MultipartParser) -> @model.ParserStats {
  {
    bytes_seen: self.bytes_seen_,
    parts_started: self.next_part_index,
    parts_completed: self.parts_completed,
    header_bytes: self.header_bytes,
    body_bytes: self.body_bytes,
    events_emitted: self.events_emitted,
    max_pending_body_bytes: self.max_pending_body_bytes,
    max_matcher_retained_bytes: self.max_matcher_retained_bytes,
    max_buffered_bytes: self.max_buffered_bytes,
  }
}