// Copyright 2026 Leo Cheng
// SPDX-License-Identifier: Apache-2.0

///|
/// A live MySQL connection over a raw TCP socket, speaking the wire protocol in
/// pure MoonBit. This is the asynchronous, full-fidelity driver: it owns the
/// socket for its whole lifetime, so `begin`/`commit`/`rollback` bracket a real
/// server-side transaction. The synchronous `@moondb.Driver` adapter
/// ([`MysqlDriver`]) is layered on top for the moondb contract.
///
/// Framing is the classic 3-byte little-endian length + 1-byte sequence id
/// header; the transport transparently reassembles the `0xFFFFFF` continuation
/// frames a large payload is split into.
pub struct MysqlConn {
  stream : @tcp.Conn
  mut seq : Int32
  mut closed : Bool
  server_kind : @moonmysql.ServerKind
  // Set while a `MysqlRowStream` is outstanding — the wire is a per-connection
  // pipeline, so issuing a command before the stream drains would read the old
  // result's leftover packets and desync the connection.
  mut streaming : Bool
  // How this server reads a string literal. Asked for at connect time rather than
  // assumed: under NO_BACKSLASH_ESCAPES a backslash-escaping writer leaves a
  // value's own quote live, which is an injection.
  mut quote_mode : @moonmysql.QuoteMode
}

///|
/// The dialect (MySQL or MariaDB) reported by the server's handshake.
pub fn MysqlConn::server_kind(self : MysqlConn) -> @moonmysql.ServerKind {
  self.server_kind
}

///|
/// Read one logical packet payload from a stream, reassembling
/// `0xFFFFFF`-length continuation frames. Returns the payload and the last
/// sequence id seen.
async fn read_packet_stream(r : &@io.Reader) -> (Bytes, Int32) {
  let out = Buffer()
  let mut seq = 0
  for ;; {
    let hdr = r.read_exactly(4)
    let len = hdr[0].to_int() | (hdr[1].to_int() << 8) | (hdr[2].to_int() << 16)
    seq = hdr[3].to_int()
    if len > 0 {
      out.write_bytes(r.read_exactly(len)[:])
    }
    if len < 0xFFFFFF {
      break
    }
  }
  (out.to_bytes(), seq)
}

///|
/// Read one logical packet payload, reassembling `0xFFFFFF`-length continuation
/// frames. Updates the connection's last-seen sequence id.
async fn MysqlConn::read_packet(self : MysqlConn) -> Bytes {
  let (payload, seq) = read_packet_stream((self.stream : &@io.Reader))
  self.seq = seq
  payload
}

///|
/// Write one packet: the header (payload length + `seq`) then the payload.
///
/// The length field is three bytes, so a payload of 2^24-1 or more goes out as
/// several packets with the sequence id advancing, and a payload that is an exact
/// multiple of the limit needs a trailing empty packet to say it has ended. Without
/// the split the length silently wraps and the server reads the tail of the payload
/// as the next packet header. Reachable in practice: a blob is hex-expanded to twice
/// its size on the way out, so an 8 MB value crosses the boundary.
async fn write_packet_stream(
  w : &@io.Writer,
  seq : Int32,
  payload : Bytes,
) -> Unit {
  let limit = 0xFFFFFF
  if payload.length() >= limit {
    let mut off = 0
    let mut s = seq
    while off < payload.length() {
      let end = if off + limit < payload.length() {
        off + limit
      } else {
        payload.length()
      }
      write_one_packet_stream(w, s, payload[off:end].to_owned())
      s = (s + 1) & 0xFF
      off = end
    }
    if payload.length() % limit == 0 {
      write_one_packet_stream(w, s, b"")
    }
    return
  }
  write_one_packet_stream(w, seq, payload)
}

///|
/// One wire packet, whose payload is already known to fit the three-byte length.
async fn write_one_packet_stream(
  w : &@io.Writer,
  seq : Int32,
  payload : Bytes,
) -> Unit {
  let buf = Buffer()
  let n = payload.length()
  buf.write_byte((n & 0xFF).to_byte())
  buf.write_byte(((n >> 8) & 0xFF).to_byte())
  buf.write_byte(((n >> 16) & 0xFF).to_byte())
  buf.write_byte((seq & 0xFF).to_byte())
  buf.write_bytes(payload[:])
  w.write(buf.to_bytes())
}

///|
async fn MysqlConn::write_packet(
  self : MysqlConn,
  seq : Int32,
  payload : Bytes,
) -> Unit {
  write_packet_stream((self.stream : &@io.Writer), seq, payload)
}

///|
/// Open a connection to `host:port` and complete the auth handshake against
/// `user`/`password`, selecting `database` (empty for none). Works against both
/// MySQL and MariaDB (the reported dialect is on [`server_kind`]).
///
/// mysql_native_password and caching_sha2_password (MySQL 8's default, including
/// the RSA full-auth path on a cache miss) are handled directly, whether the
/// server drives them from its default or via an AuthSwitchRequest. With `tls`,
/// the connection sends an SSLRequest after the greeting and runs the whole
/// handshake over TLS; the caching_sha2 full-auth step then sends the password
/// in cleartext under TLS instead of RSA-encrypting it. A switch to any other
/// plugin (e.g. client_ed25519, on the README roadmap) raises
/// [`@moonmysql.MysqlError::UnsupportedError`].
pub async fn MysqlConn::connect(
  host : String,
  port : Int32,
  user : String,
  password : String,
  database : String,
  tls? : @tls.Client,
  charset~ : Int32 = 45,
) -> MysqlConn {
  let addr = @socket.Addr::resolve(host, port~)
  let tcp = @socket.Tcp::connect(addr)
  // Phase 1 reads the greeting over the raw socket: the SSLRequest that decides
  // whether the rest runs over TLS can only be sent once the greeting is seen.
  let (greeting, greeting_seq) = read_packet_stream((tcp : &@io.Reader))
  let handshake = @moonmysql.parse_handshake(greeting)
  let use_tls = tls is Some(_)
  if use_tls && (handshake.capability & @moonmysql.client_ssl) == 0 {
    tcp.close()
    raise @moonmysql.UnsupportedError("server does not offer CLIENT_SSL")
  }
  let stream = if use_tls {
    write_packet_stream(
      (tcp : &@io.Writer),
      greeting_seq + 1,
      @moonmysql.build_ssl_request(database, charset~),
    )
    let client = match tls {
      Some(c) => c
      None => abort("tls is Some")
    }
    @tcp.Conn::client(tcp, client, host) catch {
      e => {
        tcp.close()
        raise @moonmysql.ProtocolError(
          "TLS handshake with " + host + ": " + e.to_string(),
        )
      }
    }
  } else {
    @tcp.Conn::plain(tcp)
  }
  let conn = MysqlConn::{
    stream,
    seq: if use_tls {
      greeting_seq + 1
    } else {
      greeting_seq
    },
    closed: false,
    server_kind: handshake.server_kind,
    streaming: false,
    quote_mode: @moonmysql.QuoteMode::Backslash,
  }
  let pw = @utf8.encode(password[:])
  // Close the socket on any handshake failure — a bad-credentials ERR is the
  // common case, and MoonBit has no fd finalizer to reclaim a leaked socket.
  errdefer conn.stream.close()
  conn.send_handshake_response(
    handshake,
    user,
    pw,
    database,
    ssl=use_tls,
    charset~,
  )
  conn.finish_auth(handshake, pw, use_tls)
  conn.learn_quoting()
  conn
}

///|
/// Ask the server how it reads string literals and what the connection charset is,
/// and refuse the connection when the pair is one this driver cannot escape safely.
///
/// The text protocol renders parameters as literals, so the escaping has to match
/// the server's actual `sql_mode` — assuming backslashes are escapes when they are
/// not is an injection, not a formatting difference. And in a charset where a
/// multibyte character can end in `0x5C`, backslash escaping is unsafe no matter
/// what: the quote that follows becomes the tail of a legitimate character. That
/// combination is refused rather than served insecurely.
async fn MysqlConn::learn_quoting(self : MysqlConn) -> Unit {
  let rows = self.query("SELECT @@sql_mode, @@character_set_connection", [])
  guard rows.length() > 0 else { return }
  let mode_text = match rows[0].values[0] {
    Text(t) => t
    _ => ""
  }
  let charset = match rows[0].values[1] {
    Text(t) => t
    _ => ""
  }
  if mode_text.contains("NO_BACKSLASH_ESCAPES") {
    self.quote_mode = @moonmysql.QuoteMode::DoubleQuote
    return
  }
  // The charsets whose trailing bytes can be 0x5C. Doubling a quote is safe in all
  // of them; escaping with a backslash is not.
  for unsafe_cs in ["gbk", "big5", "sjis", "cp932", "gb18030"] {
    if charset == unsafe_cs {
      raise @moonmysql.ProtocolError(
        "connection charset '" +
        charset +
        "' cannot be escaped safely without NO_BACKSLASH_ESCAPES or prepared statements",
      )
    }
  }
}

///|
/// Send the HandshakeResponse41 for `user`/`password`/`database`, carrying
/// `CLIENT_SSL` when the connection runs over TLS.
async fn MysqlConn::send_handshake_response(
  self : MysqlConn,
  handshake : @moonmysql.Handshake,
  user : String,
  pw : Bytes,
  database : String,
  ssl~ : Bool,
  charset~ : Int32,
) -> Unit {
  let response = @moonmysql.build_handshake_response(
    handshake,
    user,
    pw,
    database,
    ssl~,
    charset~,
  )
  // The handshake response follows the greeting (and the SSLRequest, when TLS
  // was negotiated) on the sequence number.
  self.write_packet(self.seq + 1, response)
}

///|
/// The authentication exchange after the handshake response, run under
/// [`connect`]'s close-on-error guard.
async fn MysqlConn::finish_auth(
  self : MysqlConn,
  handshake : @moonmysql.Handshake,
  pw : Bytes,
  tls : Bool,
) -> Unit {
  let auth = self.read_packet()
  if @moonmysql.is_err_packet(auth) {
    raise @moonmysql.parse_err(auth)
  }
  if @moonmysql.is_ok_packet(auth) {
    return
  }
  // AuthSwitchRequest (0xFE + body): the server offers a different plugin — answer
  // mysql_native_password directly, or caching_sha2_password / client_ed25519 via
  // their scramble + AuthMoreData exchange.
  if @moonmysql.is_auth_switch_request(auth) {
    let (plugin, new_salt) = @moonmysql.parse_auth_switch_request(auth)
    if plugin == "mysql_native_password" || plugin == "" {
      self.write_packet(
        self.seq + 1,
        @moonmysql.native_password_scramble(pw, new_salt),
      )
      expect_ok_or_err(self.read_packet())
      return
    }
    if plugin == "caching_sha2_password" {
      self.write_packet(
        self.seq + 1,
        @moonmysql.caching_sha2_scramble(pw, new_salt),
      )
      self.finish_caching_sha2(pw, new_salt, self.read_packet(), tls)
      return
    }
    if plugin == "client_ed25519" {
      // MariaDB signs the raw 32-byte nonce (not the NUL-stripped salt) with the
      // password-keyed Ed25519 secret.
      let nonce = @moonmysql.parse_ed25519_challenge(auth)
      self.write_packet(
        self.seq + 1,
        @moonmysql.mariadb_ed25519_response(pw, nonce),
      )
      expect_ok_or_err(self.read_packet())
      return
    }
    raise @moonmysql.UnsupportedError(
      "server requested auth-switch to '" +
      plugin +
      "'; only mysql_native_password, caching_sha2_password and client_ed25519 are implemented",
    )
  }
  // AuthMoreData (lead byte 0x01): the caching_sha2 reply to the fast scramble we
  // already sent in the handshake response.
  if auth.length() >= 1 && auth[0].to_int() == 0x01 {
    self.finish_caching_sha2(pw, handshake.salt, auth, tls)
    return
  }
  raise @moonmysql.UnsupportedError("unexpected packet during authentication")
}

///|
/// Drive the caching_sha2_password exchange to completion given the server's
/// AuthMoreData reply to the fast scramble. `0x03` is fast-auth success (a cached
/// digest); `0x04` needs full auth — without TLS the client fetches the server's
/// RSA public key and returns the nonce-obfuscated password RSA-OAEP encrypted
/// under it. `nonce` is the same salt the fast scramble used.
async fn MysqlConn::finish_caching_sha2(
  self : MysqlConn,
  pw : Bytes,
  nonce : Bytes,
  more_data : Bytes,
  tls : Bool,
) -> Unit {
  guard more_data.length() >= 2 && more_data[0].to_int() == 0x01 else {
    raise @moonmysql.ProtocolError("caching_sha2: expected AuthMoreData")
  }
  match more_data[1].to_int() {
    0x03 => expect_ok_or_err(self.read_packet()) // fast auth success
    0x04 =>
      if tls {
        // Over TLS the password is sent in cleartext (NUL-terminated) rather than
        // RSA-encrypted: the channel already protects it, and this is the path
        // MySQL documents for secure connections.
        let pwz = Buffer()
        pwz.write_bytes(pw[:])
        pwz.write_byte(b'\x00')
        self.write_packet(self.seq + 1, pwz.to_bytes())
        expect_ok_or_err(self.read_packet())
      } else {
        // Request the server's RSA public key (no TLS ⇒ full auth over RSA).
        self.write_packet(self.seq + 1, b"\x02")
        let key_pkt = self.read_packet()
        guard key_pkt.length() >= 1 && key_pkt[0].to_int() == 0x01 else {
          raise @moonmysql.ProtocolError("caching_sha2: expected public key")
        }
        let pem = @moonmysql.bytes_to_string(key_pkt[1:].to_owned())
        let seed = match @env.rand(20) {
          Some(b) => b
          None => raise @moonmysql.ProtocolError("no CSPRNG for the OAEP seed")
        }
        let token = @moonmysql.caching_sha2_full_auth_token(
          pw, nonce, pem, seed,
        )
        self.write_packet(self.seq + 1, token)
        expect_ok_or_err(self.read_packet())
      }
    other =>
      raise @moonmysql.ProtocolError(
        "caching_sha2: unexpected status " + other.to_string(),
      )
  }
}

///|
/// A terminal auth packet must be OK (success) or ERR (raise); anything else is a
/// protocol violation.
fn expect_ok_or_err(pkt : Bytes) -> Unit raise @moonmysql.MysqlError {
  if @moonmysql.is_err_packet(pkt) {
    raise @moonmysql.parse_err(pkt)
  }
  if @moonmysql.is_ok_packet(pkt) {
    return
  }
  raise @moonmysql.ProtocolError("unexpected packet after auth")
}

///|
/// The outcome of one command: either an OK (no result set) or a materialised
/// text result set (column definitions plus raw row packets).
priv enum Outcome {
  Exec(@moonmysql.OkPacket)
  Rows(Array[@moonmysql.ColumnDef], Array[Bytes])
}

///|
/// Send a `COM_QUERY` carrying `query_bytes` and read **every** result set it
/// produces. One statement can answer with more than one — a stored procedure's
/// `CALL`, a multi-statement packet — and the wire has no way to say so up front:
/// the status flag on each terminator announces the next one. Reading only the first
/// and leaving the rest on the socket desynchronises the connection, so the whole
/// chain is consumed here and the caller picks what it needs.
async fn MysqlConn::command_all(
  self : MysqlConn,
  query_bytes : Bytes,
) -> Array[Outcome] {
  if self.closed {
    raise @moondb.Closed
  }
  if self.streaming {
    raise @moonmysql.ProtocolError(
      "a query stream is still open on this connection; drain it or call close() before the next command",
    )
  }
  let payload = Buffer()
  payload.write_byte(b'\x03') // COM_QUERY
  payload.write_bytes(query_bytes[:])
  // A fresh command always starts at sequence id 0.
  self.write_packet(0, payload.to_bytes())
  let outcomes : Array[Outcome] = []
  let mut more = true
  while more {
    let first = self.read_packet()
    if @moonmysql.is_err_packet(first) {
      raise @moonmysql.parse_err(first)
    }
    if @moonmysql.is_ok_packet(first) {
      let ok = @moonmysql.parse_ok(first)
      more = (ok.status & @moonmysql.server_more_results_exists) != 0
      outcomes.push(Exec(ok))
      continue
    }
    if first.length() > 0 && first[0].to_int() == 0xFB {
      raise @moonmysql.UnsupportedError("LOCAL INFILE is not supported")
    }
    let reader = @moonmysql.PacketReader::new(first)
    let column_count = reader.lenenc_uint().to_int()
    let columns : Array[@moonmysql.ColumnDef] = []
    for _ in 0.. Array[Array[@moondb.Row]] {
  if params.length() > 0 {
    raise @moonmysql.UnsupportedError(
      "multiple result sets are only read on the text protocol; pass no parameters",
    )
  }
  let outcomes = self.command_all(
    @moonmysql.bind_params(sql, params, mode=self.quote_mode),
  )
  let sets : Array[Array[@moondb.Row]] = []
  for outcome in outcomes {
    match outcome {
      Exec(_) => sets.push([])
      Rows(columns, row_payloads) =>
        sets.push(@moonmysql.build_text_rows(columns, row_payloads))
    }
  }
  sets
}

///|
/// Send a `COM_QUERY` and return its **first** result, draining any further ones so
/// the connection stays usable. The single-result path every `execute`/`query` takes.
async fn MysqlConn::command(self : MysqlConn, query_bytes : Bytes) -> Outcome {
  let outcomes = self.command_all(query_bytes)
  match outcomes[0] {
    Exec(ok) => Exec(ok)
    Rows(columns, rows) => Rows(columns, rows)
  }
}

///|
/// COM_STMT_PREPARE: allocate a server-side statement for `sql`, consuming the
/// parameter and column definitions up to their EOF packets.
async fn MysqlConn::compile(
  self : MysqlConn,
  sql : String,
) -> @moonmysql.StmtOk {
  if self.closed {
    raise @moondb.Closed
  }
  if self.streaming {
    raise @moonmysql.ProtocolError(
      "a query stream is still open on this connection",
    )
  }
  self.write_packet(0, @moonmysql.build_stmt_prepare(sql))
  let first = self.read_packet()
  if @moonmysql.is_err_packet(first) {
    raise @moonmysql.parse_err(first)
  }
  let ok = @moonmysql.parse_stmt_prepare_ok(first)
  for _ in 0.. ignore
  }
  if ok.num_params > 0 {
    let eof = self.read_packet()
    guard @moonmysql.is_eof_packet(eof) else {
      raise @moonmysql.ProtocolError("expected EOF after statement parameters")
    }
  }
  for _ in 0.. ignore
  }
  if ok.num_columns > 0 {
    let eof = self.read_packet()
    guard @moonmysql.is_eof_packet(eof) else {
      raise @moonmysql.ProtocolError("expected EOF after statement columns")
    }
  }
  ok
}

///|
async fn MysqlConn::stmt_close(self : MysqlConn, id : Int32) -> Unit {
  self.write_packet(0, @moonmysql.build_stmt_close(id))
}

///|
/// Bind `params` to the prepared statement `id` and execute it.
///
/// Parameters too large to inline go first, streamed with
/// `COM_STMT_SEND_LONG_DATA` one chunk at a time; the execute packet that follows
/// binds them without repeating their bytes. The server sends no answer to the long
/// data, so a streamed statement costs the same one round trip as an inline one.
async fn MysqlConn::send_execute(
  self : MysqlConn,
  id : Int32,
  params : Array[@moondb.Value],
) -> Unit {
  let streamed = @moonmysql.long_data_params(params)
  for index in streamed {
    for slice in @moonmysql.long_data_slices(params, index) {
      self.write_packet(
        0,
        @moonmysql.build_stmt_send_long_data(id, index, slice),
      )
    }
  }
  self.write_packet(0, @moonmysql.build_stmt_execute(id, params, streamed~))
}

///|
/// Run a prepared statement and materialise its rows from the binary protocol.
async fn MysqlConn::read_prepared_rows(
  self : MysqlConn,
  id : Int32,
  params : Array[@moondb.Value],
) -> Array[@moondb.Row] {
  self.send_execute(id, params)
  let first = self.read_packet()
  if @moonmysql.is_err_packet(first) {
    raise @moonmysql.parse_err(first)
  }
  if @moonmysql.is_ok_packet(first) {
    return []
  }
  let reader = @moonmysql.PacketReader::new(first)
  let count = reader.lenenc_uint().to_int()
  let columns : Array[@moonmysql.ColumnDef] = []
  for _ in 0.. Array[@moondb.Row] {
  let stmt = self.compile(sql)
  errdefer @async.protect_from_cancel(() => self.stmt_close(stmt.id))
  let rows = self.read_prepared_rows(stmt.id, params)
  self.stmt_close(stmt.id)
  rows
}

///|
async fn MysqlConn::prepared_execute(
  self : MysqlConn,
  sql : String,
  params : Array[@moondb.Value],
) -> @moondb.ExecResult {
  let stmt = self.compile(sql)
  errdefer @async.protect_from_cancel(() => self.stmt_close(stmt.id))
  self.send_execute(stmt.id, params)
  let first = self.read_packet()
  if @moonmysql.is_err_packet(first) {
    raise @moonmysql.parse_err(first)
  }
  if !@moonmysql.is_ok_packet(first) {
    raise @moonmysql.ProtocolError(
      "execute() was given a row-returning statement; use query()",
    )
  }
  let ok = @moonmysql.parse_ok(first)
  let result = @moondb.ExecResult::{
    rows_affected: ok.affected_rows,
    last_insert_id: ok.last_insert_id,
  }
  self.stmt_close(stmt.id)
  result
}

///|
/// Run a non-row statement and report the rows affected and last insert id. With
/// bound `params` it uses a prepared statement (binary protocol, out-of-band
/// binding); without, the text protocol.
pub async fn MysqlConn::execute(
  self : MysqlConn,
  sql : String,
  params : Array[@moondb.Value],
) -> @moondb.ExecResult {
  if params.length() > 0 {
    return self.prepared_execute(sql, params)
  }
  match
    self.command(@moonmysql.bind_params(sql, params, mode=self.quote_mode)) {
    Exec(ok) =>
      @moondb.ExecResult::{
        rows_affected: ok.affected_rows,
        last_insert_id: ok.last_insert_id,
      }
    Rows(_, _) =>
      raise @moonmysql.ProtocolError(
        "execute() was given a row-returning statement; use query()",
      )
  }
}

///|
/// Run a row-returning statement and materialise every [`@moondb.Row`]. With
/// bound `params` it uses a prepared statement (binary protocol); without, the
/// text protocol.
pub async fn MysqlConn::query(
  self : MysqlConn,
  sql : String,
  params : Array[@moondb.Value],
) -> Array[@moondb.Row] {
  if params.length() > 0 {
    return self.prepared_query(sql, params)
  }
  match
    self.command(@moonmysql.bind_params(sql, params, mode=self.quote_mode)) {
    Rows(columns, row_payloads) =>
      @moonmysql.build_text_rows(columns, row_payloads)
    Exec(_) => []
  }
}

///|
/// A forward-only cursor that reads a query's rows off the wire on demand instead
/// of buffering them — the streaming counterpart to [`query`], for results too
/// large to materialise at once (`mysql_use_result`, SQLAlchemy's
/// `stream_results`). Bound to its `MysqlConn`, a single-command pipeline: the
/// stream must be drained (`next` returns `None`) or [`close`d](MysqlRowStream::close)
/// before another statement runs on that connection.
pub struct MysqlRowStream {
  conn : MysqlConn
  columns : Array[@moonmysql.ColumnDef]
  mut pending : Bytes?
  mut done : Bool
}

///|
/// Send a row-returning statement and return a streaming cursor over its rows. The
/// column definitions and their terminating EOF are consumed up front; the rows
/// themselves are read one at a time by [`next`](MysqlRowStream::next).
pub async fn MysqlConn::query_stream(
  self : MysqlConn,
  sql : String,
  params : Array[@moondb.Value],
) -> MysqlRowStream {
  if self.closed {
    raise @moondb.Closed
  }
  if self.streaming {
    raise @moonmysql.ProtocolError(
      "a query stream is still open on this connection; drain it or call close() first",
    )
  }
  let payload = Buffer()
  payload.write_byte(b'\x03') // COM_QUERY
  payload.write_bytes(
    @moonmysql.bind_params(sql, params, mode=self.quote_mode)[:],
  )
  self.write_packet(0, payload.to_bytes())
  let first = self.read_packet()
  if @moonmysql.is_err_packet(first) {
    raise @moonmysql.parse_err(first)
  }
  // OK ⇒ no result set: an already-exhausted stream.
  if @moonmysql.is_ok_packet(first) {
    return { conn: self, columns: [], pending: None, done: true, }
  }
  if first.length() > 0 && first[0].to_int() == 0xFB {
    raise @moonmysql.UnsupportedError("LOCAL INFILE is not supported")
  }
  let reader = @moonmysql.PacketReader::new(first)
  let column_count = reader.lenenc_uint().to_int()
  let columns : Array[@moonmysql.ColumnDef] = []
  for _ in 0.. @moondb.Row? {
  if self.done {
    return None
  }
  let packet = match self.pending {
    Some(p) => {
      self.pending = None
      p
    }
    None => self.conn.read_packet()
  }
  if @moonmysql.is_err_packet(packet) {
    self.finish()
    raise @moonmysql.parse_err(packet)
  }
  if @moonmysql.is_eof_packet(packet) {
    self.finish()
    if @moonmysql.has_more_results(packet) {
      raise @moonmysql.UnsupportedError(
        "the server returned more than one result set; run one statement per query",
      )
    }
    return None
  }
  Some(@moonmysql.build_text_rows(self.columns, [packet])[0])
}

///|
/// Mark the stream done and release the connection's streaming guard.
fn MysqlRowStream::finish(self : MysqlRowStream) -> Unit {
  self.done = true
  self.conn.streaming = false
}

///|
/// Abandon the stream early, draining any unread rows to the terminating EOF so the
/// connection can run the next statement. Idempotent once the stream is done.
pub async fn MysqlRowStream::close(self : MysqlRowStream) -> Unit {
  if self.done {
    return
  }
  self.pending = None
  for ;; {
    let p = self.conn.read_packet()
    if @moonmysql.is_eof_packet(p) || @moonmysql.is_err_packet(p) {
      break
    }
  }
  self.finish()
}

///|
/// Begin a server-side transaction.
pub async fn MysqlConn::begin(self : MysqlConn) -> Unit {
  let _ = self.command(b"START TRANSACTION")
}

///|
/// Commit the current transaction.
pub async fn MysqlConn::commit(self : MysqlConn) -> Unit {
  let _ = self.command(b"COMMIT")
}

///|
/// Roll back the current transaction.
pub async fn MysqlConn::rollback(self : MysqlConn) -> Unit {
  let _ = self.command(b"ROLLBACK")
}

///|
/// Close the socket (best-effort, idempotent).
pub fn MysqlConn::close(self : MysqlConn) -> Unit {
  if !self.closed {
    self.closed = true
    self.stream.close()
  }
}