// Payload conversion paired with the schema declared to the broker.

///|
/// Converts values to and from Pulsar message payloads. Pass `info()` to
/// producer/consumer creation, then use `encode` and `decode` for messages.
pub struct SchemaCodec[T] {
  priv info : SchemaInfo
  priv encode_value : (T) -> Bytes raise PulsarError
  priv decode_value : (Bytes) -> T raise PulsarError
}

///|
/// Use a custom payload codec with its broker schema declaration.
pub fn[T] SchemaCodec::new(
  info : SchemaInfo,
  encode : (T) -> Bytes raise PulsarError,
  decode : (Bytes) -> T raise PulsarError,
) -> SchemaCodec[T] {
  { info, encode_value: encode, decode_value: decode, }
}

///|
/// UTF-8 string payloads with Pulsar's STRING schema.
pub fn SchemaCodec::string() -> SchemaCodec[String] {
  SchemaCodec::new(SchemaInfo::string(), fn(value) { @utf8.encode(value) }, payload => {
    @utf8.decode(payload) catch {
      _ => raise InvalidFrame("invalid UTF-8 string payload")
    }
  })
}

///|
pub fn SchemaCodec::bytes() -> SchemaCodec[Bytes] {
  SchemaCodec::new(SchemaInfo::bytes(), fn(value) { value }, fn(payload) {
    payload
  })
}

///|
/// INT8 values use `Int` because MoonBit has no signed eight-bit scalar.
pub fn SchemaCodec::int8() -> SchemaCodec[Int] {
  SchemaCodec::new(
    SchemaInfo::int8(),
    value => {
      guard value >= -128 && value <= 127 else {
        raise InvalidFrame("INT8 value out of range")
      }
      Bytes::from_array([value.to_byte()])
    },
    payload => {
      guard payload.length() == 1 else {
        raise InvalidFrame("INT8 payload must be one byte")
      }
      let value = payload[0].to_int()
      if value < 128 {
        value
      } else {
        value - 256
      }
    },
  )
}

///|
/// The official Go client uses little-endian integers; pass
/// `little_endian=true` to exchange INT16 values with it.
pub fn SchemaCodec::int16(little_endian? : Bool = false) -> SchemaCodec[Int] {
  SchemaCodec::new(
    SchemaInfo::int16(),
    value => {
      guard value >= -32768 && value <= 32767 else {
        raise InvalidFrame("INT16 value out of range")
      }
      if little_endian {
        Bytes::from_array([value.to_byte(), (value >> 8).to_byte()])
      } else {
        Bytes::from_array([(value >> 8).to_byte(), value.to_byte()])
      }
    },
    payload => {
      guard payload.length() == 2 else {
        raise InvalidFrame("INT16 payload must be two bytes")
      }
      let value = if little_endian {
        (payload[1].to_int() << 8) | payload[0].to_int()
      } else {
        (payload[0].to_int() << 8) | payload[1].to_int()
      }
      if value < 32768 {
        value
      } else {
        value - 65536
      }
    },
  )
}

///|
/// The official Go client uses little-endian integers; pass
/// `little_endian=true` to exchange INT32 values with it.
pub fn SchemaCodec::int32(little_endian? : Bool = false) -> SchemaCodec[Int] {
  SchemaCodec::new(
    SchemaInfo::int32(),
    fn(value) {
      let buf = @buffer.Buffer()
      if little_endian {
        let bits = value.reinterpret_as_uint()
        for shift in 0..<4 {
          buf.write_byte((bits >> (shift * 8)).to_byte())
        }
      } else {
        write_be32(buf, value.reinterpret_as_uint())
      }
      buf.to_bytes()
    },
    payload => {
      guard payload.length() == 4 else {
        raise InvalidFrame("INT32 payload must be four bytes")
      }
      if little_endian {
        ((payload[3].to_uint() << 24) |
        (payload[2].to_uint() << 16) |
        (payload[1].to_uint() << 8) |
        payload[0].to_uint()).reinterpret_as_int()
      } else {
        read_be32(payload, 0).reinterpret_as_int()
      }
    },
  )
}

///|
/// The official Go client uses little-endian integers; pass
/// `little_endian=true` to exchange INT64 values with it.
pub fn SchemaCodec::int64(little_endian? : Bool = false) -> SchemaCodec[Int64] {
  SchemaCodec::new(
    SchemaInfo::int64(),
    fn(value) {
      if little_endian {
        let bits = value.reinterpret_as_uint64()
        let buf = @buffer.Buffer()
        for shift in 0..<8 {
          buf.write_byte((bits >> (shift * 8)).to_byte())
        }
        buf.to_bytes()
      } else {
        value.reinterpret_as_uint64().to_be_bytes()
      }
    },
    payload => {
      guard payload.length() == 8 else {
        raise InvalidFrame("INT64 payload must be eight bytes")
      }
      let mut value = 0UL
      if little_endian {
        for shift in 0..<8 {
          value = value | (payload[shift].to_uint64() << (shift * 8))
        }
      } else {
        for byte in payload {
          value = (value << 8) | byte.to_uint64()
        }
      }
      value.reinterpret_as_int64()
    },
  )
}

///|
pub fn SchemaCodec::float() -> SchemaCodec[Float] {
  SchemaCodec::new(SchemaInfo::float(), fn(value) { value.to_be_bytes() }, payload => {
    guard payload.length() == 4 else {
      raise InvalidFrame("FLOAT payload must be four bytes")
    }
    Float::reinterpret_from_uint(read_be32(payload, 0))
  })
}

///|
pub fn SchemaCodec::double() -> SchemaCodec[Double] {
  SchemaCodec::new(
    SchemaInfo::double(),
    fn(value) { value.reinterpret_as_uint64().to_be_bytes() },
    payload => {
      guard payload.length() == 8 else {
        raise InvalidFrame("DOUBLE payload must be eight bytes")
      }
      let mut bits = 0UL
      for byte in payload {
        bits = (bits << 8) | byte.to_uint64()
      }
      bits.reinterpret_as_double()
    },
  )
}

///|
/// JSON payloads for a MoonBit type with JSON conversion support. Supply
/// the broker's JSON schema definition separately from the value encoding.
pub fn[T : ToJson + FromJson] SchemaCodec::json(
  definition : Bytes,
) -> SchemaCodec[T] {
  SchemaCodec::new(
    SchemaInfo::json(definition),
    fn(value) { @utf8.encode(ToJson::to_json(value).stringify()) },
    payload => {
      let text = @utf8.decode(payload) catch {
        _ => raise InvalidFrame("invalid UTF-8 JSON payload")
      }
      let json = @json.parse(text) catch {
        _ => raise InvalidFrame("invalid JSON payload")
      }
      @json.from_json(json) catch {
        _ => raise InvalidFrame("JSON payload does not match the target type")
      }
    },
  )
}

///|
/// Avro binary payloads with typed mappings to and from Avro datums.
/// The definition must be a valid Avro schema. Record fields follow schema
/// order, and malformed payloads are reported as `InvalidFrame`.
pub fn[T] SchemaCodec::avro(
  definition : Bytes,
  to_datum : (T) -> @avro_codec.Datum raise PulsarError,
  from_datum : (@avro_codec.Datum) -> T raise PulsarError,
) -> SchemaCodec[T] raise PulsarError {
  let text = @utf8.decode(definition) catch {
    _ => raise InvalidFrame("invalid UTF-8 Avro schema")
  }
  let schema = @avro_schema.parse(text) catch {
    _ => raise InvalidFrame("invalid Avro schema")
  }
  SchemaCodec::new(
    SchemaInfo::avro(definition),
    value => {
      let datum = to_datum(value)
      @avro_codec.encode(schema, datum) catch {
        _ => raise InvalidFrame("could not encode Avro payload")
      }
    },
    payload => {
      let datum = @avro_codec.decode(schema, payload) catch {
        _ => raise InvalidFrame("invalid Avro payload")
      }
      from_datum(datum)
    },
  )
}

///|
/// Avro binary payloads represented directly as schema-checked datums.
pub fn SchemaCodec::avro_datum(
  definition : Bytes,
) -> SchemaCodec[@avro_codec.Datum] raise PulsarError {
  SchemaCodec::avro(definition, fn(value) { value }, fn(value) { value })
}

///|
/// Use a generated MoonBit protobuf message with the legacy PROTOBUF schema.
/// `definition` is the Avro-style JSON schema definition used by the Go and
/// Java clients for this schema type. Payloads are ordinary protobuf bytes.
pub fn[M : @lib.Read + @lib.Write] SchemaCodec::protobuf(
  definition : Bytes,
) -> SchemaCodec[M] {
  SchemaCodec::new(
    SchemaInfo::protobuf(definition),
    value => {
      let buffer = @buffer.Buffer()
      @lib.Write::write(value, buffer) catch {
        _ => raise InvalidFrame("could not encode Protobuf payload")
      }
      buffer.to_bytes()
    },
    payload => {
      let reader = @lib.BytesReader::from_bytes(payload)
      @lib.Read::read_with_limit(@lib.LimitedReader(reader)) catch {
        _ => raise InvalidFrame("invalid Protobuf payload")
      }
    },
  )
}

///|
/// Use a generated MoonBit protobuf message with a Protobuf Native schema.
/// The descriptor set must describe `M`. Decoder errors are reported as
/// `InvalidFrame`.
pub fn[M : @lib.Read + @lib.Write] SchemaCodec::protobuf_native(
  descriptor_set : Bytes,
  root_file : String,
  root_message : String,
) -> SchemaCodec[M] raise PulsarError {
  let info = SchemaInfo::protobuf_native(
    descriptor_set, root_file, root_message,
  )
  SchemaCodec::new(
    info,
    value => {
      let buffer = @buffer.Buffer()
      @lib.Write::write(value, buffer) catch {
        _ => raise InvalidFrame("could not encode Protobuf Native payload")
      }
      buffer.to_bytes()
    },
    payload => {
      let reader = @lib.BytesReader::from_bytes(payload)
      @lib.Read::read_with_limit(@lib.LimitedReader(reader)) catch {
        _ => raise InvalidFrame("invalid Protobuf Native payload")
      }
    },
  )
}

///|
/// Schema declaration for `Client::create_producer` or
/// `Client::create_consumer`.
pub fn[T] SchemaCodec::info(self : SchemaCodec[T]) -> SchemaInfo {
  self.info
}

///|
/// Encode a value into a message payload.
pub fn[T] SchemaCodec::encode(
  self : SchemaCodec[T],
  value : T,
) -> Bytes raise PulsarError {
  (self.encode_value)(value)
}

///|
/// Decode a payload into a typed value.
pub fn[T] SchemaCodec::decode(
  self : SchemaCodec[T],
  payload : Bytes,
) -> T raise PulsarError {
  (self.decode_value)(payload)
}

///|
/// Decode this message's payload with a schema codec. The original message
/// remains available for metadata and acknowledgement.
pub fn[T] Message::decode(
  self : Message,
  codec : SchemaCodec[T],
) -> T raise PulsarError {
  codec.decode(self.payload())
}