// Copyright 2026 Leo Cheng
// SPDX-License-Identifier: Apache-2.0
// The HTTP/1.1 client ClickHouse's interface needs and nothing more: one POST per
// request, headers in, a body out, either Content-Length or chunked, and the socket
// closed afterwards. It is deliberately not a general-purpose HTTP client — the
// driver has one peer and one verb, and a smaller surface is a smaller thing to get
// wrong.
///|
/// Write `text` into `buf` as UTF-8. A `StringView` slice carries UTF-16 code units,
/// and the buffer's `write_string_utf8` assumes a UTF-8 view, so every string goes
/// through the encoder explicitly.
fn put(buf : Buffer, text : StringView) -> Unit {
buf.write_bytes(@utf8.encode(text))
}
///|
/// A response as the driver needs it: the status, the headers it looks at, and the
/// body.
priv struct Response {
status : Int32
headers : Array[(String, String)]
body : Bytes
}
///|
/// Whether a header names `name` (case-insensitively), returning its value.
fn Response::header(self : Response, name : String) -> String? {
let want = name.to_lower()
for entry in self.headers {
if entry.0.to_lower() == want {
return Some(entry.1)
}
}
None
}
///|
/// One HTTP request on its own connection: `POST` `path` with `headers` and `body`,
/// and the response read to the end. Every failure — resolve, connect, write, a
/// malformed status line, a body that stops early — raises `ConnectError`, because
/// from the caller's point of view the request did not complete.
async fn post(
host : String,
port : Int32,
path : String,
headers : Array[(String, String)],
body : Bytes,
) -> Response raise @moondb.DbError {
let addr = @socket.Addr::resolve(host, port~) catch {
e => raise @moondb.ConnectError("resolve \{host}: \{e}")
}
let tcp = @socket.Tcp::connect(addr) catch {
e => raise @moondb.ConnectError("connect \{host}:\{port}: \{e}")
}
let request = build_request(host, port, path, headers, body)
// A request that does not finish must not leak its socket: MoonBit has no finalizer,
// and the interface opens one connection per statement.
errdefer tcp.close()
(tcp : &@io.Writer).write(request) catch {
e => raise @moondb.ConnectError("write request: \{e}")
}
let response = read_response(tcp)
tcp.close()
response
}
///|
/// The request line, the headers the driver always sends, the caller's headers and
/// the body. `Connection: close` keeps the client honest: one request, one socket, no
/// idle state to leak between statements.
fn build_request(
host : String,
port : Int32,
path : String,
headers : Array[(String, String)],
body : Bytes,
) -> Bytes {
let buf = Buffer()
put(buf, "POST ")
put(buf, path)
put(buf, " HTTP/1.1\r\nHost: ")
put(buf, host)
if port != 80 && port != 443 {
put(buf, ":")
put(buf, port.to_string())
}
put(buf, "\r\nContent-Length: ")
put(buf, body.length().to_string())
put(buf, "\r\nConnection: close\r\n")
for header in headers {
put(buf, header.0)
put(buf, ": ")
put(buf, header.1)
put(buf, "\r\n")
}
put(buf, "\r\n")
buf.write_bytes(body)
buf.to_bytes()
}
///|
/// The status line, the headers, and the body in whichever framing the server chose.
async fn read_response(reader : &@io.Reader) -> Response raise @moondb.DbError {
let status_line = reader.read_until("\r\n") catch {
e => raise @moondb.ConnectError("read status line: \{e}")
}
let status = match status_line {
Some(line) => parse_status(line)
None => 0
}
if status == 0 {
raise @moondb.ConnectError("malformed HTTP status line")
}
let headers : Array[(String, String)] = []
for ;; {
let line = reader.read_until("\r\n") catch {
e => raise @moondb.ConnectError("read header: \{e}")
}
match line {
None => break
Some(l) => {
if l == "" {
break
}
match split_header(l) {
Some(entry) => headers.push(entry)
None => ()
}
}
}
}
let partial = { status, headers, body: b"", }
let body = read_body(reader, partial)
{ status, headers, body, }
}
///|
/// The numeric status out of `HTTP/1.1 200 OK`. Zero means the line was not one.
fn parse_status(line : String) -> Int32 {
// `HTTP/1.x SP status SP reason`; the status is the second token.
match line.find(" ") {
Some(first) => {
let rest = line[first + 1:].to_owned()
let token = match rest.find(" ") {
Some(end) => rest[:end].to_owned()
None => rest
}
match parse_code(token) {
Some(code) => code
None => 0
}
}
None => 0
}
}
///|
/// `Name: value` split at the first colon, with the value trimmed.
fn split_header(line : String) -> (String, String)? {
match line.find(":") {
Some(at) =>
Some((line[:at].to_owned(), line[at + 1:].to_owned().trim().to_owned()))
None => None
}
}
///|
/// The body, framed by `Transfer-Encoding: chunked`, by `Content-Length`, or by the
/// connection closing — in that order of precedence, which is the order the protocol
/// requires.
async fn read_body(
reader : &@io.Reader,
head : Response,
) -> Bytes raise @moondb.DbError {
match head.header("Transfer-Encoding") {
Some(encoding) =>
if encoding.to_lower().contains("chunked") {
return read_chunked(reader)
}
None => ()
}
match head.header("Content-Length") {
Some(text) =>
match parse_code(text.trim().to_owned()) {
Some(len) => {
if len <= 0 {
return b""
}
return reader.read_exactly(len) catch {
e => raise @moondb.ConnectError("read body: \{e}")
}
}
None => ()
}
None => ()
}
// No framing: the body ends when the connection does.
let buf = Buffer()
for ;; {
let chunk = reader.read_some() catch {
e => raise @moondb.ConnectError("read body: \{e}")
}
match chunk {
Some(bytes) => buf.write_bytes(bytes)
None => break
}
}
buf.to_bytes()
}
///|
/// A chunked body: size line, that many bytes, CRLF, until a zero-sized chunk.
async fn read_chunked(reader : &@io.Reader) -> Bytes raise @moondb.DbError {
let buf = Buffer()
for ;; {
let size_line = reader.read_until("\r\n") catch {
e => raise @moondb.ConnectError("read chunk size: \{e}")
}
let size = match size_line {
// A chunk extension follows a `;`; only the size matters here.
Some(line) => parse_hex_size(line)
None => 0
}
if size <= 0 {
// Consume the trailers, which the driver ignores, up to the blank line.
for ;; {
let trailer = reader.read_until("\r\n") catch { _ => None }
match trailer {
Some(t) => if t == "" { break }
None => break
}
}
break
}
let chunk = reader.read_exactly(size) catch {
e => raise @moondb.ConnectError("read chunk: \{e}")
}
buf.write_bytes(chunk)
let _ = reader.read_exactly(2) catch {
e => raise @moondb.ConnectError("read chunk terminator: \{e}")
}
}
buf.to_bytes()
}
///|
/// The hexadecimal size at the head of a chunk-size line.
fn parse_hex_size(line : String) -> Int32 {
let text = match line.find(";") {
Some(at) => line[:at]
None => line[:]
}
let mut value = 0
for i = 0; i < text.length(); i = i + 1 {
let digit = match text[i] {
'0'..<='9' => text[i].to_int() - '0'.to_int()
'a'..<='f' => text[i].to_int() - 'a'.to_int() + 10
'A'..<='F' => text[i].to_int() - 'A'.to_int() + 10
_ => -1
}
if digit < 0 {
return 0
}
value = value * 16 + digit
}
value
}