///|
type WsTextSender = (String) -> Unit

///|
type WsBinarySender = (Bytes) -> Unit

///|
type WsPongSender = () -> Unit

///|
priv enum OutboundFrame {
  OutboundText(String)
  OutboundBinary(Bytes)
  OutboundPong
}

///|
priv struct WebSocketOutboundQueue {
  mut frames : Array[OutboundFrame]
  mut draining : Bool
  mut closed : Bool
}

///|
fn WebSocketOutboundQueue::new() -> WebSocketOutboundQueue {
  { frames: [], draining: false, closed: false, }
}

///|
let ws_text_senders : Map[String, WsTextSender] = Map([])

///|
let ws_binary_senders : Map[String, WsBinarySender] = Map([])

///|
let ws_pong_senders : Map[String, WsPongSender] = Map([])

///|
let ws_client_channels : Map[String, Array[String]] = Map([])

///|
let ws_channel_clients : Map[String, Map[String, Unit]] = Map([])

///|
let native_ws_handler_map : Map[Int, App] = Map([])

///|
fn request_method_to_string(meth : @http.RequestMethod) -> String {
  match meth {
    Get => "GET"
    Head => "HEAD"
    Post => "POST"
    Put => "PUT"
    Delete => "DELETE"
    Connect => "CONNECT"
    Options => "OPTIONS"
    Trace => "TRACE"
    Patch => "PATCH"
  }
}

///|
fn string_headers_to_views(
  headers : Map[@http.CaseInsensitiveString, String],
) -> Map[@http.CaseInsensitiveString, StringView] {
  let out : Map[@http.CaseInsensitiveString, StringView] = Map([])
  headers.each((key, value) => out.set(key, value))
  out
}

///|
fn view_headers_to_strings(
  headers : Map[@http.CaseInsensitiveString, StringView],
) -> Map[@http.CaseInsensitiveString, String] {
  let out : Map[@http.CaseInsensitiveString, String] = Map([])
  headers.each((key, value) => out.set(key, value.to_owned()))
  out
}

///|
fn header_contains_token(
  headers : Map[@http.CaseInsensitiveString, String],
  header_name : String,
  token : String,
) -> Bool {
  let token = token.to_lower()
  let header_name = header_name.to_lower()
  for pair in headers {
    let (key, value) = pair
    if Show::to_string(key).to_lower() == header_name {
      for part in value.split(",") {
        if part.trim().to_lower() == token {
          return true
        }
      }
    }
  }
  false
}

///|
fn header_equals(
  headers : Map[@http.CaseInsensitiveString, String],
  header_name : String,
  expected : String,
) -> Bool {
  let header_name = header_name.to_lower()
  let expected = expected.to_lower()
  for pair in headers {
    let (key, value) = pair
    if Show::to_string(key).to_lower() == header_name {
      return value.trim().to_lower() == expected
    }
  }
  false
}

///|
fn is_websocket_upgrade(request : @http.Request) -> Bool {
  header_contains_token(request.headers, "connection", "upgrade") &&
  header_equals(request.headers, "upgrade", "websocket")
}

///|
fn find_ws_route(
  mocket : App,
  path : String,
) -> (WebSocketHandler, Map[String, StringView])? {
  match mocket.ws_static_routes.get(path) {
    Some(handler) => return Some((handler, {}))
    None => ()
  }
  for route in mocket.ws_dynamic_routes {
    let (route_path, handler) = route
    match match_path(route_path, path) {
      Some(params) => return Some((handler, params))
      None => ()
    }
  }
  None
}

///|
fn next_ws_connection_id(port : Int) -> String {
  "native-\{port}-\{@env.now()}"
}

///|
pub fn register_ws_connection(
  connection_id : String,
  text_sender : WsTextSender,
  binary_sender : WsBinarySender,
  pong_sender : WsPongSender,
) -> Unit {
  ws_text_senders.set(connection_id, text_sender)
  ws_binary_senders.set(connection_id, binary_sender)
  ws_pong_senders.set(connection_id, pong_sender)
  ws_client_channels.set(connection_id, [])
}

///|
pub fn unregister_ws_connection(connection_id : String) -> Unit {
  let channels = ws_client_channels.get(connection_id)
  ignore(ws_text_senders.remove(connection_id))
  ignore(ws_binary_senders.remove(connection_id))
  ignore(ws_pong_senders.remove(connection_id))
  ignore(ws_client_channels.remove(connection_id))
  match channels {
    Some(channels) =>
      for channel in channels {
        match ws_channel_clients.get(channel) {
          Some(clients) => ignore(clients.remove(connection_id))
          None => ()
        }
      }
    None => ()
  }
}

///|
pub fn register_ws_handler(mocket : App, port : Int) -> Unit {
  native_ws_handler_map.set(port, mocket)
}

///|
pub fn ws_send(id : String, msg : String) -> Unit {
  match ws_text_senders.get(id) {
    Some(send) => send(msg)
    None => ()
  }
}

///|
pub fn ws_send_bytes(id : String, msg : Bytes) -> Unit {
  match ws_binary_senders.get(id) {
    Some(send) => send(msg)
    None => ()
  }
}

///|
pub fn ws_pong(id : String) -> Unit {
  match ws_pong_senders.get(id) {
    Some(send) => send()
    None => ()
  }
}

///|
pub fn ws_subscribe(id : String, channel : String) -> Unit {
  let client_channels = match ws_client_channels.get(id) {
    Some(channels) => channels
    None => {
      let channels = []
      ws_client_channels.set(id, channels)
      channels
    }
  }
  if !client_channels.contains(channel) {
    client_channels.push(channel)
  }
  let channel_clients = match ws_channel_clients.get(channel) {
    Some(clients) => clients
    None => {
      let clients : Map[String, Unit] = Map([])
      ws_channel_clients.set(channel, clients)
      clients
    }
  }
  channel_clients.set(id, ())
}

///|
pub fn ws_unsubscribe(id : String, channel : String) -> Unit {
  match ws_channel_clients.get(channel) {
    Some(clients) => ignore(clients.remove(id))
    None => ()
  }
  match ws_client_channels.get(id) {
    Some(channels) => {
      let mut index = None
      for i = 0; i < channels.length(); i = i + 1 {
        if channels[i] == channel {
          index = Some(i)
          break
        }
      }
      match index {
        Some(i) => ignore(channels.remove(i))
        None => ()
      }
    }
    None => ()
  }
}

///|
pub fn ws_publish(channel : String, msg : String) -> Unit {
  match ws_channel_clients.get(channel) {
    Some(clients) => clients.keys().each(id => ws_send(id, msg))
    None => ()
  }
}

///|
fn WebSocketOutboundQueue::close(self : WebSocketOutboundQueue) -> Unit {
  self.closed = true
  self.frames.clear()
}

///|
fn WebSocketOutboundQueue::enqueue(
  self : WebSocketOutboundQueue,
  ws : @websocket.Conn,
  frame : OutboundFrame,
) -> Unit {
  if self.closed {
    return
  }
  self.frames.push(frame)
  if !self.draining {
    self.draining = true
    async_run(async fn() noraise { self.drain(ws) })
  }
}

///|
async fn WebSocketOutboundQueue::drain(
  self : WebSocketOutboundQueue,
  ws : @websocket.Conn,
) -> Unit noraise {
  for ;; {
    if self.closed {
      self.frames.clear()
      self.draining = false
      return
    }
    if self.frames.length() == 0 {
      self.draining = false
      return
    }
    let frame = self.frames[0]
    ignore(self.frames.remove(0))
    try {
      match frame {
        OutboundText(msg) => ws.send_text(msg)
        OutboundBinary(msg) => ws.send_binary(msg)
        OutboundPong => ws.ping()
      }
    } catch {
      _ => {
        self.close()
        self.draining = false
        return
      }
    }
  }
}

///|
fn register_native_ws_connection(
  connection_id : String,
  ws : @websocket.Conn,
) -> WebSocketOutboundQueue {
  let outbound = WebSocketOutboundQueue::new()
  register_ws_connection(
    connection_id,
    msg => outbound.enqueue(ws, OutboundText(msg)),
    msg => outbound.enqueue(ws, OutboundBinary(msg)),
    () => outbound.enqueue(ws, OutboundPong),
  )
  outbound
}

///|
/// An `@io.Writer` adapter that flushes every response write. This prevents
/// Server-Sent Events and other incremental responders from remaining in the
/// native connection buffer until the response ends.
priv struct NativeResponseWriter {
  conn : @http.ServerConnection
}

///|
async fn NativeResponseWriter::write_and_flush(
  self : NativeResponseWriter,
  data : &@io.Data,
) -> Unit {
  self.conn.write(data)
  self.conn.flush()
}

///|
async fn NativeResponseWriter::write_chunk_and_flush(
  self : NativeResponseWriter,
  bytes : Bytes,
  offset~ : Int,
  len~ : Int,
) -> Int {
  let written = self.conn.write_once(bytes, offset~, len~)
  self.conn.flush()
  written
}

///|
async fn NativeResponseWriter::write_reader_and_flush(
  self : NativeResponseWriter,
  reader : &@io.Reader,
) -> Unit {
  self.conn.write_reader(reader)
  self.conn.flush()
}

///|
impl @io.Writer for NativeResponseWriter with fn write(self, data) {
  self.write_and_flush(data)
}

///|
impl @io.Writer for NativeResponseWriter with fn write_once(
  self,
  bytes,
  offset~,
  len~,
) {
  self.write_chunk_and_flush(bytes, offset~, len~)
}

///|
impl @io.Writer for NativeResponseWriter with fn write_reader(self, reader) {
  self.write_reader_and_flush(reader)
}

///|
async fn send_native_response(
  request : @http.Request,
  conn : @http.ServerConnection,
  response : HttpResponse,
) -> Unit {
  let raw_headers = view_headers_to_strings(response.headers)
  let headers : Map[@http.CaseInsensitiveString, String] = Map([])
  raw_headers.each((key, value) => {
    if @header.is_valid_header_name(Show::to_string(key)) {
      headers[key] = @header.sanitize_header_value(value)
    }
  })
  let cookies = response.cookies
  conn.send_response(
    response.status_code.to_int(),
    "OK",
    extra_headers=headers,
    cookies~,
  )
  if request.meth != Head {
    let writer = NativeResponseWriter::{ conn, }
    match response.body {
      Some(body) => body.output(writer)
      None => ()
    }
  }
  conn.end_response()
}

///|
async fn handle_http_request(
  mocket : App,
  request : @http.Request,
  body_reader : &@io.Reader,
  conn : @http.ServerConnection,
) -> Unit {
  let headers = string_headers_to_views(request.headers)
  let http_method = request_method_to_string(request.meth)
  if request_has_body(http_method, headers) {
    let content_length = request.headers
      .get("content-length")
      .map(s => @string.parse_int(s.trim()) catch { _ => 0 })
      .unwrap_or(0)
    if mocket.max_body_size > 0 && content_length > mocket.max_body_size {
      send_native_response(
        request,
        conn,
        HttpResponse(RequestEntityTooLarge).body("Request body too large"),
      )
      return
    }
    let body = if mocket.max_body_size > 0 {
      read_http_body_limited(body_reader, mocket.max_body_size) catch {
        _ => None
      }
    } else {
      Some(body_reader.read_all().binary())
    }
    let response = match body {
      Some(body) =>
        dispatch_http(mocket, http_method, request.path, headers, body) catch {
          err => HttpResponse(InternalServerError).body(err.to_string())
        }
      None => HttpResponse(RequestEntityTooLarge).body("Request body too large")
    }
    send_native_response(request, conn, response)
    return
  }
  let response = dispatch_http(mocket, http_method, request.path, headers, b"") catch {
    err => HttpResponse(InternalServerError).body(err.to_string())
  }
  send_native_response(request, conn, response)
}

///|
async fn read_http_body_limited(reader : &@io.Reader, max_size : Int) -> Bytes? {
  if max_size <= 0 {
    return Some(reader.read_all().binary())
  }
  let buffer = Buffer()
  let chunk = FixedArray::make(8192, b'\x00')
  let mut total = 0
  let mut done = false
  while !done {
    let read = reader.read(chunk) catch { _ => return None }
    if read <= 0 {
      done = true
    } else if total + read > max_size {
      return None
    } else {
      for index in 0.. Bytes? {
  if max_size <= 0 {
    return Some(reader.read_all().binary())
  }
  let buf = Buffer()
  let chunk_size = 8192
  let chunk = FixedArray::make(chunk_size, b'\x00')
  for total = 0 {
    let n = reader.read(chunk) catch { _ => return None }
    if n <= 0 {
      break
    }
    if total + n > max_size {
      return None
    }
    let arr : Array[Byte] = []
    for i = 0; i < n; i = i + 1 {
      arr.push(chunk[i])
    }
    buf.write_bytes(Bytes::from_array(arr))
    continue total + n
  }
  Some(buf.to_bytes())
}

///|
async fn handle_websocket_request(
  port : Int,
  mocket : App,
  request : @http.Request,
  conn : @http.ServerConnection,
) -> Unit {
  match find_ws_route(mocket, request.path) {
    Some((handler, _params)) => {
      let ws = @websocket.from_http_server(request, conn)
      defer ws.close()
      let connection_id = next_ws_connection_id(port)
      let outbound = register_native_ws_connection(connection_id, ws)
      defer outbound.close()
      let peer = WebSocketPeer::{ connection_id, subscribed_channels: [], }
      handler(Open(peer))
      try {
        for ;; {
          let msg = ws.recv()
          match msg.kind {
            Text =>
              match read_ws_limited(msg, mocket.max_body_size) {
                Some(data) =>
                  handler(Message(peer, Text(@utf8.decode_lossy(data))))
                None => ()
              }
            Binary =>
              match read_ws_limited(msg, mocket.max_body_size) {
                Some(data) => handler(Message(peer, Binary(data)))
                None => ()
              }
          }
        }
      } catch {
        @websocket.ConnectionClosed(_, _) => ()
        _ => ()
      }
      outbound.close()
      handler(Close(peer))
      unregister_ws_connection(connection_id)
    }
    None => {
      let response = HttpResponse(NotFound).body("Not Found")
      send_native_response(request, conn, response)
    }
  }
}

///|
/// Listen and serve on `address`.
///
/// The native listener enables `SO_REUSEADDR` (`reuse_addr=true`) so that a
/// process can immediately rebind a port that a previous process left in
/// `TIME_WAIT`. On macOS/BSD this has the side effect that a bind to a specific
/// address may "steal" the specific interface from an existing wildcard bind
/// (`0.0.0.0`), and a wildcard bind can bind to the remaining interfaces if a
/// specific-interface bind already exists.
pub async fn listen_ffi(mocket : App, address : String) -> Unit noraise {
  let address = normalize_listen_address(address)
  let addr = @socket.Addr::parse(address) catch {
    err => {
      println("mocket: invalid native listen address \{address}: \{err}")
      return
    }
  }
  let port = addr.port()
  register_ws_handler(mocket, port)
  let server = @http.Server(addr, reuse_addr=true) catch {
    err => {
      println("mocket: failed to listen on \{address}: \{err}")
      return
    }
  }
  server.run_forever((request, body_reader, conn) => {
    if is_websocket_upgrade(request) {
      handle_websocket_request(port, mocket, request, conn)
    } else {
      handle_http_request(mocket, request, body_reader, conn)
    }
  }) catch {
    err => println("mocket: native server on \{address} stopped: \{err}")
  }
}

///|
fn normalize_listen_address(address : String) -> String {
  if address.has_prefix(":") {
    "0.0.0.0\{address}"
  } else {
    address
  }
}

///|
pub async fn serve_ffi(mocket : App, port~ : Int) -> Unit noraise {
  listen_ffi(mocket, "127.0.0.1:\{port}")
}

///|
pub fn __ws_emit(
  event_type : Bytes,
  connection_id : Bytes,
  payload : Bytes,
) -> Unit {
  let peer = WebSocketPeer::{
    connection_id: @utf8.decode_lossy(connection_id),
    subscribed_channels: [],
  }
  let handler = match native_ws_handler_map.values().collect() {
    [mocket, ..] =>
      match mocket.ws_static_routes.values().collect() {
        [handler, ..] => handler
        [] => fn(_) {  }
      }
    [] => fn(_) {  }
  }
  dispatch_ws_event(handler, peer, @utf8.decode_lossy(event_type), payload)
}