///|
/// One completed-part integrity record. CRC-32 is intentionally exposed as a
/// fast corruption-detection signal, not as a cryptographic digest.
pub(all) struct PartDigest {
  part : UploadPart
  bytes : Int64
  crc32 : Int64
} derive(Debug, Eq)

///|
/// Aggregate redacted integrity data for a completed sequence of parts.
pub(all) struct AuditSummary {
  parts : Int
  payload_bytes : Int64
  combined_crc32 : Int64
} derive(Debug, Eq)

///|
pub(all) struct ActiveDigest {
  part : UploadPart
  mut bytes : Int64
  mut state : Int64
}

///|
/// Incremental per-part CRC-32 recorder. It retains no payload bytes and may
/// be used directly or through `AuditedSink`.
pub struct UploadAudit {
  mut active : ActiveDigest?
  digests : Array[PartDigest]
  mut payload_bytes : Int64
  mut combined_state : Int64
}

///|
/// Start an incremental CRC-32 state.
pub fn crc32_initial() -> Int64 {
  0xFFFFFFFFL
}

///|
/// Advance a CRC-32 state with binary-safe data.
pub fn crc32_update(state : Int64, bytes : Bytes) -> Int64 {
  let mut value = state
  for index in 0..> 1) ^ 0xEDB88320L
      } else {
        value >> 1
      }
    }
  }
  value
}

///|
/// Finalize an incremental CRC-32 state into its conventional unsigned value.
pub fn crc32_finish(state : Int64) -> Int64 {
  state ^ 0xFFFFFFFFL
}

///|
pub fn UploadAudit::new() -> UploadAudit {
  {
    active: None,
    digests: [],
    payload_bytes: 0L,
    combined_state: crc32_initial(),
  }
}

///|
pub fn UploadAudit::begin(
  self : UploadAudit,
  part : UploadPart,
) -> Unit raise UploadError {
  if self.active is Some(_) {
    raise UnexpectedEvent(
      reason="audit received PartBegin while another part is active",
    )
  }
  self.active = Some({ part, bytes: 0L, state: crc32_initial() })
}

///|
pub fn UploadAudit::data(
  self : UploadAudit,
  bytes : Bytes,
) -> Unit raise UploadError {
  match self.active {
    Some(active) => {
      active.bytes += bytes.length().to_int64()
      active.state = crc32_update(active.state, bytes)
      self.payload_bytes += bytes.length().to_int64()
      self.combined_state = crc32_update(self.combined_state, bytes)
    }
    None =>
      raise UnexpectedEvent(reason="audit received data without an active part")
  }
}

///|
pub fn UploadAudit::end(
  self : UploadAudit,
  part : UploadPart,
) -> Unit raise UploadError {
  let active = match self.active {
    Some(active) => active
    None =>
      raise UnexpectedEvent(
        reason="audit received PartEnd without an active part",
      )
  }
  if active.part.meta.index != part.meta.index {
    raise UnexpectedEvent(
      reason="audit PartEnd index does not match active part",
    )
  }
  self.digests.push({
    part: active.part,
    bytes: active.bytes,
    crc32: crc32_finish(active.state),
  })
  self.active = None
}

///|
pub fn UploadAudit::digests(self : UploadAudit) -> Array[PartDigest] {
  self.digests.copy()
}

///|
pub fn UploadAudit::summary(
  self : UploadAudit,
) -> AuditSummary raise UploadError {
  if self.active is Some(_) {
    raise UnexpectedEvent(reason="audit summary requested with an active part")
  }
  {
    parts: self.digests.length(),
    payload_bytes: self.payload_bytes,
    combined_crc32: crc32_finish(self.combined_state),
  }
}

///|
/// A transactional sink decorator that records checksums while forwarding all
/// callbacks to an application-owned delegate.
pub struct AuditedSink {
  sink : UploadSink
  audit : UploadAudit
}

///|
fn result_of(
  action : () -> Unit raise UploadError,
) -> Result[Unit, UploadError] {
  try {
    action()
    Ok(())
  } catch {
    error => Err(error)
  }
}

///|
pub fn AuditedSink::new(delegate : UploadSink) -> AuditedSink {
  let audit = UploadAudit::new()
  let sink = UploadSink::new(
    part => {
      match (delegate.on_begin)(part) {
        Ok(_) => result_of(() => audit.begin(part))
        Err(error) => Err(error)
      }
    },
    (part, bytes) => {
      let recorded = result_of(() => audit.data(bytes))
      match recorded {
        Ok(_) => (delegate.on_data)(part, bytes)
        Err(error) => Err(error)
      }
    },
    part => {
      match (delegate.on_end)(part) {
        Ok(_) => result_of(() => audit.end(part))
        Err(error) => Err(error)
      }
    },
    receipt => (delegate.on_commit)(receipt),
    error => (delegate.on_abort)(error),
  )
  { sink, audit }
}

///|
pub fn AuditedSink::sink(self : AuditedSink) -> UploadSink {
  self.sink
}

///|
pub fn AuditedSink::digests(self : AuditedSink) -> Array[PartDigest] {
  self.audit.digests()
}

///|
pub fn AuditedSink::summary(
  self : AuditedSink,
) -> AuditSummary raise UploadError {
  self.audit.summary()
}