///|
/// Failures produced by Avro Single-Object Encoding envelopes.
pub(all) suberror SingleObjectError {
  InvalidMagic
  TruncatedHeader(actual~ : Int, required~ : Int)
  UnknownFingerprint(UInt64)
  FingerprintMismatch(expected~ : UInt64, actual~ : UInt64)
  Datum(@codec.CodecError)
} derive(Debug, Eq)

///|
/// Return the two-byte Avro Single-Object Encoding marker (`C3 01`).
pub fn magic() -> Bytes {
  [0xc3, 0x01]
}

///|
/// Read a little-endian CRC-64-AVRO fingerprint from the eight bytes following
/// the single-object marker. The caller has already checked the input length.
fn read_fingerprint_unchecked(input : Bytes) -> UInt64 {
  let mut result = 0UL
  for index in 0..<8 {
    result = result | (input[index + 2].to_uint64() << (index * 8))
  }
  result
}

///|
/// Validate the ten-byte envelope prefix and return its schema fingerprint.
pub fn fingerprint(input : Bytes) -> UInt64 raise SingleObjectError {
  if input.length() < 2 {
    raise TruncatedHeader(actual=input.length(), required=2)
  }
  if input[0] != 0xc3 || input[1] != 0x01 {
    raise InvalidMagic
  }
  if input.length() < 10 {
    raise TruncatedHeader(actual=input.length(), required=10)
  }
  read_fingerprint_unchecked(input)
}

///|
/// Encode one datum using Avro Single-Object Encoding. The result contains the
/// marker, the schema's CRC-64-AVRO fingerprint in little-endian order, and one
/// ordinary Avro binary datum payload.
pub fn encode(
  schema : @schema.Schema,
  datum : @codec.Datum,
) -> Bytes raise SingleObjectError {
  let payload = @codec.encode(schema, datum) catch {
    error => raise Datum(error)
  }
  let output = @codec.Encoder::new(
    limits=@codec.CodecLimits::new(max_bytes=10 + payload.length()),
  )
  output.write_raw(magic()) catch {
    error => raise Datum(error)
  }
  output.write_raw(schema.fingerprint64().to_le_bytes()) catch {
    error => raise Datum(error)
  }
  output.write_raw(payload) catch {
    error => raise Datum(error)
  }
  output.to_bytes()
}

///|
/// Decode a Single-Object Encoding value with an expected schema. A mismatch
/// is reported before the datum payload is interpreted.
pub fn decode(
  schema : @schema.Schema,
  input : Bytes,
  limits? : @codec.CodecLimits = @codec.CodecLimits::new(),
) -> @codec.Datum raise SingleObjectError {
  let actual = fingerprint(input)
  let expected = schema.fingerprint64()
  if actual != expected {
    raise FingerprintMismatch(expected~, actual~)
  }
  let payload = input[10:].to_owned()
  @codec.decode(schema, payload, limits~) catch {
    error => raise Datum(error)
  }
}

///|
/// Resolve the writer schema by fingerprint and decode one Single-Object
/// Encoding value. The resolver can use a registry, generated table or local
/// map without coupling this package to a particular storage mechanism.
pub fn decode_resolved(
  input : Bytes,
  resolve : (UInt64) -> @schema.Schema?,
  limits? : @codec.CodecLimits = @codec.CodecLimits::new(),
) -> (@schema.Schema, @codec.Datum) raise SingleObjectError {
  let writer_fingerprint = fingerprint(input)
  let schema = match resolve(writer_fingerprint) {
    Some(schema) => schema
    None => raise UnknownFingerprint(writer_fingerprint)
  }
  let payload = input[10:].to_owned()
  let datum = @codec.decode(schema, payload, limits~) catch {
    error => raise Datum(error)
  }
  (schema, datum)
}

///|
/// Return whether an encoded value carries the supplied schema fingerprint.
/// Malformed and truncated envelopes return `false` rather than raising.
pub fn matches_schema(input : Bytes, schema : @schema.Schema) -> Bool {
  let value = fingerprint(input) catch { _ => return false }
  value == schema.fingerprint64()
}