///|
pub struct UploadPipeline {
  parser : @parser.MultipartParser
  mixed : @mixed.MixedStream?
  policy : UploadPolicy
  sink : UploadSink
  mut active : UploadPart?
  mut active_bytes : Int64
  mut payload_bytes : Int64
  mut fields_started : Int
  mut files_started : Int
  mut fields_completed : Int
  mut files_completed : Int
  mut completed : Bool
  mut aborted : Bool
}

///|
fn validate_policy(policy : UploadPolicy) -> Unit raise UploadError {
  @model.validate_limits(policy.parser_limits) catch {
    error => raise Protocol(error~)
  }
  if policy.max_file_bytes <= 0L ||
    policy.max_field_bytes <= 0L ||
    policy.max_total_payload_bytes <= 0L {
    raise InvalidPolicy(reason="payload byte limits must be positive")
  }
  if policy.max_file_count <= 0 || policy.max_field_count <= 0 {
    raise InvalidPolicy(reason="field and file count limits must be positive")
  }
}

///|
pub fn UploadPipeline::new(
  boundary : String,
  sink : UploadSink,
  policy? : UploadPolicy = default_upload_policy(),
) -> UploadPipeline raise UploadError {
  validate_policy(policy)
  let parser = @parser.MultipartParser::new(boundary, policy.parser_limits) catch {
    error => raise Protocol(error~)
  }
  let mixed = if policy.allow_legacy_mixed {
    Some(@mixed.MixedStream::new() catch { error => raise Mixed(error~) })
  } else {
    None
  }
  {
    parser,
    mixed,
    policy,
    sink,
    active: None,
    active_bytes: 0L,
    payload_bytes: 0L,
    fields_started: 0,
    files_started: 0,
    fields_completed: 0,
    files_completed: 0,
    completed: false,
    aborted: false,
  }
}

///|
fn[T] UploadPipeline::abort(
  self : UploadPipeline,
  error : UploadError,
) -> T raise UploadError {
  if !self.aborted {
    self.aborted = true
    (self.sink.on_abort)(error)
  }
  raise error
}

///|
fn classify(meta : @model.PartMeta) -> UploadPartKind {
  match meta.filename {
    Some(_) => File
    None => Field
  }
}

///|
fn UploadPipeline::begin(
  self : UploadPipeline,
  part : UploadPart,
) -> Unit raise UploadError {
  if self.active is Some(_) || self.completed || self.aborted {
    self.abort(
      UnexpectedEvent(reason="PartBegin is not valid in pipeline state"),
    )
  }
  match part.kind {
    Field => {
      let observed = self.fields_started + 1
      if observed > self.policy.max_field_count {
        self.abort(
          PartCountExceeded(
            kind=Field,
            limit=self.policy.max_field_count,
            observed~,
          ),
        )
      }
      self.fields_started = observed
    }
    File => {
      let observed = self.files_started + 1
      if observed > self.policy.max_file_count {
        self.abort(
          PartCountExceeded(
            kind=File,
            limit=self.policy.max_file_count,
            observed~,
          ),
        )
      }
      self.files_started = observed
    }
  }
  raise_sink((self.sink.on_begin)(part)) catch {
    error => self.abort(error)
  }
  self.active = Some(part)
  self.active_bytes = 0L
}

///|
fn UploadPipeline::data(
  self : UploadPipeline,
  bytes : Bytes,
) -> Unit raise UploadError {
  let part = match self.active {
    Some(part) => part
    None => self.abort(UnexpectedEvent(reason="PartData without active part"))
  }
  let observed = self.active_bytes + bytes.length().to_int64()
  let limit = match part.kind {
    Field => self.policy.max_field_bytes
    File => self.policy.max_file_bytes
  }
  if observed > limit {
    self.abort(PartTooLarge(part_index=part.meta.index, limit~, observed~))
  }
  let observed_payload = self.payload_bytes + bytes.length().to_int64()
  if observed_payload > self.policy.max_total_payload_bytes {
    self.abort(
      TotalPayloadTooLarge(
        limit=self.policy.max_total_payload_bytes,
        observed=observed_payload,
      ),
    )
  }
  raise_sink((self.sink.on_data)(part, bytes)) catch {
    error => self.abort(error)
  }
  self.active_bytes = observed
  self.payload_bytes = observed_payload
}

///|
fn UploadPipeline::end(
  self : UploadPipeline,
  index : Int,
) -> Unit raise UploadError {
  let part = match self.active {
    Some(part) => part
    None => self.abort(UnexpectedEvent(reason="PartEnd without active part"))
  }
  if part.meta.index != index {
    self.abort(
      UnexpectedEvent(reason="PartEnd index does not match active part"),
    )
  }
  raise_sink((self.sink.on_end)(part)) catch {
    error => self.abort(error)
  }
  match part.kind {
    Field => self.fields_completed += 1
    File => self.files_completed += 1
  }
  self.active = None
  self.active_bytes = 0L
}

///|
fn UploadPipeline::consume_mixed(
  self : UploadPipeline,
  events : Array[@mixed.MixedEvent],
) -> Unit raise UploadError {
  for mixed in events {
    match mixed.event {
      PartBegin(meta) =>
        self.begin({
          meta,
          kind: classify(meta),
          depth: mixed.depth,
          parent_part_index: mixed.parent_part_index,
        })
      PartData(bytes) => self.data(bytes)
      PartEnd(index) => self.end(index)
      MessageEnd => ()
    }
  }
}

///|
fn UploadPipeline::consume_plain(
  self : UploadPipeline,
  events : Array[@model.MultipartEvent],
) -> Unit raise UploadError {
  for event in events {
    match event {
      PartBegin(meta) =>
        self.begin({
          meta,
          kind: classify(meta),
          depth: 0,
          parent_part_index: None,
        })
      PartData(bytes) => self.data(bytes)
      PartEnd(index) => self.end(index)
      MessageEnd => ()
    }
  }
}

///|
fn UploadPipeline::consume(
  self : UploadPipeline,
  events : Array[@model.MultipartEvent],
) -> Unit raise UploadError {
  match self.mixed {
    Some(stream) => {
      let expanded = stream.feed(events) catch {
        error => self.abort(Mixed(error~))
      }
      self.consume_mixed(expanded)
    }
    None => self.consume_plain(events)
  }
}

///|
/// Feed a network, file, or application-owned chunk into the upload pipeline.
pub fn UploadPipeline::feed(
  self : UploadPipeline,
  chunk : Bytes,
) -> Unit raise UploadError {
  if self.completed || self.aborted {
    self.abort(Aborted(reason="cannot feed a completed or aborted pipeline"))
  }
  let events = self.parser.feed(chunk) catch {
    error => self.abort(Protocol(error~))
  }
  self.consume(events)
}

///|
/// Terminate an in-flight transaction because the host source or request
/// lifecycle ended outside the multipart protocol. Repeated cancellation is
/// idempotent and never calls the sink's abort callback more than once.
pub fn UploadPipeline::cancel(self : UploadPipeline, reason~ : String) -> Unit {
  if !self.completed && !self.aborted {
    self.aborted = true
    (self.sink.on_abort)(Aborted(reason~))
  }
}

///|
/// Complete the parser and atomically commit the host application's sink.
pub fn UploadPipeline::finish(
  self : UploadPipeline,
) -> UploadReceipt raise UploadError {
  if self.completed || self.aborted {
    self.abort(Aborted(reason="cannot finish a completed or aborted pipeline"))
  }
  let events = self.parser.finish() catch {
    error => self.abort(Protocol(error~))
  }
  self.consume(events)
  if self.active is Some(_) {
    self.abort(UnexpectedEvent(reason="parser completed with active part"))
  }
  match self.mixed {
    Some(stream) => stream.finish() catch { error => self.abort(Mixed(error~)) }
    None => ()
  }
  let receipt = {
    wire_bytes: self.parser.bytes_seen(),
    payload_bytes: self.payload_bytes,
    fields_completed: self.fields_completed,
    files_completed: self.files_completed,
  }
  raise_sink((self.sink.on_commit)(receipt)) catch {
    error => self.abort(error)
  }
  self.completed = true
  receipt
}

///|
pub fn UploadPipeline::progress(self : UploadPipeline) -> UploadProgress {
  {
    wire_bytes: self.parser.bytes_seen(),
    payload_bytes: self.payload_bytes,
    fields_completed: self.fields_completed,
    files_completed: self.files_completed,
    active_part: self.active,
  }
}

///|
/// Return parser resource statistics for host-side metrics and admission logs.
pub fn UploadPipeline::parser_stats(
  self : UploadPipeline,
) -> @model.ParserStats {
  self.parser.stats()
}