#include <compat.h> // IWYU pragma: keep

#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <tlsuv/websocket.h>
#include <uv.h>
#include <zlib.h>

#include "ant.h"
#include "cfunc.h"
#include "ptr.h"
#include "list.h"
#include "loop.h"
#include "common.h"
#include "errors.h"
#include "inspector.h"
#include "internal.h"
#include "modules/timer.h"
#include "descriptors.h"

#include "http/websocket.h"
#include "modules/blob.h"
#include "modules/buffer.h"
#include "modules/events.h"
#include "modules/websocket.h"
#include "modules/url.h"
#include "net/listener.h"

struct websocket_state_s {
  ant_value_t obj;
  ant_value_t request_obj;
  tlsuv_websocket_t client;
  uv_connect_t connect_req;
  ant_conn_t *server_conn;
  uint64_t inspector_request_id;
  websocket_state_t *next;
  websocket_state_t **pprev;
  uint8_t *fragment_buf;
  size_t fragment_len;
  size_t fragment_cap;
  size_t max_payload_len;
  z_stream deflate_strm;
  z_stream inflate_strm;
  ant_ws_opcode_t fragment_opcode;
  uint16_t client_close_code;
  uint8_t ready_state;
  bool is_client: 1;
  bool close_emitted: 1;
  bool client_close_started: 1;
  bool client_close_was_clean: 1;
  bool fragmenting: 1;
  bool fragment_compressed: 1;
  bool per_message_deflate: 1;
  bool deflate_ready: 1;
  bool inflate_ready: 1;
  bool binary_type_blob: 1;
};

enum { WS_HANDLER_OPEN, WS_HANDLER_MESSAGE, WS_HANDLER_CLOSE, WS_HANDLER_ERROR, WS_HANDLER_COUNT };
static const char *const ws_handler_types[WS_HANDLER_COUNT] = { "open", "message", "close", "error" };

enum {
  WS_CONNECTING = 0,
  WS_OPEN = 1,
  WS_CLOSING = 2,
  WS_CLOSED = 3,
};

static constexpr uint32_t WS_NATIVE_TAG = Ant_NativeTag("WSOB");
static constexpr uint32_t WS_DEFAULT_MAX_PAYLOAD_LEN = 16u * 1024u * 1024u;

static websocket_state_t *websocket_data(ant_value_t value) {
  return (websocket_state_t *)js_get_native(value, WS_NATIVE_TAG);
}

static websocket_state_t *websocket_from_client(tlsuv_websocket_t *client) {
  return client ? (websocket_state_t *)((char *)client - offsetof(websocket_state_t, client)) : NULL;
}

static void websocket_fragment_reset(websocket_state_t *ws) {
  if (!ws) return;
  ws->fragment_len = 0;
  ws->fragment_opcode = 0;
  ws->fragmenting = false;
  ws->fragment_compressed = false;
}

static void websocket_fragment_clear(websocket_state_t *ws) {
  if (!ws) return;
  websocket_fragment_reset(ws);
  free(ws->fragment_buf);
  ws->fragment_buf = NULL;
  ws->fragment_cap = 0;
}

static bool websocket_fragment_append(websocket_state_t *ws, const uint8_t *data, size_t len) {
  if (!ws) return false;
  if (len > SIZE_MAX - ws->fragment_len) return false;
  size_t need = ws->fragment_len + len;
  if (need > ws->fragment_cap) {
    size_t cap = ws->fragment_cap ? ws->fragment_cap : 1024;
    while (cap < need) {
      if (cap > SIZE_MAX / 2) { cap = need; break; }
      cap *= 2;
    }
    uint8_t *next = realloc(ws->fragment_buf, cap);
    if (!next) return false;
    ws->fragment_buf = next;
    ws->fragment_cap = cap;
  }
  if (len > 0) memcpy(ws->fragment_buf + ws->fragment_len, data, len);
  ws->fragment_len = need;
  return true;
}

static void websocket_free_state(websocket_state_t *ws) {
  if (!ws) return;
  Ant_List_Remove(ws);
  websocket_fragment_clear(ws);
  if (ws->deflate_ready) deflateEnd(&ws->deflate_strm);
  if (ws->inflate_ready) inflateEnd(&ws->inflate_strm);
  free(ws);
}

static void websocket_sync_state(ant_t *js, websocket_state_t *ws) {
  if (!ws || !is_object_type(ws->obj)) return;
  js_set(js, ws->obj, "readyState", js_mknum(ws->ready_state));
}

static const uint8_t ws_deflate_flush_tail[] = { 
  0x00, 0x00, 
  0xff, 0xff 
};

static bool websocket_deflate_message(websocket_state_t *ws, const uint8_t *data, size_t len, uint8_t **out, size_t *out_len) {
  z_stream *strm = &ws->deflate_strm;
  uint8_t *buf = NULL;
  
  uLong bound = 0;
  int rc = 0;

  if (out) *out = NULL;
  if (out_len) *out_len = 0;
  if (!out || !out_len) return false;

  if (!ws->deflate_ready) {
    memset(strm, 0, sizeof(*strm));
    if (deflateInit2(strm, Z_DEFAULT_COMPRESSION, Z_DEFLATED, -15, 8, Z_DEFAULT_STRATEGY) != Z_OK) return false;
    ws->deflate_ready = true;
  } else if (deflateReset(strm) != Z_OK) return false;

  bound = deflateBound(strm, (uLong)len) + sizeof(ws_deflate_flush_tail);
  buf = malloc((size_t)bound);
  if (!buf) return false;

  strm->next_in = (Bytef *)(uintptr_t)data;
  strm->avail_in = (uInt)len;
  strm->next_out = buf;
  strm->avail_out = (uInt)bound;

  rc = deflate(strm, Z_SYNC_FLUSH);
  if (rc != Z_OK || strm->avail_in != 0) {
    free(buf);
    return false;
  }

  *out_len = strm->total_out;
  if (*out_len >= sizeof(ws_deflate_flush_tail) &&
      memcmp(buf + *out_len - sizeof(ws_deflate_flush_tail), ws_deflate_flush_tail, sizeof(ws_deflate_flush_tail)) == 0) {
    *out_len -= sizeof(ws_deflate_flush_tail);
  }

  *out = buf;
  return true;
}

static bool websocket_inflate_chunk(
  z_stream *strm, const uint8_t *in, size_t in_len, size_t max_len,
  uint8_t **buf, size_t *cap, bool *done
) {
  strm->next_in = (Bytef *)(uintptr_t)in;
  strm->avail_in = (uInt)in_len;

  while (strm->avail_in > 0) {
    if (strm->total_out == *cap) {
      size_t next_cap = *cap * 2;
      if (max_len > 0 && *cap >= max_len) return false;
      if (next_cap < *cap) return false;
      if (max_len > 0 && next_cap > max_len) next_cap = max_len;
      uint8_t *next = realloc(*buf, next_cap);
      if (!next) return false;
      *buf = next;
      *cap = next_cap;
    }

    strm->next_out = *buf + strm->total_out;
    strm->avail_out = (uInt)(*cap - strm->total_out);
    
    int rc = inflate(strm, Z_SYNC_FLUSH);
    if (rc == Z_BUF_ERROR && strm->avail_in == 0) break;
    if (rc != Z_OK && rc != Z_STREAM_END) return false;
    if (rc == Z_STREAM_END) { *done = true; break; }
    if (strm->avail_out != 0 && strm->avail_in == 0) break;
  }
  return true;
}

static bool websocket_inflate_message(
  websocket_state_t *ws,
  const uint8_t *data,
  size_t len,
  size_t max_len,
  uint8_t **out,
  size_t *out_len
) {
  z_stream *strm = &ws->inflate_strm;
  uint8_t *buf = NULL;
  size_t cap = len > 1024 ? len * 2 : 1024;
  bool done = false;

  if (out) *out = NULL;
  if (out_len) *out_len = 0;
  if (!out || !out_len) return false;
  if (max_len > 0 && cap > max_len) cap = max_len;
  if (cap == 0) cap = 1;

  if (!ws->inflate_ready) {
    memset(strm, 0, sizeof(*strm));
    if (inflateInit2(strm, -15) != Z_OK) return false;
    ws->inflate_ready = true;
  } else if (inflateReset(strm) != Z_OK) return false;

  buf = malloc(cap);
  if (!buf) return false;

  if (!websocket_inflate_chunk(strm, data, len, max_len, &buf, &cap, &done)) goto fail;
  if (!done && !websocket_inflate_chunk(strm, ws_deflate_flush_tail, sizeof(ws_deflate_flush_tail), max_len, &buf, &cap, &done)) goto fail;

  if (max_len > 0 && strm->total_out > max_len) goto fail;
  *out_len = strm->total_out;
  *out = buf;
  
  return true;

fail:
  free(buf);
  return false;
}

static void websocket_emit_simple(ant_t *js, websocket_state_t *ws, int idx) {
  ant_value_t event = Ant_EventTarget_CreateEvent(js, ws->obj, js->builtins.event_proto, ANT_EVENT_BASE, ws_handler_types[idx]);
  if (!is_object_type(event)) return;
  Ant_EventTarget_Dispatch(js, ws->obj, event);
  js_maybe_drain_microtasks(js);
}

static ant_value_t websocket_invalid_this(ant_t *js, ant_value_t this_obj) {
  return websocket_data(this_obj) ? 0 : js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WebSocket");
}

ANT_EVENT_HANDLER_ACCESSORS(js_websocket, open, websocket_invalid_this)
ANT_EVENT_HANDLER_ACCESSORS(js_websocket, message, websocket_invalid_this)
ANT_EVENT_HANDLER_ACCESSORS(js_websocket, close, websocket_invalid_this)
ANT_EVENT_HANDLER_ACCESSORS(js_websocket, error, websocket_invalid_this)

static ant_value_t websocket_on_impl(ant_t *js, ant_value_t *args, int nargs, bool once, bool remove) {
  ant_value_t self = js_getthis(js);
  if (!websocket_data(self)) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WebSocket");
  if (nargs < 2 || !is_callable(args[1])) return js_mkerr_typed(js, JS_ERR_TYPE, "listener must be a function");
  
  ant_value_t key = js_tostring_val(js, args[0]);
  if (is_err(key)) return key;
  if (remove) eventemitter_remove_listener_val(js, self, key, args[1]);
  else eventemitter_add_listener_val(js, self, key, args[1], once);
  
  return self;
}

static ant_value_t js_websocket_on(ant_params_t) { return websocket_on_impl(js, args, nargs, false, false); }
static ant_value_t js_websocket_once(ant_params_t) { return websocket_on_impl(js, args, nargs, true, false); }
static ant_value_t js_websocket_off(ant_params_t) { return websocket_on_impl(js, args, nargs, false, true); }

static void websocket_emit_message(ant_t *js, websocket_state_t *ws, const uint8_t *data, size_t len, bool binary) {
  ant_value_t event = Ant_EventTarget_CreateEvent(js, ws->obj, js->builtins.message_event_proto, ANT_EVENT_MESSAGE, "message");
  if (!is_object_type(event)) return;

  GC_ROOT_SAVE(mark, js);
  GC_ROOT_PIN(js, event);

  ant_value_t data_val;
  if (!binary) data_val = js_mkstr(js, data, len);
  else if (ws->binary_type_blob) data_val = blob_create(js, data, len, "");
  else data_val = create_arraybuffer_copy(js, data, len);

  if (!is_err(data_val)) {
    ant_value_t empty = js_mkstr(js, "", 0);
    Ant_Event_SetField(js, event, ANT_EVENT_FIELD_MESSAGE_data, data_val);
    Ant_Event_SetField(js, event, ANT_EVENT_FIELD_MESSAGE_origin, empty);
    Ant_Event_SetField(js, event, ANT_EVENT_FIELD_MESSAGE_lastEventId, empty);
    Ant_EventTarget_Dispatch(js, ws->obj, event);
  }
  
  js_maybe_drain_microtasks(js);
  GC_ROOT_RESTORE(js, mark);
}

static void websocket_emit_close(ant_t *js, websocket_state_t *ws, uint16_t code, const char *reason, bool was_clean) {
  if (!ws || ws->close_emitted) return;
  ant_inspector_websocket_closed(ws->inspector_request_id);
  
  ws->close_emitted = true;
  ws->ready_state = WS_CLOSED;
  websocket_fragment_clear(ws);
  websocket_sync_state(js, ws);

  ant_value_t event = Ant_EventTarget_CreateEvent(js, ws->obj, js->builtins.close_event_proto, ANT_EVENT_CLOSE, "close");
  if (!is_object_type(event)) return;
  
  Ant_Event_SetField(js, event, ANT_EVENT_FIELD_CLOSE_code, js_mknum(code));
  Ant_Event_SetField(js, event, ANT_EVENT_FIELD_CLOSE_reason, js_mkstr(js, reason ? reason : "", reason ? strlen(reason) : 0));
  Ant_Event_SetField(js, event, ANT_EVENT_FIELD_CLOSE_wasClean, js_bool(was_clean));
  
  Ant_EventTarget_Dispatch(js, ws->obj, event);
  js_maybe_drain_microtasks(js);
}

static void websocket_client_close_cb(uv_handle_t *handle) {
  ant_t *js = Ant_Loop_Isolate(handle->loop);
  websocket_state_t *ws = websocket_from_client((tlsuv_websocket_t *)handle);
  websocket_emit_close(js, ws, ws->client_close_code, "", ws->client_close_was_clean);
  Ant_List_Remove(ws);
}

static void websocket_close_client(websocket_state_t *ws, uint16_t code, bool was_clean) {
  if (!ws || ws->client_close_started) return;
  ws->client_close_started = true;
  ws->client_close_code = code;
  ws->client_close_was_clean = was_clean;
  tlsuv_websocket_close(&ws->client, websocket_client_close_cb);
}

static void websocket_client_connect_cb(uv_connect_t *req, int status) {
  ant_t *js = Ant_Loop_Isolate(req->handle->loop);
  websocket_state_t *ws = websocket_from_client((tlsuv_websocket_t *)req->handle);
  if (!ws) return;

  if (status == 0) {
    ant_http_header_t upgrade = { "Upgrade", "websocket", NULL };
    ant_http_header_t connection = { "Connection", "Upgrade", &upgrade };
    
    ant_inspector_websocket_response(ws->inspector_request_id, 101, "Switching Protocols", &connection);
    ws->ready_state = WS_OPEN;
    
    websocket_sync_state(js, ws);
    websocket_emit_simple(js, ws, WS_HANDLER_OPEN);
    
    return;
  }

  ws->ready_state = WS_CLOSED;
  ant_inspector_websocket_error(ws->inspector_request_id, uv_strerror(status));
  
  websocket_sync_state(js, ws);
  websocket_emit_simple(js, ws, WS_HANDLER_ERROR);
  websocket_close_client(ws, 1006, false);
}

static void websocket_client_read_cb(uv_stream_t *handle, ssize_t nread, const uv_buf_t *buf) {
  ant_t *js = Ant_Loop_Isolate(handle->loop);
  websocket_state_t *ws = websocket_from_client((tlsuv_websocket_t *)handle);
  if (!ws) return;

  if (nread > 0) {
    bool binary = ws->client.last_frame_binary;
    ant_inspector_websocket_frame_received(ws->inspector_request_id, (const uint8_t *)buf->base, (size_t)nread, binary);
    websocket_emit_message(js, ws, (const uint8_t *)buf->base, (size_t)nread, binary);
    return;
  }

  if (nread < 0) {
    if (nread != UV_EOF) {
      ant_inspector_websocket_error(ws->inspector_request_id, uv_strerror((int)nread));
      websocket_emit_simple(js, ws, WS_HANDLER_ERROR);
    }
    ws->ready_state = WS_CLOSING;
    websocket_sync_state(js, ws);
    websocket_close_client(ws, nread == UV_EOF ? 1000 : 1006, nread == UV_EOF);
  }
}

static void websocket_write_cb(uv_write_t *req, int status) {
  free(req->data);
  free(req);
}

static int websocket_client_write_frame(websocket_state_t *ws, ant_ws_opcode_t opcode, const uint8_t *bytes, size_t len) {
  if (!ws || !ws->client.tr || !ws->client.tr_write) return UV_ENOTCONN;

  size_t frame_len = 0;
  uint8_t *frame = ant_ws_encode_frame(opcode, bytes, len, true, false, &frame_len);
  if (!frame) return UV_ENOMEM;

  uv_write_t *wr = calloc(1, sizeof(*wr));
  if (!wr) {
    free(frame);
    return UV_ENOMEM;
  }

  wr->data = frame;
  uv_buf_t buf = uv_buf_init((char *)frame, (unsigned int)frame_len);
  int rc = ws->client.tr_write(wr, ws->client.tr, &buf, 1, websocket_write_cb);
  if (rc != 0) {
    free(frame);
    free(wr);
  }
  
  return rc;
}

static void websocket_server_close_with_code(websocket_state_t *ws, ant_conn_t *conn, uint16_t code, const char *reason) {
  size_t frame_len = 0;
  uint8_t *frame = ant_ws_encode_close_frame(code, reason, false, &frame_len);
  if (frame && conn && !ant_conn_is_closing(conn)) ant_conn_write(conn, (char *)frame, frame_len, NULL, NULL);
  if (conn) ant_conn_shutdown(conn);
}

static void websocket_finalize(ant_t *js, ant_object_t *obj) {
  (void)js;
  ant_value_t value = js_obj_from_ptr(obj);
  websocket_state_t *ws = websocket_data(value);
  if (!ws) return;
  js_clear_native(value, WS_NATIVE_TAG);
  websocket_free_state(ws);
}

static websocket_state_t *websocket_state_new(ant_t *js, ant_value_t obj) {
  websocket_state_t *ws = calloc(1, sizeof(*ws));
  if (!ws) return NULL;
  ws->obj = obj;
  ws->ready_state = WS_CONNECTING;
  ws->max_payload_len = WS_DEFAULT_MAX_PAYLOAD_LEN;
  Ant_List_Insert(&js->active_websockets, ws);
  return ws;
}

static ant_value_t websocket_create_object(ant_t *js) {
  ant_value_t obj = Ant_Object_CreateBare(js);
  if (is_object_type(js->builtins.websocket_proto)) js_set_proto_init(obj, js->builtins.websocket_proto);
  js_set_slot(obj, SLOT_BRAND, js_mknum(BRAND_EVENTTARGET));
  js_set(js, obj, "bufferedAmount", js_mknum(0));
  js_set(js, obj, "extensions", js_mkstr(js, "", 0));
  js_set(js, obj, "protocol", js_mkstr(js, "", 0));
  js_set_finalizer(obj, websocket_finalize);
  return obj;
}

static ant_value_t websocket_resolve_url(ant_t *js, ant_value_t input) {
  const char *text = js_getstr(js, input, NULL);
  url_state_t parsed = {0};
  
  if (!text || Ant_URL_ParseDocumentRelative(js, text, &parsed) != 0) {
    url_state_clear(&parsed);
    return js_mkerr_typed(js, JS_ERR_SYNTAX, "Failed to construct 'WebSocket': The URL '%s' is invalid.", text ? text : "");
  }

  const char *protocol = parsed.protocol ? parsed.protocol : "";
  const char *scheme = NULL;
  size_t skip = 0;
  
  if (strcmp(protocol, "ws:") == 0 || strcmp(protocol, "wss:") == 0) scheme = "";
  else if (strcmp(protocol, "http:") == 0) { scheme = "ws"; skip = 4; }
  else if (strcmp(protocol, "https:") == 0) { scheme = "wss"; skip = 5; }

  ant_value_t result;
  if (!scheme) result = js_mkerr_typed(
    js, JS_ERR_SYNTAX, "Failed to construct 'WebSocket': The URL's scheme must be either 'http', 'https', 'ws', or 'wss'. '%s' is not allowed.", protocol
  );
  else if (parsed.hash && parsed.hash[0]) result = js_mkerr_typed(
    js, JS_ERR_SYNTAX, "Failed to construct 'WebSocket': The URL contains a fragment identifier ('%s'). Fragment identifiers are not allowed in WebSocket URLs.", parsed.hash
  );
  else {
    char *href = build_href(&parsed);
    if (!href) result = js_mkerr(js, "oom");
    else {
      size_t scheme_len = strlen(scheme), rest_len = strlen(href) - skip;
      char *out = malloc(scheme_len + rest_len + 1);
      if (!out) result = js_mkerr(js, "oom");
      else {
        memcpy(out, scheme, scheme_len);
        memcpy(out + scheme_len, href + skip, rest_len + 1);
        result = js_mkstr(js, out, scheme_len + rest_len);
        free(out);
      }
      free(href);
    }
  }

  url_state_clear(&parsed);
  return result;
}

static ant_value_t js_websocket_ctor(ant_params_t) {
  if (vtype(call_new_target) == kTypeUndefined)
    return js_mkerr_typed(js, JS_ERR_TYPE, "WebSocket constructor requires 'new'");
  if (nargs < 1) return js_mkerr_typed(js, JS_ERR_TYPE, "WebSocket URL is required");

  ant_value_t url_val = js_tostring_val(js, args[0]);
  if (is_err(url_val)) return url_val;
  
  ant_value_t resolved = websocket_resolve_url(js, url_val);
  if (is_err(resolved)) return resolved;
  
  url_val = resolved;
  const char *url = js_getstr(js, url_val, NULL);

  ant_value_t obj = websocket_create_object(js);
  websocket_state_t *ws = websocket_state_new(js, obj);
  if (!ws) return js_mkerr_typed(js, JS_ERR_TYPE, "Out of memory");
  
  ws->is_client = true;
  js_set_native(obj, ws, WS_NATIVE_TAG);
  js_set(js, obj, "url", url_val);
  websocket_sync_state(js, ws);
  
  ws->inspector_request_id = ant_inspector_websocket_created(url);
  ant_inspector_websocket_request(ws->inspector_request_id, NULL);

  if (tlsuv_websocket_init(js->loop, &ws->client) != 0) {
    ws->ready_state = WS_CLOSED;
    ant_inspector_websocket_error(ws->inspector_request_id, "WebSocket initialization failed");
    websocket_sync_state(js, ws);
    Ant_List_Remove(ws);
    return obj;
  }

  int rc = tlsuv_websocket_connect(
    &ws->connect_req, &ws->client, url, 
    websocket_client_connect_cb, websocket_client_read_cb
  );
  
  if (rc != 0) {
    ws->ready_state = WS_CLOSED;
    ant_inspector_websocket_error(ws->inspector_request_id, uv_strerror(rc));
    
    websocket_sync_state(js, ws);
    websocket_emit_simple(js, ws, WS_HANDLER_ERROR);
    websocket_close_client(ws, 1006, false);
  }

  return obj;
}

static ant_value_t js_websocket_get_binary_type(ant_params_t) {
  websocket_state_t *ws = websocket_data(js_getthis(js));
  if (!ws) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WebSocket");
  return ws->binary_type_blob ? js_mkstr(js, "blob", 4) : js_mkstr(js, "arraybuffer", 11);
}

static ant_value_t js_websocket_set_binary_type(ant_params_t) {
  websocket_state_t *ws = websocket_data(js_getthis(js));
  if (!ws) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WebSocket");
  if (nargs < 1) return js_mkundef();
  
  ant_value_t str = js_tostring_val(js, args[0]);
  if (is_err(str)) return str;
  
  const char *value = js_getstr(js, str, NULL);
  if (value && strcmp(value, "blob") == 0) ws->binary_type_blob = true;
  else if (value && strcmp(value, "arraybuffer") == 0) ws->binary_type_blob = false;
  
  return js_mkundef();
}

static bool websocket_bytes_from_value(ant_t *js, ant_value_t value, const uint8_t **bytes, size_t *len, ant_value_t *owned_str) {
  if (buffer_source_get_bytes(js, value, bytes, len)) {
    if (owned_str) *owned_str = js_mkundef();
    return true;
  }

  ant_value_t str = js_tostring_val(js, value);
  if (is_err(str)) return false;
  if (owned_str) *owned_str = str;
  *bytes = (const uint8_t *)js_getstr(js, str, len);
  return *bytes != NULL;
}

static ant_value_t js_websocket_send(ant_params_t) {
  websocket_state_t *ws = websocket_data(js_getthis(js));
  const uint8_t *bytes = NULL;
  size_t len = 0;
  ant_value_t keepalive = js_mkundef();
  bool binary = false;

  if (!ws) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WebSocket");
  if (ws->ready_state != WS_OPEN) return js_mkerr_typed(js, JS_ERR_TYPE, "WebSocket is not open");
  if (nargs < 1) return js_mkundef();
  binary = buffer_source_get_bytes(js, args[0], &bytes, &len);
  if (!binary && !websocket_bytes_from_value(js, args[0], &bytes, &len, &keepalive))
    return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WebSocket message");

  if (ws->is_client) {
    int rc = websocket_client_write_frame(ws, binary ? ANT_WS_OPCODE_BINARY : ANT_WS_OPCODE_TEXT, bytes, len);
    if (rc != 0) return js_mkerr_typed(js, JS_ERR_TYPE, "%s", uv_strerror(rc));
    ant_inspector_websocket_frame_sent(ws->inspector_request_id, bytes, len, binary);
    return js_mkundef();
  }

  if (ws->server_conn) {
    size_t frame_len = 0;
    uint8_t *frame = NULL;
    uint8_t *compressed = NULL;
    size_t compressed_len = 0;
    bool compressed_frame = false;
    if (ws->per_message_deflate) {
      if (!websocket_deflate_message(ws, bytes, len, &compressed, &compressed_len))
        return js_mkerr_typed(js, JS_ERR_TYPE, "Out of memory");
      bytes = compressed;
      len = compressed_len;
      compressed_frame = true;
    }
    frame = ant_ws_encode_frame(
      binary ? ANT_WS_OPCODE_BINARY : ANT_WS_OPCODE_TEXT,
      bytes, len, false, compressed_frame, &frame_len
    );
    free(compressed);
    if (!frame) return js_mkerr_typed(js, JS_ERR_TYPE, "Out of memory");
    int rc = ant_conn_write(ws->server_conn, (char *)frame, frame_len, NULL, NULL);
    if (rc != 0) {
      free(frame);
      return js_mkerr_typed(js, JS_ERR_TYPE, "%s", uv_strerror(rc));
    }
  }
  return js_mkundef();
}

static ant_value_t js_websocket_close(ant_params_t) {
  websocket_state_t *ws = websocket_data(js_getthis(js));
  uint16_t code = nargs > 0 && vtype(args[0]) == kTypeNumber ? (uint16_t)js_getnum(args[0]) : 1000;
  const char *reason = NULL;

  if (!ws) return js_mkerr_typed(js, JS_ERR_TYPE, "Invalid WebSocket");
  if (ws->ready_state == WS_CLOSED || ws->ready_state == WS_CLOSING) return js_mkundef();
  if (nargs > 1 && vtype(args[1]) != kTypeUndefined && vtype(args[1]) != kTypeNull) {
    ant_value_t reason_val = js_tostring_val(js, args[1]);
    if (is_err(reason_val)) return reason_val;
    reason = js_getstr(js, reason_val, NULL);
  }

  ws->ready_state = WS_CLOSING;
  websocket_sync_state(js, ws);
  if (ws->is_client) {
    websocket_close_client(ws, 1000, true);
  } else if (ws->server_conn) {
    size_t frame_len = 0;
    uint8_t *frame = ant_ws_encode_close_frame(code, reason, false, &frame_len);
    if (frame) ant_conn_write(ws->server_conn, (char *)frame, frame_len, NULL, NULL);
    ant_conn_shutdown(ws->server_conn);
  }
  return js_mkundef();
}

static ant_value_t websocket_event_ctor_init(
  ant_t *js, ant_value_t *args, int nargs, ant_value_t new_target,
  ant_value_t fallback_proto, ant_event_kind_t kind, const char *default_type
) {
  ant_value_t type = nargs > 0 ? js_tostring_val(js, args[0]) : js_mkstr(js, default_type, strlen(default_type));
  if (is_err(type)) return type;
  ant_value_t proto = js_instance_proto_from_new_target(js, fallback_proto, new_target);
  return Ant_Event_Create(js, proto, kind, type, 0);
}

static ant_value_t websocket_event_init_field(
  ant_t *js, ant_value_t event, ant_value_t init, const char *key,
  unsigned int index, ant_value_t fallback, ant_value_t (*convert)(ant_t *, ant_value_t)
) {
  ant_value_t value = is_object_type(init) ? js_get(js, init, key) : js_mkundef();
  if (is_err(value)) return value;
  if (vtype(value) == kTypeUndefined) value = fallback;
  else if (convert) value = convert(js, value);
  if (is_err(value)) return value;
  Ant_Event_SetField(js, event, index, value);
  return js_mkundef();
}

static ant_value_t websocket_to_bool(ant_t *js, ant_value_t value) {
  return js_bool(js_truthy(js, value));
}

static ant_value_t websocket_to_number(ant_t *js, ant_value_t value) {
  double number = js_to_number(js, value);
  return Ant_Exception_Pending(js) ? Ant_Exception_Current(js) : js_mknum(number);
}

static ant_value_t js_message_event_ctor(ant_params_t) {
  if (vtype(call_new_target) == kTypeUndefined)
    return js_mkerr_typed(js, JS_ERR_TYPE, "MessageEvent constructor requires 'new'");
  
  ant_value_t event = websocket_event_ctor_init(js, args, nargs, call_new_target, js->builtins.message_event_proto, ANT_EVENT_MESSAGE, "message");
  if (is_err(event)) return event;
  
  ant_value_t init = nargs > 1 ? args[1] : js_mkundef();
  ant_value_t empty = js_mkstr(js, "", 0);
  ant_value_t r;
  
  if (is_err(r = websocket_event_init_field(js, event, init, "data", ANT_EVENT_FIELD_MESSAGE_data, js_mknull(), NULL))) return r;
  if (is_err(r = websocket_event_init_field(js, event, init, "origin", ANT_EVENT_FIELD_MESSAGE_origin, empty, js_tostring_val))) return r;
  if (is_err(r = websocket_event_init_field(js, event, init, "lastEventId", ANT_EVENT_FIELD_MESSAGE_lastEventId, empty, js_tostring_val))) return r;
  if (is_err(r = websocket_event_init_field(js, event, init, "source", ANT_EVENT_FIELD_MESSAGE_source, js_mknull(), NULL))) return r;
  if (is_err(r = websocket_event_init_field(js, event, init, "ports", ANT_EVENT_FIELD_MESSAGE_ports, js_mkarr(js), NULL))) return r;
  
  return event;
}

static ant_value_t js_close_event_ctor(ant_params_t) {
  if (vtype(call_new_target) == kTypeUndefined)
    return js_mkerr_typed(js, JS_ERR_TYPE, "CloseEvent constructor requires 'new'");

  ant_value_t event = websocket_event_ctor_init(js, args, nargs, call_new_target, js->builtins.close_event_proto, ANT_EVENT_CLOSE, "close");
  if (is_err(event)) return event;
  
  ant_value_t init = nargs > 1 ? args[1] : js_mkundef();
  ant_value_t r;
  
  if (is_err(r = websocket_event_init_field(js, event, init, "wasClean", ANT_EVENT_FIELD_CLOSE_wasClean, js_false, websocket_to_bool))) return r;
  if (is_err(r = websocket_event_init_field(js, event, init, "code", ANT_EVENT_FIELD_CLOSE_code, js_mknum(0), websocket_to_number))) return r;
  if (is_err(r = websocket_event_init_field(js, event, init, "reason", ANT_EVENT_FIELD_CLOSE_reason, js_mkstr(js, "", 0), js_tostring_val))) return r;
  
  return event;
}

ant_value_t ant_websocket_accept_server(
  ant_t *js, ant_conn_t *conn, ant_value_t request_obj, 
  const char *protocol, const ant_websocket_server_options_t *options
) {
  ant_value_t obj = websocket_create_object(js);
  websocket_state_t *ws = websocket_state_new(js, obj);
  if (!ws) return js_mkerr_typed(js, JS_ERR_TYPE, "Out of memory");
  
  ws->server_conn = conn;
  ws->request_obj = request_obj;
  ws->ready_state = WS_CONNECTING;
  
  if (options) {
    ws->max_payload_len = options->max_payload_len;
    ws->per_message_deflate = options->per_message_deflate;
  }
  
  js_set_native(obj, ws, WS_NATIVE_TAG);
  js_set(js, obj, "url", js_mkstr(js, "", 0));
  
  if (protocol) js_set(js, obj, "protocol", js_mkstr(js, protocol, strlen(protocol)));
  websocket_sync_state(js, ws);
  
  return obj;
}

void ant_websocket_server_open(ant_t *js, ant_value_t socket_obj) {
  websocket_state_t *ws = websocket_data(socket_obj);
  if (!ws) return;
  
  ws->ready_state = WS_OPEN;
  websocket_sync_state(js, ws);
  websocket_emit_simple(js, ws, WS_HANDLER_OPEN);
}

void ant_websocket_server_on_read(ant_t *js, ant_value_t socket_obj, ant_conn_t *conn) {
  static const void *dispatch[] = {
    [ANT_WS_OPCODE_CONTINUATION] = &&l_continuation,
    [ANT_WS_OPCODE_TEXT]         = &&l_message,
    [ANT_WS_OPCODE_BINARY]       = &&l_message,
    [ANT_WS_OPCODE_CLOSE]        = &&l_close,
    [ANT_WS_OPCODE_PING]         = &&l_ping,
  };

  websocket_state_t *ws = websocket_data(socket_obj);
  if (!ws || !conn) return;

  while (ant_conn_buffer_len(conn) > 0) {
    ant_ws_frame_t frame = {0};
    
    ant_ws_frame_result_t result = ant_ws_parse_frame(
      (uint8_t *)(uintptr_t)ant_conn_buffer(conn),
      ant_conn_buffer_len(conn), true,
      ws->per_message_deflate, true, &frame
    );
    
    if (result == ANT_WS_FRAME_INCOMPLETE) return;
    
    if (result == ANT_WS_FRAME_PROTOCOL_ERROR) {
      websocket_emit_simple(js, ws, WS_HANDLER_ERROR);
      ant_conn_close(conn);
      return;
    }

    ant_conn_consume(conn, frame.consumed_len);
    if (frame.opcode < sizeof(dispatch) / sizeof(*dispatch) && dispatch[frame.opcode])
      goto *dispatch[frame.opcode];
    goto l_done;

    l_message:
      if (ws->fragmenting) goto l_protocol_error;
      if (frame.payload_len > ws->max_payload_len) goto l_too_big;
      if (frame.fin) {
        if (frame.rsv1) {
          uint8_t *inflated = NULL;
          size_t inflated_len = 0;
          if (!websocket_inflate_message(ws, frame.payload, frame.payload_len, ws->max_payload_len, &inflated, &inflated_len))
            goto l_protocol_error;
          websocket_emit_message(js, ws, inflated, inflated_len, frame.opcode == ANT_WS_OPCODE_BINARY);
          free(inflated);
        } else websocket_emit_message(js, ws, frame.payload, frame.payload_len, frame.opcode == ANT_WS_OPCODE_BINARY);
      } else {
        ws->fragmenting = true;
        ws->fragment_opcode = frame.opcode;
        ws->fragment_compressed = frame.rsv1;
        if (!websocket_fragment_append(ws, frame.payload, frame.payload_len)) goto l_protocol_error;
      }
      goto l_done;

    l_continuation:
      if (!ws->fragmenting) goto l_protocol_error;
      if (frame.rsv1) goto l_protocol_error;
      if (!websocket_fragment_append(ws, frame.payload, frame.payload_len)) goto l_protocol_error;
      if (ws->fragment_len > ws->max_payload_len) goto l_too_big;
      if (frame.fin) {
        if (ws->fragment_compressed) {
          uint8_t *inflated = NULL;
          size_t inflated_len = 0;
          if (!websocket_inflate_message(ws, ws->fragment_buf, ws->fragment_len, ws->max_payload_len, &inflated, &inflated_len))
            goto l_protocol_error;
          websocket_emit_message(js, ws, inflated, inflated_len, ws->fragment_opcode == ANT_WS_OPCODE_BINARY);
          free(inflated);
        } else websocket_emit_message(js, ws, ws->fragment_buf, ws->fragment_len, ws->fragment_opcode == ANT_WS_OPCODE_BINARY);
        websocket_fragment_reset(ws);
      }
      goto l_done;

    l_ping: {
      size_t out_len = 0;
      uint8_t *out = ant_ws_encode_frame(ANT_WS_OPCODE_PONG, frame.payload, frame.payload_len, false, false, &out_len);
      if (out) ant_conn_write(conn, (char *)out, out_len, NULL, NULL);
      goto l_done;
    }

    l_close: {
      uint16_t code = 1000;
      const char *reason = "";
      char reason_buf[124];
      if (frame.payload_len >= 2) {
        code = (uint16_t)(((uint16_t)frame.payload[0] << 8) | frame.payload[1]);
        size_t reason_len = frame.payload_len - 2;
        if (reason_len > sizeof(reason_buf) - 1) reason_len = sizeof(reason_buf) - 1;
        if (reason_len > 0) memcpy(reason_buf, frame.payload + 2, reason_len);
        reason_buf[reason_len] = '\0';
        reason = reason_buf;
      }
      size_t out_len = 0;
      uint8_t *out = ant_ws_encode_frame(ANT_WS_OPCODE_CLOSE, frame.payload, frame.payload_len, false, false, &out_len);
      if (out) ant_conn_write(conn, (char *)out, out_len, NULL, NULL);
      websocket_emit_close(js, ws, code, reason, true);
      ant_conn_shutdown(conn);
      goto l_done;
    }

    l_protocol_error:
      websocket_emit_simple(js, ws, WS_HANDLER_ERROR);
      ant_ws_frame_clear(&frame);
      ant_conn_close(conn);
      return;

    l_too_big:
      websocket_fragment_clear(ws);
      ant_ws_frame_clear(&frame);
      websocket_server_close_with_code(ws, conn, 1009, "Message Too Big");
      return;

    l_done:
    ant_ws_frame_clear(&frame);
  }
}

void ant_websocket_server_on_close(ant_t *js, ant_value_t socket_obj) {
  websocket_state_t *ws = websocket_data(socket_obj);
  if (!ws) return;
  ws->server_conn = NULL;
  websocket_emit_close(js, ws, 1000, "", true);
  Ant_List_Remove(ws);
}

void init_websocket_module(ant_t *js) {
  ant_value_t event_proto = js->builtins.event_proto;
  ant_value_t eventtarget_proto = js->builtins.eventtarget_proto;

  js->builtins.websocket_proto = Ant_Object_CreateBare(js);
  if (is_object_type(eventtarget_proto)) js_set_proto_init(js->builtins.websocket_proto, eventtarget_proto);
  js_set(js, js->builtins.websocket_proto, "send", Ant_Function_Create(js_websocket_send));
  js_set(js, js->builtins.websocket_proto, "close", Ant_Function_Create(js_websocket_close));
  js_set(js, js->builtins.websocket_proto, "on", Ant_Function_Create(js_websocket_on));
  js_set(js, js->builtins.websocket_proto, "once", Ant_Function_Create(js_websocket_once));
  js_set(js, js->builtins.websocket_proto, "off", Ant_Function_Create(js_websocket_off));
  
  js_set_accessor_desc(
    js, js->builtins.websocket_proto, "binaryType", 10,
    Ant_Function_Create(js_websocket_get_binary_type), Ant_Function_Create(js_websocket_set_binary_type),
    JS_DESC_E | JS_DESC_C
  );

  ANT_EVENT_HANDLER_DEFINE(js->builtins.websocket_proto, js_websocket, open);
  ANT_EVENT_HANDLER_DEFINE(js->builtins.websocket_proto, js_websocket, message);
  ANT_EVENT_HANDLER_DEFINE(js->builtins.websocket_proto, js_websocket, close);
  ANT_EVENT_HANDLER_DEFINE(js->builtins.websocket_proto, js_websocket, error);
  
  js_set(js, js->builtins.websocket_proto, "CONNECTING", js_mknum(WS_CONNECTING));
  js_set(js, js->builtins.websocket_proto, "OPEN", js_mknum(WS_OPEN));
  js_set(js, js->builtins.websocket_proto, "CLOSING", js_mknum(WS_CLOSING));
  js_set(js, js->builtins.websocket_proto, "CLOSED", js_mknum(WS_CLOSED));
  js_set_tostringtag(js, js->builtins.websocket_proto, js_mkstr(js, "WebSocket", 9));
  js->builtins.websocket_ctor = Ant_Function_CreateIntrinsicConstructor(js, js_websocket_ctor, js->builtins.websocket_proto, "WebSocket", 1);
  js_set(js, js->builtins.websocket_ctor, "CONNECTING", js_mknum(WS_CONNECTING));
  js_set(js, js->builtins.websocket_ctor, "OPEN", js_mknum(WS_OPEN));
  js_set(js, js->builtins.websocket_ctor, "CLOSING", js_mknum(WS_CLOSING));
  js_set(js, js->builtins.websocket_ctor, "CLOSED", js_mknum(WS_CLOSED));
  js_set_global_builtin(js, "WebSocket", js->builtins.websocket_ctor);

  js->builtins.message_event_proto = Ant_Object_CreateBare(js);
  if (is_object_type(event_proto)) js_set_proto_init(js->builtins.message_event_proto, event_proto);
  js_set_tostringtag(js, js->builtins.message_event_proto, js_mkstr(js, "MessageEvent", 12));
  Ant_Event_DefineFields(js, js->builtins.message_event_proto, ANT_EVENT_MESSAGE);
  ant_value_t message_ctor = Ant_Function_CreateIntrinsicConstructor(js, js_message_event_ctor, js->builtins.message_event_proto, "MessageEvent", 1);
  js_set_global_builtin(js, "MessageEvent", message_ctor);

  js->builtins.close_event_proto = Ant_Object_CreateBare(js);
  if (is_object_type(event_proto)) js_set_proto_init(js->builtins.close_event_proto, event_proto);
  js_set_tostringtag(js, js->builtins.close_event_proto, js_mkstr(js, "CloseEvent", 10));
  Ant_Event_DefineFields(js, js->builtins.close_event_proto, ANT_EVENT_CLOSE);
  ant_value_t close_ctor = Ant_Function_CreateIntrinsicConstructor(js, js_close_event_ctor, js->builtins.close_event_proto, "CloseEvent", 1);
  js_set_global_builtin(js, "CloseEvent", close_ctor);
}

void cleanup_websocket_module(ant_t *js) {
  for (websocket_state_t *ws = js->active_websockets, *next; ws; ws = next) {
    next = ws->next;
    if (!ws->is_client) continue;
    ws->close_emitted = true;
    websocket_close_client(ws, 1001, false);
  }
}

void gc_mark_websocket(ant_t *js, gc_mark_fn mark) {
  for (websocket_state_t *ws = js->active_websockets; ws; ws = ws->next) {
    mark(js, ws->obj);
    if (is_object_type(ws->request_obj)) mark(js, ws->request_obj);
  }
}