///|
/// Bolt message codec: the message signatures, typed client-message builders, a
/// typed server-message parser, and the chunked wire framing.
///
/// Every Bolt message is a PackStream struct whose single-byte tag (the struct
/// signature) identifies the message type. Messages are sent in "chunks": each
/// chunk is prefixed with a big-endian 16-bit length, and a `0x0000` chunk
/// terminates the message. See https://neo4j.com/docs/bolt/current/bolt/message/.

// ---------------------------------------------------------------------------
// Message signatures.
// ---------------------------------------------------------------------------

///|
/// Client signature: HELLO.
pub const SIG_HELLO : Int = 0x01

///|
/// Client signature: GOODBYE.
pub const SIG_GOODBYE : Int = 0x02

///|
/// Client signature: RESET.
pub const SIG_RESET : Int = 0x0F

///|
/// Client signature: RUN.
pub const SIG_RUN : Int = 0x10

///|
/// Client signature: BEGIN.
pub const SIG_BEGIN : Int = 0x11

///|
/// Client signature: COMMIT.
pub const SIG_COMMIT : Int = 0x12

///|
/// Client signature: ROLLBACK.
pub const SIG_ROLLBACK : Int = 0x13

///|
/// Client signature: DISCARD.
pub const SIG_DISCARD : Int = 0x2F

///|
/// Client signature: PULL.
pub const SIG_PULL : Int = 0x3F

///|
/// Server signature: SUCCESS.
pub const SIG_SUCCESS : Int = 0x70

///|
/// Server signature: RECORD.
pub const SIG_RECORD : Int = 0x71

///|
/// Server signature: IGNORED.
pub const SIG_IGNORED : Int = 0x7E

///|
/// Server signature: FAILURE.
pub const SIG_FAILURE : Int = 0x7F

// ---------------------------------------------------------------------------
// Client messages.
// ---------------------------------------------------------------------------

///|
/// HELLO: initiate a session with connection metadata (user agent, auth, ...).
pub fn hello(metadata : Array[(String, PackStreamValue)]) -> PackStreamValue {
  PackStreamValue::struct_(SIG_HELLO, [PackStreamValue::map(metadata)])
}

///|
/// GOODBYE: gracefully close the connection.
pub fn goodbye() -> PackStreamValue {
  PackStreamValue::struct_(SIG_GOODBYE, [])
}

///|
/// RESET: return the connection to a clean state, discarding any open transaction.
pub fn reset() -> PackStreamValue {
  PackStreamValue::struct_(SIG_RESET, [])
}

///|
/// RUN: start a query/transaction. `extra` carries mode, bookmarks and
/// transaction metadata.
pub fn run(
  query : String,
  parameters : Array[(String, PackStreamValue)],
  extra : Array[(String, PackStreamValue)],
) -> PackStreamValue {
  PackStreamValue::struct_(SIG_RUN, [
    PackStreamValue::str(query),
    PackStreamValue::map(parameters),
    PackStreamValue::map(extra),
  ])
}

///|
/// BEGIN: start an explicit transaction.
pub fn begin(extra : Array[(String, PackStreamValue)]) -> PackStreamValue {
  PackStreamValue::struct_(SIG_BEGIN, [PackStreamValue::map(extra)])
}

///|
/// COMMIT: commit the open transaction.
pub fn commit() -> PackStreamValue {
  PackStreamValue::struct_(SIG_COMMIT, [])
}

///|
/// ROLLBACK: roll back the open transaction.
pub fn rollback() -> PackStreamValue {
  PackStreamValue::struct_(SIG_ROLLBACK, [])
}

///|
/// DISCARD: discard the remaining records of a result.
pub fn discard(extra : Array[(String, PackStreamValue)]) -> PackStreamValue {
  PackStreamValue::struct_(SIG_DISCARD, [PackStreamValue::map(extra)])
}

///|
/// PULL: request the next records of a result.
pub fn pull(extra : Array[(String, PackStreamValue)]) -> PackStreamValue {
  PackStreamValue::struct_(SIG_PULL, [PackStreamValue::map(extra)])
}

// ---------------------------------------------------------------------------
// Server messages.
// ---------------------------------------------------------------------------

///|
/// A message sent by the server in response to client messages.
pub enum ServerMessage {
  Success(Array[(String, PackStreamValue)])
  Record(Array[PackStreamValue])
  Failure(Array[(String, PackStreamValue)])
  Ignored
} derive(Eq, @debug.Debug)

///|
/// Parse a PackStream struct into a [`ServerMessage`], or `None` when the tag
/// is unknown or the fields do not match the message's shape.
pub fn parse_message(value : PackStreamValue) -> ServerMessage? {
  match value {
    Struct(tag, fields) => parse_server(tag, fields)
    _ => None
  }
}

///|
fn parse_server(tag : Int, fields : Array[PackStreamValue]) -> ServerMessage? {
  if tag == SIG_SUCCESS {
    match fields {
      [Map(m)] => Some(ServerMessage::Success(m))
      _ => None
    }
  } else if tag == SIG_RECORD {
    match fields {
      [List(record)] => Some(ServerMessage::Record(record))
      _ => None
    }
  } else if tag == SIG_FAILURE {
    match fields {
      [Map(m)] => Some(ServerMessage::Failure(m))
      _ => None
    }
  } else if tag == SIG_IGNORED {
    if fields.length() == 0 {
      Some(ServerMessage::Ignored)
    } else {
      None
    }
  } else {
    None
  }
}

// ---------------------------------------------------------------------------
// Chunked framing.
// ---------------------------------------------------------------------------

///|
/// Frame a message struct into Bolt's chunked wire encoding: the PackStream
/// payload split into ≤65535-byte chunks, each prefixed by a big-endian 16-bit
/// length, terminated by a `0x0000` chunk.
pub fn frame(value : PackStreamValue) -> Bytes {
  let payload = packstream_encode(value)
  let buf = Buffer::Buffer()
  let mut i = 0
  let n = payload.length()
  while i < n {
    let chunk = if n - i > 65535 { 65535 } else { n - i }
    write_chunk_len(buf, chunk)
    buf.write_bytes(payload.exact_view(start=i, end=i + chunk))
    i = i + chunk
  }
  write_chunk_len(buf, 0)
  buf.to_bytes()
}

///|
/// Reassemble a chunked message into its PackStream payload bytes. Returns
/// `None` on a truncated chunk or a missing `0x0000` terminator.
pub fn unframe(bytes : Bytes) -> Bytes? {
  let buf = Buffer::Buffer()
  let mut pos = 0
  let n = bytes.length()
  while pos + 2 <= n {
    let chunk = bytes[pos].to_int() * 256 + bytes[pos + 1].to_int()
    pos = pos + 2
    if chunk == 0 {
      return Some(buf.to_bytes())
    }
    if pos + chunk > n {
      return None
    }
    buf.write_bytes(bytes.exact_view(start=pos, end=pos + chunk))
    pos = pos + chunk
  }
  None
}

///|
/// Write a 16-bit big-endian length.
fn write_chunk_len(buf : Buffer, len : Int) -> Unit {
  buf.write_byte((len >> 8).to_byte())
  buf.write_byte((len & 0xFF).to_byte())
}