// The pure server-side HTTP/2 protocol engine that turns a stream of decoded
// frames into a stream of frames to send back — the transport-independent core
// of a gRPC server. It drives the §5.1 stream state machine, the stateful HPACK
// encoder/decoder, and connection- and stream-level flow control (RFC 7540 §6.9),
// routing a completed `application/grpc` request to one of the four call kinds
// (unary / server- / client- / bidi-streaming) and framing each produced message
// as its own length-prefixed DATA, closed by `grpc-status` trailers. No sockets
// and no async here: `feed` is a total function over frames, so the whole engine
// runs in-memory on every backend; the socket driver in `net/` only pumps bytes.

///|
/// The HTTP/2 default flow-control window and initial `SETTINGS_INITIAL_WINDOW_SIZE`
/// (RFC 7540 §6.9.2): 65 535 octets.
pub let default_window_size : Int = 65535

///|
/// The HTTP/2 default (and minimum) `SETTINGS_MAX_FRAME_SIZE` (RFC 7540 §6.5.2).
pub let default_max_frame_size : Int = 16384

///|
/// One server-side stream: its lifecycle state, the accumulating request header
/// block and DATA (with a cursor over the length-prefixed messages already pulled
/// out of it), the per-stream flow-control windows, and the response side — the
/// bytes still to send, whether the initial HEADERS and the trailers have gone
/// out, and any live bidi call state.
pub(all) struct SrvStream {
  id : Int
  mut state : StreamState
  header_block : Buffer
  data : Buffer
  mut req_off : Int
  req_msgs : Array[Bytes]
  mut bidi_processed : Int
  mut headers_complete : Bool
  mut end_stream_recv : Bool
  mut path : String
  mut ctx : RpcContext
  mut recv_window : Int
  mut send_window : Int
  mut out : Bytes
  mut out_off : Int
  mut started : Bool
  mut started_response : Bool
  mut response_ended : Bool
  mut finalized : Bool
  mut trailers_only : Bool
  mut status_code : Int
  mut headers_sent : Bool
  mut trailers_sent : Bool
  mut bidi : BidiHandler?
}

///|
fn SrvStream::new(id : Int, send_window : Int) -> SrvStream {
  {
    id,
    state: Idle,
    header_block: Buffer(),
    data: Buffer(),
    req_off: 0,
    req_msgs: [],
    bidi_processed: 0,
    headers_complete: false,
    end_stream_recv: false,
    path: "",
    ctx: RpcContext::empty(),
    recv_window: default_window_size,
    send_window,
    out: b"",
    out_off: 0,
    started: false,
    started_response: false,
    response_ended: false,
    finalized: false,
    trailers_only: false,
    status_code: 0,
    headers_sent: false,
    trailers_sent: false,
    bidi: None,
  }
}

///|
/// The server side of one HTTP/2 connection: the HPACK codec pair, the live
/// streams, the connection-level flow-control windows, and the peer's settings
/// that bound what we may send. Persistent across the whole connection because
/// HPACK and flow control are stateful.
pub struct H2Server {
  handlers : Map[String, Handler]
  encoder : HpackEncoder
  decoder : HpackDecoder
  streams : Map[Int, SrvStream]
  unary_interceptors : Array[UnaryInterceptor]
  stream_interceptors : Array[StreamInterceptor]
  mut conn_recv_window : Int
  mut conn_send_window : Int
  mut remote_initial_window : Int
  mut remote_max_frame : Int
  mut goaway_received : Bool
}

///|
/// A fresh server engine with no registered handlers. Flow-control windows start
/// at the HTTP/2 defaults until the peer's SETTINGS adjust them.
pub fn H2Server::new() -> H2Server {
  {
    handlers: Map([]),
    encoder: HpackEncoder::new(),
    decoder: HpackDecoder::new(),
    streams: Map([]),
    unary_interceptors: [],
    stream_interceptors: [],
    conn_recv_window: default_window_size,
    conn_send_window: default_window_size,
    remote_initial_window: default_window_size,
    remote_max_frame: default_max_frame_size,
    goaway_received: false,
  }
}

///|
/// Register a unary handler for a fully-qualified gRPC path (`/pkg.Service/Method`):
/// one request message in, one reply message out. An unmatched path gets a
/// trailers-only `grpc-status: 12` (UNIMPLEMENTED) response.
pub fn H2Server::register(
  self : H2Server,
  path : String,
  handler : (Bytes) -> Bytes,
) -> Unit {
  self.handlers[path] = Unary((_ctx, req) => handler(req))
}

///|
/// Register a handler of any of the four gRPC call kinds.
pub fn H2Server::register_handler(
  self : H2Server,
  path : String,
  handler : Handler,
) -> Unit {
  self.handlers[path] = handler
}

///|
/// Register a unary handler that also sees the call context (metadata, deadline,
/// and the response metadata slots).
pub fn H2Server::register_unary(
  self : H2Server,
  path : String,
  handler : (RpcContext, Bytes) -> Bytes,
) -> Unit {
  self.handlers[path] = Unary(handler)
}

///|
/// Register a server-streaming handler: one request message, an ordered sequence
/// of reply messages, each framed as its own gRPC message.
pub fn H2Server::register_server_streaming(
  self : H2Server,
  path : String,
  handler : (RpcContext, Bytes) -> Array[Bytes],
) -> Unit {
  self.handlers[path] = ServerStreaming(handler)
}

///|
/// Register a client-streaming handler: every request message the client sends is
/// collected, and after the client half-closes the handler returns one reply.
pub fn H2Server::register_client_streaming(
  self : H2Server,
  path : String,
  handler : (RpcContext, Array[Bytes]) -> Bytes,
) -> Unit {
  self.handlers[path] = ClientStreaming(handler)
}

///|
/// Register a bidirectional-streaming handler. The factory runs once per call and
/// returns a `BidiHandler` whose `on_message` fires per request message (its
/// replies stream out immediately) and whose `on_end` fires at half-close.
pub fn H2Server::register_bidi(
  self : H2Server,
  path : String,
  factory : (RpcContext) -> BidiHandler,
) -> Unit {
  self.handlers[path] = Bidi(factory)
}

///|
/// Whether the peer has sent GOAWAY; the driver stops accepting new streams once
/// this is set.
pub fn H2Server::goaway_received(self : H2Server) -> Bool {
  self.goaway_received
}

///|
/// The server's opening frames (RFC 7540 §3.5): a SETTINGS frame disabling server
/// push. Sent immediately after the client connection preface is validated, before
/// any request frame is read.
pub fn H2Server::preface(self : H2Server) -> Array[Frame] {
  ignore(self)
  [Settings(params=[(settings_enable_push, 0)], ack=false)]
}

// -- small octet helpers ----------------------------------------------------

///|
/// Latin-1 `Bytes` → `String`, one code unit per octet. HTTP/2 header names/values
/// are byte strings; `:path` and `:method` are ASCII, so this is exact for them.
fn bytes_to_ascii(b : Bytes) -> String {
  let sb = StringBuilder::new()
  for i = 0; i < b.length(); i = i + 1 {
    sb.write_char(b[i].to_int().unsafe_to_char())
  }
  sb.to_string()
}

///|
/// A small non-negative integer as its ASCII decimal `Bytes` (for `grpc-status`).
fn int_to_ascii_bytes(n : Int) -> Bytes {
  let buf = Buffer()
  if n == 0 {
    buf.write_byte(b'0')
  } else {
    let digits : Array[Int] = []
    let mut v = n
    while v > 0 {
      digits.push(v % 10)
      v = v / 10
    }
    for i = digits.length() - 1; i >= 0; i = i - 1 {
      buf.write_byte((digits[i] + 0x30).to_byte())
    }
  }
  buf.to_bytes()
}

///|
/// Concatenate two byte strings.
fn cat(a : Bytes, b : Bytes) -> Bytes {
  let buf = Buffer()
  buf.write_bytes(a)
  buf.write_bytes(b)
  buf.to_bytes()
}

///|
/// The value of the first header named `name` in `headers`, or `None`.
fn header_value(headers : Array[Header], name : Bytes) -> Bytes? {
  for h in headers {
    if h.name == name {
      return Some(h.value)
    }
  }
  None
}

///|
/// Whether a request header is a pseudo-header (`:`-prefixed) or a reserved gRPC/
/// HTTP header, and so is *not* surfaced as call metadata (RFC 7540 §8.1.2.1 plus
/// the gRPC HTTP/2 mapping).
fn is_reserved_header(name : Bytes) -> Bool {
  (name.length() > 0 && name[0] == b':') ||
  name == b"content-type" ||
  name == b"te" ||
  name == b"grpc-timeout" ||
  name == b"grpc-encoding" ||
  name == b"grpc-accept-encoding" ||
  name == b"user-agent"
}

///|
fn min3(a : Int, b : Int, c : Int) -> Int {
  let m = if a < b { a } else { b }
  if m < c {
    m
  } else {
    c
  }
}

///|
/// Pull every complete length-prefixed message now buffered on this stream into
/// `req_msgs`, advancing the read cursor. Partial trailing bytes stay buffered for
/// a later DATA frame to complete.
fn drain_messages(s : SrvStream) -> Unit {
  let all = s.data.to_bytes()
  let n = all.length()
  let mut off = s.req_off
  while n - off >= 5 {
    let len = (all[off + 1].to_int() << 24) |
      (all[off + 2].to_int() << 16) |
      (all[off + 3].to_int() << 8) |
      all[off + 4].to_int()
    if n - off < 5 + len {
      break
    }
    s.req_msgs.push(all[off + 5:off + 5 + len].to_owned())
    off = off + 5 + len
  }
  s.req_off = off
}

// -- the frame-processing core ----------------------------------------------

///|
fn H2Server::stream(self : H2Server, id : Int) -> SrvStream {
  match self.streams.get(id) {
    Some(s) => s
    None => {
      let s = SrvStream::new(id, self.remote_initial_window)
      self.streams[id] = s
      s
    }
  }
}

///|
/// Apply a peer SETTINGS parameter. A change to `INITIAL_WINDOW_SIZE` retroactively
/// shifts every open stream's send window by the delta (RFC 7540 §6.9.2).
fn H2Server::apply_setting(self : H2Server, id : Int, value : Int) -> Unit {
  if id == settings_initial_window_size {
    let delta = value - self.remote_initial_window
    self.remote_initial_window = value
    for _, s in self.streams {
      s.send_window = s.send_window + delta
    }
  } else if id == settings_max_frame_size {
    self.remote_max_frame = value
  } else if id == settings_header_table_size {
    self.encoder.table.set_max_size(value)
  }
}

///|
/// Emit connection- and stream-level WINDOW_UPDATE frames when a receive window
/// has fallen below half the default, replenishing it to the default (a simple,
/// correct auto-tuning policy; RFC 7540 §6.9).
fn H2Server::replenish(self : H2Server, s : SrvStream) -> Array[Frame] {
  let frames : Array[Frame] = []
  let threshold = default_window_size / 2
  if self.conn_recv_window < threshold {
    let inc = default_window_size - self.conn_recv_window
    self.conn_recv_window = self.conn_recv_window + inc
    frames.push(WindowUpdate(stream_id=0, increment=inc))
  }
  if s.recv_window < threshold {
    let inc = default_window_size - s.recv_window
    s.recv_window = s.recv_window + inc
    frames.push(WindowUpdate(stream_id=s.id, increment=inc))
  }
  frames
}

///|
/// Feed one decoded incoming frame to the engine, advancing all state and
/// returning the frames to write back (SETTINGS ack, PING pong, WINDOW_UPDATE,
/// and — as the request stream progresses — the framed gRPC response messages and
/// trailers). Raises on an illegal stream transition or a malformed header block.
pub fn H2Server::feed(self : H2Server, frame : Frame) -> Array[Frame] raise {
  match frame {
    Settings(params~, ack~) =>
      if ack {
        []
      } else {
        for p in params {
          self.apply_setting(p.0, p.1)
        }
        [Settings(params=[], ack=true)]
      }
    Ping(payload~, ack~) => if ack { [] } else { [Ping(payload~, ack=true)] }
    WindowUpdate(stream_id~, increment~) => {
      if stream_id == 0 {
        self.conn_send_window = self.conn_send_window + increment
      } else {
        self.stream(stream_id).send_window += increment
      }
      self.pump_all()
    }
    Headers(stream_id~, fragment~, end_stream~, end_headers~, ..) => {
      let s = self.stream(stream_id)
      s.state = s.state.on_recv(Headers(end_stream~))
      s.header_block.write_bytes(fragment)
      if end_stream {
        s.end_stream_recv = true
      }
      if end_headers {
        self.complete_headers(s)
      }
      self.advance(s)
    }
    Continuation(stream_id~, fragment~, end_headers~) => {
      let s = self.stream(stream_id)
      s.header_block.write_bytes(fragment)
      if end_headers {
        self.complete_headers(s)
      }
      self.advance(s)
    }
    Data(stream_id~, data~, end_stream~, padding~) => {
      let s = self.stream(stream_id)
      s.state = s.state.on_recv(Data(end_stream~))
      let flow_len = data.length() + padding + (if padding > 0 { 1 } else { 0 })
      self.conn_recv_window = self.conn_recv_window - flow_len
      s.recv_window = s.recv_window - flow_len
      s.data.write_bytes(data)
      if end_stream {
        s.end_stream_recv = true
      }
      let frames = self.replenish(s)
      for f in self.advance(s) {
        frames.push(f)
      }
      frames
    }
    RstStream(stream_id~, ..) => {
      match self.streams.get(stream_id) {
        Some(s) => s.state = Closed
        None => ()
      }
      []
    }
    GoAway(..) => {
      self.goaway_received = true
      []
    }
    Priority(..) | PushPromise(..) | Unknown(..) => []
  }
}

///|
/// HPACK-decode this stream's accumulated header block, pull out `:path`, and build
/// the call context (surfaced metadata + parsed `grpc-timeout` deadline).
fn H2Server::complete_headers(self : H2Server, s : SrvStream) -> Unit raise {
  let headers = self.decoder.decode(s.header_block.to_bytes())
  s.path = match header_value(headers, b":path") {
    Some(p) => bytes_to_ascii(p)
    None => ""
  }
  let metadata : Array[Header] = []
  for h in headers {
    if !is_reserved_header(h.name) {
      metadata.push(h)
    }
  }
  let deadline = match header_value(headers, b"grpc-timeout") {
    Some(v) => parse_grpc_timeout(v)
    None => None
  }
  s.ctx = {
    path: s.path,
    metadata,
    deadline_millis: deadline,
    resp_headers: [],
    resp_trailers: [],
  }
  s.headers_complete = true
}

///|
/// Advance a stream after new frames: resolve its handler, route any newly
/// completed request messages to it (streaming replies flow out as they are
/// produced), finalize once the client half-closes, and emit whatever the send
/// windows now permit. Idempotent — safe to call after every frame.
fn H2Server::advance(self : H2Server, s : SrvStream) -> Array[Frame] raise {
  if !s.headers_complete {
    return []
  }
  if !s.started {
    s.started = true
    match self.handlers.get(s.path) {
      Some(Bidi(factory)) => s.bidi = Some(factory(s.ctx))
      Some(_) => ()
      None => {
        s.trailers_only = true
        s.status_code = Status::code(Unimplemented)
        s.response_ended = true
      }
    }
  }
  if !s.trailers_only {
    drain_messages(s)
    match self.handlers.get(s.path) {
      Some(Bidi(_)) =>
        match s.bidi {
          Some(bh) =>
            while s.bidi_processed < s.req_msgs.length() {
              let m = s.req_msgs[s.bidi_processed]
              s.bidi_processed = s.bidi_processed + 1
              for r in (bh.on_message)(m) {
                self.enqueue(s, r)
              }
            }
          None => ()
        }
      _ => ()
    }
    if s.end_stream_recv && !s.finalized {
      s.finalized = true
      self.finalize(s)
    }
  }
  self.produce(s)
}

///|
/// Append a reply message to the stream's outbound buffer, length-prefixed, and
/// mark that a normal response has begun (so its initial HEADERS get sent).
fn H2Server::enqueue(self : H2Server, s : SrvStream, msg : Bytes) -> Unit {
  ignore(self)
  s.out = cat(s.out, encode_message(msg))
  s.started_response = true
}

///|
/// Run the resolved handler at half-close, enqueue its reply message(s), and mark
/// the response body complete so the trailers follow once the body drains.
fn H2Server::finalize(self : H2Server, s : SrvStream) -> Unit {
  match self.handlers.get(s.path) {
    Some(Unary(h)) => {
      let msg = if s.req_msgs.length() > 0 { s.req_msgs[0] } else { b"" }
      let effective = compose_unary(self.unary_interceptors, h)
      self.enqueue(s, effective(s.ctx, msg))
    }
    Some(ServerStreaming(h)) => {
      let msg = if s.req_msgs.length() > 0 { s.req_msgs[0] } else { b"" }
      let effective = compose_stream(self.stream_interceptors, h)
      for r in effective(s.ctx, msg) {
        self.enqueue(s, r)
      }
    }
    Some(ClientStreaming(h)) => self.enqueue(s, h(s.ctx, s.req_msgs))
    Some(Bidi(_)) =>
      match s.bidi {
        Some(bh) =>
          for r in (bh.on_end)() {
            self.enqueue(s, r)
          }
        None => ()
      }
    None => ()
  }
  s.started_response = true
  s.response_ended = true
}

///|
/// Emit the frames a stream can send right now: the response HEADERS once a normal
/// response has begun, as much buffered DATA as the connection and stream send
/// windows and `remote_max_frame` allow, then the trailer HEADERS carrying
/// `grpc-status` once the body is fully drained. The trailers-only UNIMPLEMENTED
/// path is a single END_STREAM HEADERS with no DATA.
fn H2Server::produce(self : H2Server, s : SrvStream) -> Array[Frame] raise {
  let frames : Array[Frame] = []
  if s.trailers_only {
    if !s.trailers_sent {
      let block = self.encoder.encode([
        { name: b":status", value: b"200" },
        { name: b"content-type", value: b"application/grpc" },
        { name: b"grpc-status", value: int_to_ascii_bytes(s.status_code) },
        { name: b"grpc-message", value: b"method not found" },
      ])
      frames.push(
        Headers(
          stream_id=s.id,
          fragment=block,
          end_stream=true,
          end_headers=true,
          priority=None,
          padding=0,
        ),
      )
      s.state = s.state.on_send(Headers(end_stream=true))
      s.trailers_sent = true
    }
    return frames
  }
  if s.started_response && !s.headers_sent {
    let headers : Array[Header] = [
      { name: b":status", value: b"200" },
      { name: b"content-type", value: b"application/grpc" },
      { name: b"grpc-encoding", value: b"identity" },
    ]
    for h in s.ctx.resp_headers {
      headers.push(h)
    }
    let block = self.encoder.encode(headers)
    frames.push(
      Headers(
        stream_id=s.id,
        fragment=block,
        end_stream=false,
        end_headers=true,
        priority=None,
        padding=0,
      ),
    )
    s.state = s.state.on_send(Headers(end_stream=false))
    s.headers_sent = true
  }
  while s.out_off < s.out.length() {
    let remaining = s.out.length() - s.out_off
    let budget = min3(
      self.conn_send_window,
      s.send_window,
      self.remote_max_frame,
    )
    if budget <= 0 {
      break
    }
    let n = if remaining < budget { remaining } else { budget }
    let chunk = s.out[s.out_off:s.out_off + n].to_owned()
    s.out_off = s.out_off + n
    self.conn_send_window = self.conn_send_window - n
    s.send_window = s.send_window - n
    frames.push(Data(stream_id=s.id, data=chunk, end_stream=false, padding=0))
  }
  if s.response_ended &&
    s.out_off >= s.out.length() &&
    s.headers_sent &&
    !s.trailers_sent {
    let trailers : Array[Header] = [
      { name: b"grpc-status", value: int_to_ascii_bytes(s.status_code) },
    ]
    for h in s.ctx.resp_trailers {
      trailers.push(h)
    }
    let block = self.encoder.encode(trailers)
    frames.push(
      Headers(
        stream_id=s.id,
        fragment=block,
        end_stream=true,
        end_headers=true,
        priority=None,
        padding=0,
      ),
    )
    s.state = s.state.on_send(Headers(end_stream=true))
    s.trailers_sent = true
  }
  frames
}

///|
/// Pump every stream that has started but not finished sending (after a
/// connection-level WINDOW_UPDATE lifts back-pressure on all of them at once).
fn H2Server::pump_all(self : H2Server) -> Array[Frame] raise {
  let frames : Array[Frame] = []
  for _, s in self.streams {
    if s.started && !s.trailers_sent {
      for f in self.produce(s) {
        frames.push(f)
      }
    }
  }
  frames
}

///|
/// The lifecycle state of stream `id`, or `Idle` if the engine has never seen it.
pub fn H2Server::stream_state(self : H2Server, id : Int) -> StreamState {
  match self.streams.get(id) {
    Some(s) => s.state
    None => Idle
  }
}