rpc.mbt

///|
/// The default cap on a single received gRPC message (4 MiB, matching gRPC's default
/// `MaxRecvMsgSize`). A length prefix above this — including one whose 4 bytes decode
/// to a negative `Int` because the high bit is set — is rejected rather than trusted,
/// so a hostile prefix can neither slice out of bounds nor pin unbounded buffer.
pub let max_message_size : Int = 4 * 1024 * 1024

///|
/// The cap on one accumulated header block (HEADERS plus its CONTINUATION frames).
/// Without it a peer could stream endless non-final CONTINUATION frames and grow the
/// buffer without bound (a CONTINUATION flood); 128 KiB is far above any real gRPC
/// request's headers.
pub let max_header_list_size : Int = 128 * 1024

///|
/// Encode a payload as a gRPC *Length-Prefixed-Message*: a 1-byte compression
/// flag, a 4-byte big-endian length, then the payload. This is the framing every
/// gRPC transport shares (gRPC-Web over HTTP/1.1 and real gRPC over HTTP/2 alike).
pub fn encode_message(payload : Bytes, compressed? : Bool = false) -> Bytes {
  let n = payload.length()
  let buf = Buffer()
  buf.write_byte((if compressed { 1 } else { 0 }).to_byte())
  @fixed.write_u32(buf, n.reinterpret_as_uint())
  buf.write_bytes(payload)
  buf.to_bytes()
}

///|
/// Decode one gRPC length-prefixed message from the front of `data`, returning
/// `(compressed, payload)`, or `None` if fewer than a full frame is present.
pub fn decode_message(data : Bytes) -> (Bool, Bytes)? {
  guard message_len(data[:]) is Some(len) else { return None }
  // A high-bit-set prefix decodes to a negative `Int`; compare without recomputing
  // `5 + len` (which would wrap) so a hostile length can't slice past the buffer.
  if len < 0 || len > data.length() - 5 {
    return None
  }
  Some((data[0].to_int() != 0, data[5:5 + len].to_owned()))
}

///|
/// The payload length of the length-prefixed message at `at`, or `None` when fewer
/// than the five octets of a prefix are there.
///
/// A prefix with its high bit set reads back negative, which is how a hostile length
/// is told from a real one before anything is sliced on it. Every reader of this
/// framing goes through here — there are five of them across the client, the server
/// and the socket transport, and they have to agree.
pub fn message_len(data : BytesView, at? : Int = 0) -> Int? {
  @fixed.read_u32(data, at=at + 1).map(v => v.reinterpret_as_int())
}

///|
/// Split a run of concatenated length-prefixed messages into their payloads, stopping
/// at the first one that has not fully arrived.
pub fn split_messages(data : Bytes) -> Array[Bytes] {
  let out : Array[Bytes] = []
  let mut off = 0
  while message_len(data[:], at=off) is Some(len) {
    if len < 0 || data.length() - off < 5 + len {
      break
    }
    out.push(data[off + 5:off + 5 + len].to_owned())
    off = off + 5 + len
  }
  out
}

///|
/// The 17 canonical gRPC status codes (`grpc-status`).
pub(all) enum Status {
  Ok
  Cancelled
  Unknown
  InvalidArgument
  DeadlineExceeded
  NotFound
  AlreadyExists
  PermissionDenied
  ResourceExhausted
  FailedPrecondition
  Aborted
  OutOfRange
  Unimplemented
  Internal
  Unavailable
  DataLoss
  Unauthenticated
} derive(Eq)

///|
/// The numeric `grpc-status` code.
pub fn Status::code(self : Status) -> Int {
  match self {
    Ok => 0
    Cancelled => 1
    Unknown => 2
    InvalidArgument => 3
    DeadlineExceeded => 4
    NotFound => 5
    AlreadyExists => 6
    PermissionDenied => 7
    ResourceExhausted => 8
    FailedPrecondition => 9
    Aborted => 10
    OutOfRange => 11
    Unimplemented => 12
    Internal => 13
    Unavailable => 14
    DataLoss => 15
    Unauthenticated => 16
  }
}

///|
/// The canonical uppercase status name.
pub fn Status::name(self : Status) -> String {
  match self {
    Ok => "OK"
    Cancelled => "CANCELLED"
    Unknown => "UNKNOWN"
    InvalidArgument => "INVALID_ARGUMENT"
    DeadlineExceeded => "DEADLINE_EXCEEDED"
    NotFound => "NOT_FOUND"
    AlreadyExists => "ALREADY_EXISTS"
    PermissionDenied => "PERMISSION_DENIED"
    ResourceExhausted => "RESOURCE_EXHAUSTED"
    FailedPrecondition => "FAILED_PRECONDITION"
    Aborted => "ABORTED"
    OutOfRange => "OUT_OF_RANGE"
    Unimplemented => "UNIMPLEMENTED"
    Internal => "INTERNAL"
    Unavailable => "UNAVAILABLE"
    DataLoss => "DATA_LOSS"
    Unauthenticated => "UNAUTHENTICATED"
  }
}

///|
/// The status a call ends with when the peer resets its stream instead of sending
/// trailers, from the RST_STREAM error code (gRPC PROTOCOL-HTTP2, "HTTP2 Error Code →
/// Status"). A refused stream was never processed, so it is UNAVAILABLE and safe to
/// retry; anything the table does not name is INTERNAL, since the peer broke the
/// transport rather than the call.
pub fn Status::from_h2_error(code : Int) -> Status {
  if code == @http2.error_refused_stream {
    Unavailable
  } else if code == @http2.error_enhance_your_calm {
    ResourceExhausted
  } else if code == @http2.error_inadequate_security {
    PermissionDenied
  } else if code == @http2.error_cancel {
    Cancelled
  } else {
    Internal
  }
}

///|
/// The status a non-200 HTTP response maps to when it carries no `grpc-status` — a
/// proxy or a plain HTTP server answering on the gRPC port (gRPC
/// http-grpc-status-mapping). Anything outside the table is UNKNOWN.
pub fn Status::from_http(code : Int) -> Status {
  match code {
    400 => Internal
    401 => Unauthenticated
    403 => PermissionDenied
    404 => Unimplemented
    429 | 502 | 503 | 504 => Unavailable
    _ => Unknown
  }
}

///|
/// A fully-qualified RPC method: `package.Service` and the method name.
pub(all) struct Method {
  service : String
  name : String
}

///|
/// The gRPC HTTP/2 `:path`, i.e. `/package.Service/Method`.
pub fn Method::path(self : Method) -> String {
  "/" + self.service + "/" + self.name
}

///|
pub extend Status with Eq::{not_equal, equal}