// The WebSocket bridge over mooncat's own raw HTTP/1.1 transport (`&@io.Reader`/`&@io.Writer`), the
// self-built counterpart of `handle_websocket` (which rides the async transport's
// `Conn::from_http_server`). Because we own the 101 response here, the negotiated subprotocol is
// echoed in `Sec-WebSocket-Protocol` — the header the `from_http_server` path cannot set — and the
// frames go through mooncat's own `websocket_frame` codec.

///|
/// Bridge a WebSocket connection to a moonasgi application over the raw HTTP/1.1 stream `reader` /
/// `writer` (← uvicorn's `WSProtocol`), the raw-transport twin of `handle_websocket`:
///
/// * `receive()` yields `websocket.connect` first, then reads messages with `ws_read_message`
///   (which reassembles fragments, answers pings, and echoes a close) — a text message becomes
///   `WebSocketReceive(text=..)`, a binary one `WebSocketReceive(bytes=..)`, and a peer close
///   `WebSocketDisconnect(code=..)`.
/// * `send(WebSocketAccept)` writes the self-built 101 (with the negotiated `Sec-WebSocket-Protocol`
///   echoed); `send(WebSocketSendText / WebSocketSendBytes)` writes a message frame;
///   `send(WebSocketClose)` writes a close frame; a close sent *before* accept declines the upgrade
///   with `403 Forbidden`, as uvicorn does.
///
/// A peer that breaks RFC 6455 fails the connection rather than being tolerated: the violation's
/// close code goes out as a close frame (§7.1.7) and the app sees a disconnect carrying that same
/// code, so an app can tell a protocol failure from a peer that simply went away (`1006`).
async fn serve_websocket_raw(
  app : @moonasgi.AsgiApp,
  req : Http1Request,
  reader : &@io.Reader,
  writer : &@io.Writer,
  config~ : Config,
  client_addr? : @socket.Addr,
) -> Unit {
  let (path, query) = split_query(req.target)
  let peer = match client_addr {
    Some(a) => Some(addr_pair(a))
    None => None
  }
  let (client, scheme) = if config.proxy_headers {
    proxy_rewrite(
      req.headers,
      peer,
      "ws",
      trusted=config.forwarded_allow_ips,
      websocket=true,
    )
  } else {
    (peer, "ws")
  }
  let scope = @moonasgi.Scope::WebSocket({
    http_version: "1.1",
    scheme,
    path,
    raw_path: @utf8.encode(path),
    query_string: @utf8.encode(query),
    root_path: config.root_path,
    headers: headers_to_pairs(req.headers),
    client,
    server: None,
    subprotocols: parse_subprotocols(req.headers),
    asgi: @moonasgi.AsgiVersion::websocket(),
    extensions: @moonasgi.Extensions::none(),
    state: Map([]),
  })
  let connected = Ref(false)
  let accepted = Ref(false)
  let closed = Ref(false)
  let receive : @moonasgi.Receive = () => {
    if not_yet(connected) {
      @moonasgi.Event::WebSocketConnect
    } else if closed.val {
      @moonasgi.Event::WebSocketDisconnect(code=1006, reason=None)
    } else {
      let msg : Result[WsMessage, Int] = Ok(
        ws_read_message(reader, writer, max_size=config.ws.max_size),
      ) catch {
        // A violation names the code the connection must be failed with; anything else is the
        // stream ending under us, which has no status to report.
        WsError(code, _) => Err(code)
        _ => Err(1006)
      }
      match msg {
        Err(code) => {
          closed.val = true
          if code != 1006 {
            // §7.1.7: fail the connection, but say why on the wire first.
            writer.write(
              ws_encode_frame(true, Close, ws_close_payload(code, "")),
            ) catch {
              _ => ()
            }
          }
          @moonasgi.Event::WebSocketDisconnect(code~, reason=None)
        }
        Ok(WsText(text)) =>
          @moonasgi.Event::WebSocketReceive(text=Some(text), bytes=None)
        Ok(WsBinary(bytes)) =>
          @moonasgi.Event::WebSocketReceive(text=None, bytes=Some(bytes))
        Ok(WsClose(code, _)) => {
          closed.val = true
          @moonasgi.Event::WebSocketDisconnect(code~, reason=None)
        }
      }
    }
  }
  let send : @moonasgi.Send = event => {
    match event {
      WebSocketAccept(subprotocol~, headers~) =>
        if !accepted.val {
          accepted.val = true
          let key = req.headers.get("sec-websocket-key").unwrap_or("")
          writer.write(websocket_handshake_response(key, subprotocol, headers))
        }
      WebSocketSendText(text) =>
        if accepted.val {
          writer.write(ws_encode_frame(true, Text, @utf8.encode(text)))
        }
      WebSocketSendBytes(data) =>
        if accepted.val {
          writer.write(ws_encode_frame(true, Binary, data))
        }
      WebSocketClose(code~, reason~) => {
        if accepted.val {
          writer.write(
            ws_encode_frame(true, Close, ws_close_payload(code, reason)),
          )
        } else {
          writer.write(@utf8.encode("HTTP/1.1 403 Forbidden\r\n\r\n"))
        }
        closed.val = true
      }
      _ => ()
    }
  }
  app(scope, receive, send)
}