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

#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <signal.h>
#include <uv.h>

#include "ant.h"
#include "internal.h"
#include "ptr.h"

#include "gc/modules.h"
#include "net/listener.h"
#include "streams/readable.h"

#include "http/http1_parser.h"
#include "http/http1_writer.h"

#include "modules/assert.h"
#include "modules/abort.h"
#include "modules/blob.h"
#include "modules/buffer.h"
#include "modules/headers.h"
#include "modules/request.h"
#include "modules/response.h"
#include "modules/server.h"
#include "modules/timer.h"
#include "modules/domexception.h"

typedef struct server_runtime_s    server_runtime_t;
typedef struct server_request_s    server_request_t;
typedef struct server_conn_state_s server_conn_state_t;

static server_runtime_t *g_server = NULL;

typedef struct stop_waiter_s {
  ant_value_t promise;
  struct stop_waiter_s *next;
} stop_waiter_t;

typedef enum {
  SERVER_WRITE_NONE = 0,
  SERVER_WRITE_CLOSE_CLIENT,
  SERVER_WRITE_STREAM_READ,
  SERVER_WRITE_KEEP_ALIVE,
} server_write_action_t;

typedef struct {
  server_request_t *request;
  ant_conn_t *conn;
  server_write_action_t action;
} server_write_req_t;

struct server_request_s {
  server_runtime_t *server;
  server_conn_state_t *conn_state;
  ant_conn_t *conn;
  
  ant_value_t request_obj;
  ant_value_t response_obj;
  ant_value_t response_promise;
  ant_value_t response_reader;
  ant_value_t response_read_promise;
  
  struct server_request_s *next;
  size_t consumed_len;
  
  int refs;
  bool keep_alive;
  bool response_started;
};

struct server_conn_state_s {
  server_runtime_t *server;
  ant_conn_t *conn;
  
  ant_http1_conn_parser_t parser;
  uv_timer_t drain_timer;
  server_request_t request;
  server_request_t *active_req;
  
  bool drain_timer_closed;
  bool drain_scheduled;
};

struct server_runtime_s {
  ant_t *js;
  
  ant_value_t export_obj;
  ant_value_t fetch_fn;
  ant_value_t server_ctx;
  uv_loop_t *loop;
  
  ant_listener_t listener;
  uv_signal_t sigint_handle;
  uv_signal_t sigterm_handle;
  stop_waiter_t *stop_waiters;
  server_request_t *requests;
  
  char *hostname;
  char *unix_path;
  
  uint64_t request_timeout_ms;
  uint64_t idle_timeout_ms;
  
  int port;
  bool sigint_closed;
  bool sigterm_closed;
  bool stopping;
  bool force_stop;
};

static inline void server_request_retain(server_request_t *req) {
  if (req) req->refs++;
}

static inline server_request_t *server_current_request(ant_t *js) {
  return (server_request_t *)js_get_native_ptr(js->current_func);
}

static inline server_runtime_t *server_current_runtime(ant_t *js) {
  return (server_runtime_t *)js_get_native_ptr(js->current_func);
}

static ant_value_t server_mkreqfun(
  ant_t *js,
  ant_value_t (*fn)(ant_t *, ant_value_t *, int),
  server_request_t *req
) {
  ant_value_t func = js_heavy_mkfun(js, fn, js_mkundef());
  js_set_native_ptr(func, req);
  return func;
}

static ant_value_t server_mkruntimefun(
  ant_t *js,
  ant_value_t (*fn)(ant_t *, ant_value_t *, int),
  server_runtime_t *server
) {
  ant_value_t func = js_heavy_mkfun(js, fn, js_mkundef());
  js_set_native_ptr(func, server);
  return func;
}

static ant_value_t server_exception_reason(ant_t *js, ant_value_t value) {
  if (!is_err(value)) return value;
  if (!js->thrown_exists) return value;

  value = js->thrown_value;
  js->thrown_exists = false;
  js->thrown_value = js_mkundef();
  js->thrown_stack = js_mkundef();
  return value;
}

static void server_remove_request(server_runtime_t *server, server_request_t *req) {
  server_request_t **it = NULL;
  if (!server || !req) return;

  for (it = &server->requests; *it; it = &(*it)->next) {
  if (*it == req) {
    *it = req->next;
    return;
  }}
}

static void server_request_reset(server_request_t *req) {
  server_runtime_t *server = NULL;
  server_conn_state_t *conn_state = NULL;

  if (!req) return;
  server = req->server;
  conn_state = req->conn_state;
  
  *req = (server_request_t){
    .server = server,
    .conn_state = conn_state,
    .conn = conn_state ? conn_state->conn : NULL,
    .request_obj = js_mkundef(),
    .response_obj = js_mkundef(),
    .response_promise = js_mkundef(),
    .response_reader = js_mkundef(),
    .response_read_promise = js_mkundef(),
  };
}

static void server_conn_state_maybe_free(server_conn_state_t *cs) {
  if (!cs) return;
  if (cs->conn) return;
  if (cs->active_req) return;
  if (cs->request.refs > 0) return;
  if (!cs->drain_timer_closed) return;
  free(cs);
}

static void server_request_release(server_request_t *req) {
  if (!req) return;
  if (--req->refs > 0) return;

  server_remove_request(req->server, req);
  server_request_reset(req);
  server_conn_state_maybe_free(req->conn_state);
}

static server_request_t *server_find_request(server_runtime_t *server, ant_value_t request_obj) {
  server_request_t *req = NULL;
  if (!server || !is_object_type(request_obj)) return NULL;

  for (req = server->requests; req; req = req->next) {
    if (req->request_obj == request_obj) return req;
  }
  return NULL;
}
static void stop_waiters_resolve(server_runtime_t *server) {
  stop_waiter_t *waiter = server->stop_waiters;
  server->stop_waiters = NULL;

  while (waiter) {
    stop_waiter_t *next = waiter->next;
    js_resolve_promise(server->js, waiter->promise, js_mkundef());
    free(waiter);
    waiter = next;
  }
}

static void server_maybe_finish_stop(server_runtime_t *server) {
  if (!server || !server->stopping) return;
  if (ant_listener_has_connections(&server->listener)) return;
  if (!ant_listener_is_closed(&server->listener) || !server->sigint_closed || !server->sigterm_closed) return;

  stop_waiters_resolve(server);
}

static void server_signal_close_cb(uv_handle_t *handle) {
  server_runtime_t *server = (server_runtime_t *)handle->data;
  if (!server) return;

  if (handle == (uv_handle_t *)&server->sigint_handle) server->sigint_closed = true;
  if (handle == (uv_handle_t *)&server->sigterm_handle) server->sigterm_closed = true;
  server_maybe_finish_stop(server);
}

static void server_begin_stop(server_runtime_t *server, bool force) {
  if (!server) return;
  if (force) server->force_stop = true;
  
  server->stopping = true;
  ant_listener_stop(&server->listener, server->force_stop);

  if (!uv_is_closing((uv_handle_t *)&server->sigint_handle))
    uv_close((uv_handle_t *)&server->sigint_handle, server_signal_close_cb);
  else server->sigint_closed = true;

  if (!uv_is_closing((uv_handle_t *)&server->sigterm_handle))
    uv_close((uv_handle_t *)&server->sigterm_handle, server_signal_close_cb);
  else server->sigterm_closed = true;

  server_maybe_finish_stop(server);
}

static void server_signal_cb(uv_signal_t *handle, int signum) {
  server_runtime_t *server = (server_runtime_t *)handle->data;
  server_begin_stop(server, false);
}

static ant_value_t server_headers_from_parsed(ant_t *js, const ant_http1_parsed_request_t *parsed) {
  ant_value_t headers = headers_create_empty(js);
  const ant_http_header_t *hdr = NULL;

  if (is_err(headers)) return headers;
  for (hdr = parsed->headers; hdr; hdr = hdr->next) {
    ant_value_t step = headers_append_literal(js, headers, hdr->name, hdr->value);
    if (is_err(step)) return step;
  }
  return headers;
}

static ant_value_t server_call_fetch(server_runtime_t *server, ant_value_t request_obj) {
  ant_t *js = server->js;
  ant_value_t args[2] = { request_obj, server->server_ctx };
  ant_value_t saved_this = js->this_val;
  ant_value_t result = js_mkundef();

  js->this_val = server->export_obj;
  if (vtype(server->fetch_fn) == T_CFUNC) result = js_as_cfunc(server->fetch_fn)(js, args, 2);
  else result = sv_vm_call(js->vm, js, server->fetch_fn, server->export_obj, args, 2, NULL, false);
  js->this_val = saved_this;
  
  return result;
}

static ant_value_t server_abort_signal_task(ant_t *js, ant_value_t *args, int nargs) {
  ant_value_t payload = js_get_slot(js->current_func, SLOT_DATA);
  ant_value_t signal = 0; ant_value_t reason = 0;
  if (vtype(payload) != T_OBJ) return js_mkundef();

  signal = js_get(js, payload, "signal");
  reason = js_get(js, payload, "reason");

  if (abort_signal_is_signal(signal) && !abort_signal_is_aborted(signal))
    signal_do_abort(js, signal, reason);

  return js_mkundef();
}

static void server_queue_abort_signal(ant_t *js, ant_value_t signal, ant_value_t reason) {
  ant_value_t callback = 0;
  ant_value_t payload = 0;

  if (!abort_signal_is_signal(signal) || abort_signal_is_aborted(signal)) return;
  payload = js_mkobj(js);
  if (is_err(payload)) return;

  js_set(js, payload, "signal", signal);
  js_set(js, payload, "reason", reason);
  callback = js_heavy_mkfun(js, server_abort_signal_task, payload);
  
  queue_microtask(js, callback);
}

static void server_abort_request(server_request_t *req, const char *message) {
  ant_t *js = NULL;
  
  ant_value_t signal = 0; ant_value_t reason = 0; ant_value_t abort_reason = 0;
  if (!req || !req->server || !is_object_type(req->request_obj)) return;

  js = req->server->js;
  abort_reason = js_get_slot(req->request_obj, SLOT_REQUEST_ABORT_REASON);
  if (vtype(abort_reason) != T_UNDEF) return;

  reason = make_dom_exception(js, 
    message ? message : "The request was aborted", "AbortError"
  );
  
  js_set_slot_wb(js, req->request_obj, SLOT_REQUEST_ABORT_REASON, reason);
  signal = js_get_slot(req->request_obj, SLOT_REQUEST_SIGNAL);
  
  if (!abort_signal_is_signal(signal) || abort_signal_is_aborted(signal)) return;
  server_queue_abort_signal(js, signal, reason);
}

static bool server_response_chunk(server_request_t *req, ant_value_t value, const uint8_t **out, size_t *len) {
  blob_data_t *blob = NULL;

  if (vtype(value) == T_STR) {
    *out = (const uint8_t *)js_getstr(req->server->js, value, len);
    if (!*out) *len = 0;
    return true;
  }

  if (is_object_type(value) && blob_is_blob(req->server->js, value)) {
    blob = blob_get_data(value);
    *out = blob ? blob->data : NULL;
    *len = blob ? blob->size : 0;
    return true;
  }

  return buffer_source_get_bytes(req->server->js, value, out, len);
}

static void server_cancel_response_body(server_request_t *req, const char *message) {
  ant_t *js = NULL;
  
  ant_value_t stream = 0;
  ant_value_t reason = 0;
  ant_value_t result = 0;

  if (!req || !req->server || !is_object_type(req->response_obj)) return;

  js = req->server->js;
  stream = js_get_slot(req->response_obj, SLOT_RESPONSE_BODY_STREAM);
  if (!rs_is_stream(stream)) return;

  reason = make_dom_exception(
    js, message ? message : "The response body was canceled",
    "AbortError"
  );
  
  result = readable_stream_cancel(js, stream, reason);
  if (vtype(result) == T_PROMISE) promise_mark_handled(result);
}

static void server_start_stream_read(server_request_t *req);
static void server_on_read(ant_conn_t *conn, ssize_t nread, void *user_data);
static void server_write_cb(ant_conn_t *conn, int status, void *user_data);

static void server_on_deferred_drain(uv_timer_t *handle) {
  server_conn_state_t *cs = handle ? (server_conn_state_t *)handle->data : NULL;

  if (!cs) return;
  cs->drain_scheduled = false;

  if (!cs->conn || cs->active_req) return;
  if (ant_conn_is_closing(cs->conn)) return;
  if (ant_conn_buffer_len(cs->conn) == 0) return;

  server_on_read(cs->conn, (ssize_t)ant_conn_buffer_len(cs->conn), cs->server);
}

static void server_on_drain_timer_close(uv_handle_t *handle) {
  server_conn_state_t *cs = handle ? (server_conn_state_t *)handle->data : NULL;

  if (!cs) return;
  cs->drain_timer_closed = true;
  cs->drain_scheduled = false;
  server_conn_state_maybe_free(cs);
}

static void server_schedule_drain(server_conn_state_t *cs) {
  if (!cs || !cs->conn) return;
  if (cs->drain_scheduled) return;

  cs->drain_scheduled = true;
  if (uv_timer_start(&cs->drain_timer, server_on_deferred_drain, 0, 0) != 0)
    cs->drain_scheduled = false;
}

static bool server_queue_write(ant_conn_t *conn, server_request_t *req, char *data, size_t len, server_write_action_t action) {
  server_write_req_t *wr = calloc(1, sizeof(*wr));
  int rc = 0;

  if (!conn || ant_conn_is_closing(conn)) {
    free(data);
    return false;
  }

  if (!wr) {
    free(data);
    return false;
  }

  wr->request = req;
  wr->conn = conn;
  wr->action = action;
  if (req) server_request_retain(req);

  rc = ant_conn_write(conn, data, len, server_write_cb, wr);
  if (rc != 0) {
    if (req) server_request_release(req);
    free(wr);
    ant_conn_close(conn);
    return false;
  }

  return true;
}

static bool server_queue_final_chunk(server_request_t *req, server_write_action_t action) {
  ant_http1_buffer_t buf;
  char *out = NULL;
  size_t out_len = 0;

  if (!req || !req->conn || ant_conn_is_closing(req->conn)) return false;

  ant_http1_buffer_init(&buf);
  if (!ant_http1_write_final_chunk(&buf)) {
    ant_http1_buffer_free(&buf);
    ant_conn_close(req->conn);
    return false;
  }

  out = ant_http1_buffer_take(&buf, &out_len);
  return server_queue_write(req->conn, req, out, out_len, action);
}

static void server_send_basic_response(
  ant_conn_t *conn, int status, const char *status_text, 
  const char *content_type, const uint8_t *body, size_t body_len
) {
  ant_http1_buffer_t buf;
  char *out = NULL;
  size_t out_len = 0;

  ant_http1_buffer_init(&buf);
  if (!ant_http1_write_basic_response(&buf, status, status_text, content_type, body, body_len, false)) {
    ant_http1_buffer_free(&buf);
    ant_conn_close(conn);
    return;
  }

  out = ant_http1_buffer_take(&buf, &out_len);
  server_queue_write(conn, NULL, out, out_len, SERVER_WRITE_CLOSE_CLIENT);
}

static inline void server_send_text_response(
  ant_conn_t *conn,
  int status,
  const char *status_text,
  const char *body
) {
  if (!body) body = "";
  server_send_basic_response(
    conn, status,
    status_text,
    "text/plain;charset=UTF-8",
    (const uint8_t *)body,
    strlen(body)
  );
}

static inline void server_send_internal_error(ant_conn_t *conn, const char *body) {
  server_send_text_response(
    conn, 500,
    "Internal Server Error",
    body ? body : "Internal Server Error"
  );
}

static void server_finish_with_response(server_request_t *req, ant_value_t response_obj) {
  response_data_t *resp = response_get_data(response_obj);
  ant_value_t headers = response_get_headers(response_obj);
  
  ant_value_t stream = js_get_slot(response_obj, SLOT_RESPONSE_BODY_STREAM);
  bool body_is_stream = resp && resp->body_is_stream && rs_is_stream(stream);
  bool head_only = false;
  
  const char *status_text = NULL;
  ant_http1_buffer_t buf;
  
  char *out = NULL;
  size_t out_len = 0;

  if (!req->conn || ant_conn_is_closing(req->conn)) return;
  if (!resp) {
    server_send_internal_error(req->conn, "Invalid Response");
    return;
  }

  req->response_obj = response_obj;
  req->response_started = true;
  head_only = strcasecmp(request_get_data(req->request_obj)->method, "HEAD") == 0;
  status_text = (resp->status_text && resp->status_text[0]) ? resp->status_text : ant_http1_default_status_text(resp->status);

  ant_http1_buffer_init(&buf);
  if (!ant_http1_write_response_head(&buf, resp->status, status_text, headers, body_is_stream, resp->body_size, req->keep_alive)) {
    ant_http1_buffer_free(&buf);
    ant_conn_close(req->conn);
    return;
  }
  
  if (!body_is_stream && !head_only && resp->body_size > 0) 
    ant_http1_buffer_append(&buf, resp->body_data, resp->body_size);
  
  if (buf.failed) {
    ant_http1_buffer_free(&buf);
    ant_conn_close(req->conn);
    return;
  }

  out = ant_http1_buffer_take(&buf, &out_len);
  ant_conn_set_timeout_ms(req->conn, req->server->idle_timeout_ms);
  
  if (body_is_stream && head_only) server_cancel_response_body(
    req, "The response body was canceled for a HEAD request"
  );
  
  server_queue_write(
    req->conn,
    req, out, out_len,
    body_is_stream && !head_only
      ? SERVER_WRITE_STREAM_READ
      : (req->keep_alive ? SERVER_WRITE_KEEP_ALIVE : SERVER_WRITE_CLOSE_CLIENT)
  );
}

static ant_value_t server_on_response_reject(ant_t *js, ant_value_t *args, int nargs) {
  server_request_t *req = server_current_request(js);
  ant_value_t reason = (nargs > 0) ? args[0] : js_mkundef();
  const char *msg = NULL;

  if (!req) return js_mkundef();
  req->response_promise = js_mkundef();

  if (req->conn && !ant_conn_is_closing(req->conn)) {
    msg = js_str(js, reason);
    server_send_internal_error(req->conn, msg);
  }

  server_request_release(req);
  return js_mkundef();
}

static ant_value_t server_on_response_fulfill(ant_t *js, ant_value_t *args, int nargs) {
  server_request_t *req = server_current_request(js);
  ant_value_t value = (nargs > 0) ? args[0] : js_mkundef();

  if (!req) return js_mkundef();
  req->response_promise = js_mkundef();

  if (!response_get_data(value)) {
    if (req->conn && !ant_conn_is_closing(req->conn))
      server_send_internal_error(req->conn, "fetch handler must return a Response");
  } else if (req->conn && !ant_conn_is_closing(req->conn)) {
    server_finish_with_response(req, value);
  }

  server_request_release(req);
  return js_mkundef();
}

static void server_handle_fetch_result(server_request_t *req, ant_value_t result) {
  ant_t *js = req->server->js;

  if (is_err(result)) {
    ant_value_t reason = server_exception_reason(js, result);
    const char *msg = js_str(js, reason);
    server_send_internal_error(req->conn, msg);
    return;
  }

  if (vtype(result) == T_PROMISE) {
    ant_value_t fulfill = server_mkreqfun(js, server_on_response_fulfill, req);
    ant_value_t reject = server_mkreqfun(js, server_on_response_reject, req);
    ant_value_t then_result = 0;

    req->response_promise = result;
    server_request_retain(req);
    then_result = js_promise_then(js, result, fulfill, reject);
    promise_mark_handled(then_result);
    return;
  }

  if (!response_get_data(result)) {
    server_send_internal_error(req->conn, "fetch handler must return a Response");
    return;
  }

  server_finish_with_response(req, result);
}

static ant_value_t server_stream_read_reject(ant_t *js, ant_value_t *args, int nargs) {
  server_request_t *req = server_current_request(js);
  if (!req) return js_mkundef();
  
  req->response_read_promise = js_mkundef();
  server_queue_final_chunk(req, SERVER_WRITE_CLOSE_CLIENT);
  server_request_release(req);
  
  return js_mkundef();
}

static ant_value_t server_stream_read_fulfill(ant_t *js, ant_value_t *args, int nargs) {
  server_request_t *req = server_current_request(js);
  ant_value_t result = (nargs > 0) ? args[0] : js_mkundef();
  
  ant_value_t done = 0;
  ant_value_t value = 0;
  
  const uint8_t *chunk = NULL;
  size_t chunk_len = 0;
  
  ant_http1_buffer_t buf;
  char *out = NULL;
  size_t out_len = 0;

  if (!req) return js_mkundef();
  req->response_read_promise = js_mkundef();

  if (!req->conn || ant_conn_is_closing(req->conn)) {
    server_request_release(req);
    return js_mkundef();
  }

  done = js_get(js, result, "done");
  if (done == js_true) {
    ant_conn_set_timeout_ms(req->conn, req->server->idle_timeout_ms);
    server_queue_final_chunk(
      req, req->keep_alive 
        ? SERVER_WRITE_KEEP_ALIVE 
        : SERVER_WRITE_CLOSE_CLIENT
    );
    
    server_request_release(req);
    return js_mkundef();
  }

  value = js_get(js, result, "value");
  if (!server_response_chunk(req, value, &chunk, &chunk_len)) {
    fprintf(stderr, "Response body stream chunk must be a string, Blob, ArrayBuffer, DataView, or TypedArray\n");
    server_cancel_response_body(req, "Invalid response body chunk");
    server_queue_final_chunk(req, SERVER_WRITE_CLOSE_CLIENT);
    server_request_release(req);
    return js_mkundef();
  }

  ant_http1_buffer_init(&buf);
  if (!ant_http1_write_chunk(&buf, chunk, chunk_len)) {
    ant_http1_buffer_free(&buf);
    ant_conn_close(req->conn);
    server_request_release(req);
    return js_mkundef();
  }

  out = ant_http1_buffer_take(&buf, &out_len);
  ant_conn_set_timeout_ms(req->conn, req->server->idle_timeout_ms);
  
  server_queue_write(req->conn, req, out, out_len, SERVER_WRITE_STREAM_READ);
  server_request_release(req);
  
  return js_mkundef();
}

static void server_start_stream_read(server_request_t *req) {
  ant_t *js = req->server->js;
  
  ant_value_t next_p = 0;
  ant_value_t fulfill = 0;
  ant_value_t reject = 0;
  ant_value_t then_result = 0;

  if (!req->conn || ant_conn_is_closing(req->conn)) return;
  if (!is_object_type(req->response_reader)) return;

  next_p = rs_default_reader_read(js, req->response_reader);
  req->response_read_promise = next_p;
  fulfill = server_mkreqfun(js, server_stream_read_fulfill, req);
  reject = server_mkreqfun(js, server_stream_read_reject, req);

  server_request_retain(req);
  then_result = js_promise_then(js, next_p, fulfill, reject);
  promise_mark_handled(then_result);
}

static bool server_request_ensure_reader(server_request_t *req) {
  ant_t *js;
  ant_value_t reader_args[1];
  ant_value_t saved;

  if (!req || !is_object_type(req->response_obj)) return false;
  if (vtype(req->response_reader) != T_UNDEF) return true;

  js = req->server->js;
  reader_args[0] = js_get_slot(req->response_obj, SLOT_RESPONSE_BODY_STREAM);

  saved = js->new_target;
  js->new_target = g_reader_proto;
  req->response_reader = js_rs_reader_ctor(js, reader_args, 1);
  js->new_target = saved;

  return !is_err(req->response_reader);
}


static void server_write_cb(ant_conn_t *conn, int status, void *user_data) {
  server_write_req_t *wr = (server_write_req_t *)user_data;
  server_request_t *req = wr->request;
  server_conn_state_t *cs = conn ? (server_conn_state_t *)ant_conn_get_user_data(conn) : NULL;

  if (status < 0 && conn && !ant_conn_is_closing(conn))
    ant_conn_close(conn);

  if (status == 0 && conn && !ant_conn_is_closing(conn)) {
  switch (wr->action) {
  case SERVER_WRITE_STREAM_READ:
    if (!server_request_ensure_reader(req)) {
      ant_conn_close(conn);
      break;
    }
    server_start_stream_read(req);
    break;

  case SERVER_WRITE_KEEP_ALIVE:
    if (cs && req) {
      ant_conn_consume(conn, req->consumed_len);
      ant_http1_conn_parser_reset(&cs->parser);
      cs->active_req = NULL;
      ant_conn_set_timeout_ms(conn, cs->server->idle_timeout_ms);
      ant_conn_resume_read(conn);
      server_request_release(req);
      if (ant_conn_buffer_len(conn) > 0) server_schedule_drain(cs);
    }
    break;
    
  case SERVER_WRITE_CLOSE_CLIENT:
    ant_conn_close(conn);
    break;
    
  default: break;
  }}

  if (req) server_request_release(req);
  free(wr);
}

static ant_value_t server_request_ip(ant_t *js, ant_value_t *args, int nargs) {
  server_runtime_t *server = server_current_runtime(js);
  server_request_t *req = NULL;
  ant_value_t out = 0;

  if (!server || nargs < 1) return js_mknull();
  req = server_find_request(server, args[0]);
  if (!req || !req->conn || !ant_conn_has_remote_addr(req->conn)) return js_mknull();

  out = js_mkobj(js);
  js_set(js, out, "address", js_mkstr(js, ant_conn_remote_addr(req->conn), strlen(ant_conn_remote_addr(req->conn))));
  js_set(js, out, "port", js_mknum(ant_conn_remote_port(req->conn)));
  return out;
}

static ant_value_t server_timeout(ant_t *js, ant_value_t *args, int nargs) {
  server_runtime_t *server = server_current_runtime(js);
  server_request_t *req = NULL;
  int timeout = 0;

  if (!server || nargs < 2) return js_mkundef();
  req = server_find_request(server, args[0]);
  if (!req || !req->conn) return js_mkundef();

  timeout = (int)js_getnum(args[1]);
  ant_conn_set_timeout_ms(req->conn, (uint64_t)timeout * 1000ULL);
  
  return js_mkundef();
}

static ant_value_t server_stop(ant_t *js, ant_value_t *args, int nargs) {
  server_runtime_t *server = server_current_runtime(js);
  stop_waiter_t *waiter = NULL;
  ant_value_t promise = js_mkpromise(js);
  bool force = (nargs > 0 && js_truthy(js, args[0]));

  if (!server) {
    js_resolve_promise(js, promise, js_mkundef());
    return promise;
  }

  waiter = calloc(1, sizeof(*waiter));
  if (!waiter) {
    js_reject_promise(js, promise, js_mkerr(js, "out of memory"));
    return promise;
  }

  waiter->promise = promise;
  waiter->next = server->stop_waiters;
  server->stop_waiters = waiter;
  server_begin_stop(server, force);
  return promise;
}

static void server_process_client_request(
  ant_conn_t *conn,
  ant_http1_parsed_request_t *parsed,
  size_t consumed_len
) {
  server_conn_state_t *cs = (server_conn_state_t *)ant_conn_get_user_data(conn);
  server_runtime_t *server = cs ? cs->server : NULL;
  server_request_t *req = NULL;
  
  ant_t *js = NULL;
  ant_value_t headers = 0;
  ant_value_t request_obj = 0;
  ant_value_t result = 0;
  bool keep_alive = false;

  if (!server || !cs) {
    ant_http1_free_parsed_request(parsed);
    ant_conn_close(conn);
    return;
  }

  js = server->js;
  req = &cs->request;
  keep_alive = parsed->keep_alive;
  headers = server_headers_from_parsed(js, parsed);
  
  if (is_err(headers)) {
    ant_http1_free_parsed_request(parsed);
    server_send_internal_error(conn, NULL);
    return;
  }

  request_obj = request_create_server(
    js,
    parsed->method,
    parsed->target,
    parsed->absolute_target,
    parsed->host,
    server->hostname,
    server->port,
    headers,
    parsed->body,
    parsed->body_len,
    parsed->content_type
  );
  ant_http1_free_parsed_request(parsed);

  if (is_err(request_obj)) {
    server_send_internal_error(conn, NULL);
    return;
  }

  server_request_reset(req);
  req->server = server;
  req->conn_state = cs;
  req->conn = conn;
  req->request_obj = request_obj;
  req->consumed_len = consumed_len;
  req->keep_alive = keep_alive;
  req->refs = 1;
  req->next = server->requests;
  server->requests = req;
  cs->active_req = req;
  ant_conn_pause_read(conn);
  ant_conn_set_timeout_ms(conn, server->request_timeout_ms);

  result = server_call_fetch(server, request_obj);
  server_handle_fetch_result(req, result);
}

static void server_on_read(ant_conn_t *conn, ssize_t nread, void *user_data) {
  server_conn_state_t *cs = (server_conn_state_t *)ant_conn_get_user_data(conn);
  ant_http1_parsed_request_t parsed = {0};
  ant_http1_parse_result_t parse_result = ANT_HTTP1_PARSE_INCOMPLETE;
  size_t consumed = 0;

  if (!conn || !cs) return;
  if (cs->active_req) return;
  if (ant_conn_buffer_len(conn) == 0) return;

  ant_conn_set_timeout_ms(conn, cs->server->request_timeout_ms);
  parse_result = ant_http1_conn_parser_execute(
    &cs->parser,
    ant_conn_buffer(conn),
    ant_conn_buffer_len(conn),
    &parsed,
    &consumed
  );
  
  if (parse_result == ANT_HTTP1_PARSE_ERROR) {
    ant_http1_free_parsed_request(&parsed);
    server_send_text_response(conn, 400, "Bad Request", "Bad Request");
    return;
  }

  if (parse_result != ANT_HTTP1_PARSE_OK) return;
  server_process_client_request(conn, &parsed, consumed);
}

static void server_on_end(ant_conn_t *conn, void *user_data) {
  if (conn) ant_conn_close(conn);
}

static void server_on_conn_close(ant_conn_t *conn, void *user_data) {
  server_runtime_t *server = (server_runtime_t *)user_data;
  server_conn_state_t *cs = (server_conn_state_t *)ant_conn_get_user_data(conn);

  if (cs) {
    ant_conn_set_user_data(conn, NULL);
    cs->conn = NULL;
    ant_http1_conn_parser_free(&cs->parser);
    if (!uv_is_closing((uv_handle_t *)&cs->drain_timer))
      uv_close((uv_handle_t *)&cs->drain_timer, server_on_drain_timer_close);
    if (cs->active_req) {
      server_abort_request(cs->active_req, "Client disconnected");
      server_cancel_response_body(cs->active_req, "Client disconnected");
      cs->active_req->conn = NULL;
      cs->active_req = NULL;
      server_request_release(&cs->request);
      cs = NULL;
    }
    if (cs) server_conn_state_maybe_free(cs);
  }

  server_maybe_finish_stop(server);
}

static void server_on_listener_close(ant_listener_t *listener, void *user_data) {
  server_maybe_finish_stop((server_runtime_t *)user_data);
}

static void server_on_accept(ant_listener_t *listener, ant_conn_t *conn, void *user_data) {
  server_runtime_t *server = (server_runtime_t *)user_data;
  server_conn_state_t *cs = NULL;

  (void)listener;
  if (!conn || !server) return;

  cs = calloc(1, sizeof(*cs));
  if (!cs) {
    ant_conn_close(conn);
    return;
  }

  cs->server = server;
  cs->conn = conn;
  cs->drain_timer_closed = true;
  cs->request.server = server;
  cs->request.conn_state = cs;
  server_request_reset(&cs->request);
  
  if (uv_timer_init(server->loop, &cs->drain_timer) != 0) {
    free(cs);
    ant_conn_close(conn);
    return;
  }
  
  cs->drain_timer.data = cs;
  cs->drain_timer_closed = false;
  ant_http1_conn_parser_init(&cs->parser);
  ant_conn_set_user_data(conn, cs);
  ant_conn_set_no_delay(conn, true);
}

static bool server_export_has_fetch_handler(ant_t *js, ant_value_t default_export, bool *looks_like_config) {
  ant_value_t fetch = 0;
  
  if (looks_like_config) *looks_like_config = false;
  if (!is_object_type(default_export)) return false;

  fetch = js_get(js, default_export, "fetch");
  if (is_callable(fetch)) return true;

  if (looks_like_config) {
    ant_value_t port = js_get(js, default_export, "port");
    ant_value_t hostname = js_get(js, default_export, "hostname");
    ant_value_t idle_timeout = js_get(js, default_export, "idleTimeout");
    ant_value_t request_timeout = js_get(js, default_export, "requestTimeout");
    ant_value_t unix_path = js_get(js, default_export, "unix");
    ant_value_t tls = js_get(js, default_export, "tls");

    *looks_like_config =
      vtype(fetch)           != T_UNDEF ||
      vtype(port)            != T_UNDEF ||
      vtype(hostname)        != T_UNDEF ||
      vtype(idle_timeout)    != T_UNDEF ||
      vtype(request_timeout) != T_UNDEF ||
      vtype(unix_path)       != T_UNDEF ||
      vtype(tls)             != T_UNDEF;
  }

  return false;
}

int server_maybe_start_from_export(ant_t *js, ant_value_t default_export) {
  bool looks_like_server = false;
  ant_value_t server_result = 0;
  const char *error = NULL;

  if (!server_export_has_fetch_handler(js, default_export, &looks_like_server)) {
    if (!looks_like_server) return EXIT_SUCCESS;
    error = "Module does not export a fetch handler";
    goto fail;
  }

  server_result = server_start_from_export(js, default_export);
  if (is_err(server_result)) {
    error = js_str(js, server_result);
    goto fail;
  }

  return EXIT_SUCCESS;

fail:
  fprintf(stderr, "%s\n", error);
  return EXIT_FAILURE;
}

ant_value_t server_start_from_export(ant_t *js, ant_value_t default_export) {
  server_runtime_t *server = NULL;
  
  ant_value_t port_v = 0;
  ant_value_t hostname_v = 0;
  ant_value_t idle_timeout_v = 0;
  ant_value_t request_timeout_v = 0;
  ant_value_t unix_v = 0;
  ant_value_t tls_v = 0;
  
  ant_listener_callbacks_t callbacks = {0};
  int rc = 0;

  if (g_server) return js_mkerr(js, "server is already running");
  if (!is_object_type(default_export)) return js_mkerr_typed(js, JS_ERR_TYPE, "Module does not export a fetch handler");

  server = malloc(sizeof(*server));
  if (!server) return js_mkerr(js, "out of memory");

  *server = (server_runtime_t){
    .js = js,
    .export_obj = default_export,
    .fetch_fn = js_get(js, default_export, "fetch"),
    .hostname = strdup("0.0.0.0"),
    .unix_path = NULL,
    .port = 3000,
    .request_timeout_ms = 30000,
    .idle_timeout_ms = 30000,
    .loop = uv_default_loop(),
  };

  if (!server->hostname) {
    free(server);
    return js_mkerr(js, "out of memory");
  }

  unix_v = js_get(js, default_export, "unix");
  tls_v = js_get(js, default_export, "tls");
  
  if (vtype(unix_v) != T_UNDEF && vtype(unix_v) != T_NULL) {
    if (vtype(unix_v) != T_STR) {
      free(server->hostname);
      free(server);
      return js_mkerr_typed(js, JS_ERR_TYPE, "server unix must be a string");
    }

    server->unix_path = strdup(js_getstr(js, unix_v, NULL));
    if (!server->unix_path) {
      free(server->hostname);
      free(server);
      return js_mkerr(js, "out of memory");
    }
  }
  
  if (vtype(tls_v) != T_UNDEF && vtype(tls_v) != T_NULL) {
    free(server->unix_path);
    free(server->hostname);
    free(server);
    return js_mkerr_typed(js, JS_ERR_TYPE, "tls server config is not implemented yet");
  }

  port_v = js_get(js, default_export, "port");
  hostname_v = js_get(js, default_export, "hostname");
  idle_timeout_v = js_get(js, default_export, "idleTimeout");
  request_timeout_v = js_get(js, default_export, "requestTimeout");

  if (vtype(port_v) != T_UNDEF && vtype(port_v) != T_NULL) {
    if (vtype(port_v) != T_NUM) {
      free(server->unix_path);
      free(server->hostname);
      free(server);
      return js_mkerr_typed(js, JS_ERR_TYPE, "server port must be a number");
    }
    server->port = (int)js_getnum(port_v);
    if (server->port < 0 || server->port > 65535) {
      free(server->unix_path);
      free(server->hostname);
      free(server);
      return js_mkerr_typed(js, JS_ERR_RANGE, "server port must be between 0 and 65535");
    }
  }

  if (vtype(hostname_v) != T_UNDEF && vtype(hostname_v) != T_NULL) {
    char *next_hostname = NULL;
    if (vtype(hostname_v) != T_STR) {
      free(server->unix_path);
      free(server->hostname);
      free(server);
      return js_mkerr_typed(js, JS_ERR_TYPE, "server hostname must be a string");
    }
    next_hostname = strdup(js_getstr(js, hostname_v, NULL));
    if (!next_hostname) {
      free(server->unix_path);
      free(server->hostname);
      free(server);
      return js_mkerr(js, "out of memory");
    }
    free(server->hostname);
    server->hostname = next_hostname;
  }

  if (vtype(idle_timeout_v) != T_UNDEF && vtype(idle_timeout_v) != T_NULL) {
    double timeout = 0;
    if (vtype(idle_timeout_v) != T_NUM) {
      free(server->unix_path);
      free(server->hostname);
      free(server);
      return js_mkerr_typed(js, JS_ERR_TYPE, "server idleTimeout must be a number");
    }
    timeout = js_getnum(idle_timeout_v);
    if (timeout < 0) {
      free(server->unix_path);
      free(server->hostname);
      free(server);
      return js_mkerr_typed(js, JS_ERR_RANGE, "server idleTimeout must be >= 0");
    }
    server->idle_timeout_ms = (uint64_t)(timeout * 1000.0);
  }

  if (vtype(request_timeout_v) != T_UNDEF && vtype(request_timeout_v) != T_NULL) {
    double timeout = 0;
    if (vtype(request_timeout_v) != T_NUM) {
      free(server->unix_path);
      free(server->hostname);
      free(server);
      return js_mkerr_typed(js, JS_ERR_TYPE, "server requestTimeout must be a number");
    }
    
    timeout = js_getnum(request_timeout_v);
    if (timeout < 0) {
      free(server->unix_path);
      free(server->hostname);
      free(server);
      return js_mkerr_typed(js, JS_ERR_RANGE, "server requestTimeout must be >= 0");
    }
    
    server->request_timeout_ms = (uint64_t)(timeout * 1000.0);
  }

  uv_signal_init(server->loop, &server->sigint_handle);
  uv_signal_init(server->loop, &server->sigterm_handle);
  server->sigint_handle.data = server;
  server->sigterm_handle.data = server;

  callbacks.on_accept = server_on_accept;
  callbacks.on_read = server_on_read;
  callbacks.on_end = server_on_end;
  callbacks.on_conn_close = server_on_conn_close;
  callbacks.on_listener_close = server_on_listener_close;

  if (server->unix_path) {
    rc = ant_listener_listen_pipe(
      &server->listener, server->loop,
      server->unix_path,
      128, server->idle_timeout_ms, &callbacks, server
    );
  } else {
    rc = ant_listener_listen_tcp(
      &server->listener, server->loop,
      server->hostname, server->port,
      128, server->idle_timeout_ms, &callbacks, server
    );
  }
  
  if (rc != 0) {
    free(server->unix_path);
    free(server->hostname);
    free(server);
    return js_mkerr_typed(js, JS_ERR_TYPE, "%s", uv_strerror(rc));
  }

  server->port = ant_listener_port(&server->listener);
  uv_signal_start(&server->sigint_handle, server_signal_cb, SIGINT);
  uv_signal_start(&server->sigterm_handle, server_signal_cb, SIGTERM);

  server->server_ctx = js_mkobj(js);
  js_set(js, server->server_ctx, "requestIP", server_mkruntimefun(js, server_request_ip, server));
  js_set(js, server->server_ctx, "timeout", server_mkruntimefun(js, server_timeout, server));
  js_set(js, server->server_ctx, "stop", server_mkruntimefun(js, server_stop, server));

  g_server = server;
  return js_mkundef();
}

void gc_mark_server(ant_t *js, gc_mark_fn mark) {
  server_request_t *req = NULL;
  stop_waiter_t *waiter = NULL;

  if (!g_server) return;
  mark(js, g_server->export_obj);
  mark(js, g_server->fetch_fn);
  mark(js, g_server->server_ctx);

  for (req = g_server->requests; req; req = req->next) {
    mark(js, req->request_obj);
    mark(js, req->response_obj);
    mark(js, req->response_promise);
    mark(js, req->response_reader);
    mark(js, req->response_read_promise);
  }

  for (waiter = g_server->stop_waiters; waiter; waiter = waiter->next)
    mark(js, waiter->promise);
}