module fasthttp

import net
import sync.stdatomic
import time

#include <sys/epoll.h>
#include <sys/sendfile.h>
#include <sys/stat.h>
#include <netinet/tcp.h>

const epoll_wait_timeout_ms = 100
const status_408_response = 'HTTP/1.1 408 Request Timeout\r\nContent-Type: text/plain\r\nContent-Length: 19\r\nConnection: close\r\n\r\n408 Request Timeout'.bytes()

fn C.accept4(sockfd i32, addr &net.Addr, addrlen &u32, flags i32) i32

fn C.epoll_create1(__flags i32) i32

fn C.epoll_ctl(__epfd i32, __op i32, __fd i32, __event &C.epoll_event) i32

fn C.epoll_wait(__epfd i32, __events &C.epoll_event, __maxevents i32, __timeout i32) i32

fn C.sendfile(out_fd i32, in_fd i32, offset &i64, count usize) i32

fn C.fstat(fd i32, buf &C.stat) i32

@[typedef]
union C.epoll_data_t {
mut:
	ptr voidptr
	fd  int
	u32 u32
	u64 u64
}

struct C.epoll_event {
mut:
	events u32
	data   C.epoll_data_t
}

pub struct Server {
pub:
	family                  net.AddrFamily = .ip6
	port                    int            = 3000
	max_request_buffer_size int            = 8192
	timeout_in_seconds      int            = 30
	user_data               voidptr
mut:
	listen_fds      []int    = []int{len: max_thread_pool_size, cap: max_thread_pool_size, init: -1}
	epoll_fds       []int    = []int{len: max_thread_pool_size, cap: max_thread_pool_size, init: -1}
	threads         []thread = []thread{len: max_thread_pool_size, cap: max_thread_pool_size}
	request_handler fn (HttpRequest) !HttpResponse @[required]
	running         &stdatomic.AtomicVal[bool] = stdatomic.new_atomic(false)
	shutting_down   &stdatomic.AtomicVal[bool] = stdatomic.new_atomic(false)
	stopped         &stdatomic.AtomicVal[bool] = stdatomic.new_atomic(true)
	active_requests &stdatomic.AtomicVal[int]  = stdatomic.new_atomic(0)
}

// new_server creates and initializes a new Server instance.
pub fn new_server(config ServerConfig) !&Server {
	if config.max_request_buffer_size <= 0 {
		return error('max_request_buffer_size must be greater than 0')
	}
	mut server := &Server{
		family:                  config.family
		port:                    config.port
		max_request_buffer_size: config.max_request_buffer_size
		timeout_in_seconds:      config.timeout_in_seconds
		user_data:               config.user_data
		request_handler:         config.handler
		running:                 stdatomic.new_atomic(false)
		shutting_down:           stdatomic.new_atomic(false)
		stopped:                 stdatomic.new_atomic(true)
		active_requests:         stdatomic.new_atomic(0)
	}
	unsafe {
		server.listen_fds.flags.set(.noslices | .noshrink | .nogrow)
		server.epoll_fds.flags.set(.noslices | .noshrink | .nogrow)
		server.threads.flags.set(.noslices | .noshrink | .nogrow)
	}
	return server
}

fn set_blocking(fd int, blocking bool) {
	flags := C.fcntl(fd, C.F_GETFL, 0)
	if flags == -1 {
		// TODO: better error handling
		eprintln(@LOCATION)
		return
	}
	if blocking {
		// This removes the O_NONBLOCK flag from flags and set it.
		C.fcntl(fd, C.F_SETFL, flags & ~C.O_NONBLOCK)
	} else {
		// This adds the O_NONBLOCK flag from flags and set it.
		C.fcntl(fd, C.F_SETFL, flags | C.O_NONBLOCK)
	}
}

// ClientWriteState carries the pending response bytes plus optional file data
// for a connection that could not be drained in a single non-blocking write.
// It lives in `client_write_states` until both the in-memory content and the
// file body have been fully transferred. While a state exists, the connection
// is also armed for EPOLLOUT so process_events can resume the write without
// blocking the worker.
struct ClientWriteState {
mut:
	content        []u8
	content_pos    int
	content_owned  bool
	file_fd        int = -1
	file_len       i64
	file_pos       i64
	should_close   bool
	request_arena  voidptr
	request_active bool
	start_ns       i64
	// epollout_armed records whether we actually had to register EPOLLOUT for
	// this connection. When set, complete_write performs a DEL+ADD on the fd to
	// re-deliver any pipelined request bytes that arrived during the write as a
	// fresh EPOLLIN edge (level-triggered semantics are not used here).
	epollout_armed bool
}

// drain_status describes the outcome of one try_drain_write() pass.
enum DrainStatus {
	done    // all bytes (content + file) have been sent
	blocked // the kernel send buffer is full; come back on EPOLLOUT
	failed  // unrecoverable error or peer closed mid-transfer
}

fn close_socket(fd int) bool {
	ret := C.close(fd)
	if ret == -1 {
		if C.errno == C.EINTR {
			// Interrupted by signal, retry is safe
			return close_socket(fd)
		}
		eprintln('ERROR: close(fd=${fd}) failed with errno=${C.errno}')
		return false
	}
	return true
}

fn create_server_socket(server Server) int {
	// Create a socket with non-blocking mode
	server_fd := C.socket(i32(server.family), i32(net.SocketType.tcp), 0)
	if server_fd < 0 {
		eprintln(@LOCATION)
		C.perror(c'Socket creation failed')
		return -1
	}

	set_blocking(server_fd, false)

	// Enable SO_REUSEADDR and SO_REUSEPORT
	opt := 1
	if C.setsockopt(server_fd, C.SOL_SOCKET, C.SO_REUSEADDR, &opt, sizeof(opt)) < 0 {
		eprintln(@LOCATION)
		C.perror(c'setsockopt SO_REUSEADDR failed')
		close_socket(server_fd)
		return -1
	}
	if C.setsockopt(server_fd, C.SOL_SOCKET, C.SO_REUSEPORT, &opt, sizeof(opt)) < 0 {
		eprintln(@LOCATION)
		C.perror(c'setsockopt SO_REUSEPORT failed')
		close_socket(server_fd)
		return -1
	}

	addr := if server.family == .ip6 {
		net.new_ip6(u16(server.port), [u8(0), 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]!)
	} else {
		net.new_ip(u16(server.port), [u8(0), 0, 0, 0]!)
	}
	alen := addr.len()
	if C.bind(server_fd, voidptr(&addr), alen) < 0 {
		eprintln(@LOCATION)
		C.perror(c'Bind failed')
		close_socket(server_fd)
		return -1
	}
	if C.listen(server_fd, max_connection_size) < 0 {
		eprintln(@LOCATION)
		C.perror(c'Listen failed')
		close_socket(server_fd)
		return -1
	}
	return server_fd
}

// add_fd_to_epoll adds a file descriptor to the epoll instance
fn add_fd_to_epoll(epoll_fd int, fd int, events u32) int {
	mut ev := C.epoll_event{
		events: events
	}
	ev.data.fd = fd
	if C.epoll_ctl(epoll_fd, C.EPOLL_CTL_ADD, fd, &ev) == -1 {
		eprintln(@LOCATION)
		C.perror(c'epoll_ctl')
		return -1
	}
	return 0
}

// remove_fd_from_epoll removes a file descriptor from the epoll instance
fn remove_fd_from_epoll(epoll_fd int, fd int) bool {
	ret := C.epoll_ctl(epoll_fd, C.EPOLL_CTL_DEL, fd, C.NULL)
	if ret == -1 {
		eprintln('ERROR: epoll_ctl(DEL, fd=${fd}) failed with errno=${C.errno}')
		return false
	}
	return true
}

fn handle_accept_loop(epoll_fd int, listen_fd int, mut client_fds map[int]bool) {
	for {
		client_fd := C.accept4(listen_fd, C.NULL, C.NULL, C.SOCK_NONBLOCK)
		if client_fd < 0 {
			if C.errno == C.EAGAIN || C.errno == C.EWOULDBLOCK {
				break // No more incoming connections; exit loop.
			}
			eprintln(@LOCATION)
			C.perror(c'Accept failed')
			break
		}
		// Enable TCP_NODELAY for lower latency
		opt := 1
		C.setsockopt(client_fd, C.IPPROTO_TCP, C.TCP_NODELAY, &opt, sizeof(opt))
		// Register client socket with epoll
		if add_fd_to_epoll(epoll_fd, client_fd, u32(C.EPOLLIN | C.EPOLLET)) == -1 {
			close_socket(client_fd)
			continue
		}
		client_fds[client_fd] = true
	}
}

fn free_write_state(server &Server, client_fd int, mut client_write_states map[int]&ClientWriteState) {
	mut state := client_write_states[client_fd] or { return }
	if state.content_owned && state.content.cap > 0 {
		unsafe { state.content.free() }
	}
	state.content = []u8{}
	if state.file_fd != -1 {
		C.close(state.file_fd)
		state.file_fd = -1
	}
	$if prealloc {
		if state.request_arena != unsafe { nil } {
			unsafe { prealloc_scope_free_after(state.request_arena) }
		}
	}
	state.request_arena = unsafe { nil }
	if state.request_active {
		server.end_request()
		state.request_active = false
	}
	client_write_states.delete(client_fd)
	unsafe { free(state) }
}

fn handle_client_closure(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]bool, mut client_buffers map[int][]u8, mut client_read_starts map[int]i64, mut closing_client_fds map[int]bool, mut client_write_states map[int]&ClientWriteState) {
	// Never close the listening socket here
	if client_fd == 0 {
		return
	}
	if client_fd <= 0 {
		eprintln('ERROR: Invalid FD=${client_fd} for closure')
		return
	}
	client_fds.delete(client_fd)
	client_buffers.delete(client_fd)
	client_read_starts.delete(client_fd)
	closing_client_fds.delete(client_fd)
	free_write_state(server, client_fd, mut client_write_states)
	remove_fd_from_epoll(epoll_fd, client_fd)
	close_socket(client_fd)
}

fn close_worker_clients(server &Server, epoll_fd int, mut client_fds map[int]bool, mut client_buffers map[int][]u8, mut client_read_starts map[int]i64, mut closing_client_fds map[int]bool, mut client_write_states map[int]&ClientWriteState) {
	for client_fd in client_fds.keys() {
		handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
			client_read_starts, mut closing_client_fds, mut client_write_states)
	}
}

fn drain_closing_client(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]bool, mut client_buffers map[int][]u8, mut client_read_starts map[int]i64, mut closing_client_fds map[int]bool, mut client_write_states map[int]&ClientWriteState) {
	mut drain_buf := []u8{len: 4096}
	for {
		bytes_read := C.recv(client_fd, unsafe { &drain_buf[0] }, drain_buf.len, 0)
		if bytes_read < 0 {
			if C.errno == C.EAGAIN || C.errno == C.EWOULDBLOCK {
				return
			}
			handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
			return
		}
		if bytes_read == 0 {
			handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
			return
		}
	}
}

fn send_terminal_response_and_drain(client_fd int, response []u8, mut client_buffers map[int][]u8, mut client_read_starts map[int]i64, mut closing_client_fds map[int]bool) {
	C.send(client_fd, response.data, response.len, C.MSG_NOSIGNAL)
	net.shutdown(client_fd, how: .write)
	client_buffers.delete(client_fd)
	client_read_starts.delete(client_fd)
	closing_client_fds[client_fd] = true
}

// try_drain_write attempts to push the remaining bytes of `state.content` and
// (if present) `state.file_fd` to the socket in non-blocking mode. It returns
// `.done` once all data has been transferred, `.blocked` when the kernel send
// buffer is full (EAGAIN/EWOULDBLOCK), or `.failed` on any unrecoverable error.
// The caller is responsible for parking the state until EPOLLOUT fires (when
// `.blocked` is returned) and for cleaning up via `free_write_state` once
// `.done` or `.failed` is reached.
fn try_drain_write(client_fd int, mut state ClientWriteState) DrainStatus {
	// 1. Drain in-memory content (response headers + possibly a body).
	for state.content_pos < state.content.len {
		remaining := state.content.len - state.content_pos
		sent := C.send(client_fd, unsafe { &state.content[state.content_pos] }, remaining,
			C.MSG_NOSIGNAL)
		if sent > 0 {
			state.content_pos += sent
			continue
		}
		if sent < 0 {
			errno_val := C.errno
			if errno_val == C.EAGAIN || errno_val == C.EWOULDBLOCK {
				return .blocked
			}
			if errno_val == C.EINTR {
				continue
			}
			eprintln('ERROR: send() failed with errno=${errno_val}')
			return .failed
		}
		// send() returning 0 on a non-blocking TCP socket is unusual; treat it as
		// a failure rather than spin.
		return .failed
	}

	// 2. Drain the optional file body via sendfile(2). sendfile updates the
	// offset pointer in place, so `state.file_pos` advances automatically.
	if state.file_fd != -1 {
		for state.file_pos < state.file_len {
			remaining := state.file_len - state.file_pos
			ssize := C.sendfile(client_fd, state.file_fd, &state.file_pos, usize(remaining))
			if ssize > 0 {
				continue
			}
			if ssize == 0 {
				// Short transfer: the file shrank between fstat() and now, or the
				// peer closed. We have already promised the full Content-Length,
				// so the response is now truncated -- closing the connection is
				// the only way to keep keep-alive clients in sync.
				eprintln('ERROR: sendfile() returned 0 with ${remaining} bytes still pending; closing connection to avoid desync')
				return .failed
			}
			errno_val := C.errno
			if errno_val == C.EAGAIN || errno_val == C.EWOULDBLOCK {
				return .blocked
			}
			if errno_val == C.EINTR {
				continue
			}
			match errno_val {
				C.EBADF {
					eprintln('ERROR: sendfile() EBADF: input fd or socket not open for required access (errno=${errno_val})')
				}
				C.EFAULT {
					eprintln('ERROR: sendfile() EFAULT: bad address for offset (errno=${errno_val})')
				}
				C.EINVAL {
					eprintln('ERROR: sendfile() EINVAL: invalid descriptor state or non-seekable input (errno=${errno_val})')
				}
				C.EIO {
					eprintln('ERROR: sendfile() EIO: I/O error while reading input file (errno=${errno_val})')
				}
				C.ENOMEM {
					eprintln('ERROR: sendfile() ENOMEM: insufficient kernel memory (errno=${errno_val})')
				}
				C.EOVERFLOW {
					eprintln('ERROR: sendfile() EOVERFLOW: count exceeds file/socket limits (errno=${errno_val})')
				}
				C.ESPIPE {
					eprintln('ERROR: sendfile() ESPIPE: input file not seekable with offset (errno=${errno_val})')
				}
				else {
					eprintln('ERROR: sendfile() failed with errno=${errno_val}')
				}
			}

			return .failed
		}
		// Done with the file -- close the fd eagerly so keep-alive doesn't hold it open.
		C.close(state.file_fd)
		state.file_fd = -1
	}

	return .done
}

// arm_epollout switches the connection's epoll mask from EPOLLIN|EPOLLET to
// EPOLLIN|EPOLLOUT|EPOLLET so the next round of pending bytes is delivered as
// an EPOLLOUT event instead of blocking the worker.
fn arm_epollout(epoll_fd int, client_fd int) int {
	mut ev := C.epoll_event{
		events: u32(C.EPOLLIN | C.EPOLLOUT | C.EPOLLET)
	}
	ev.data.fd = client_fd
	return C.epoll_ctl(epoll_fd, C.EPOLL_CTL_MOD, client_fd, &ev)
}

// complete_write tears down the per-fd write state after a successful drain,
// then either closes the connection or leaves it in keep-alive mode.
fn complete_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]bool, mut client_buffers map[int][]u8, mut client_read_starts map[int]i64, mut closing_client_fds map[int]bool, mut client_write_states map[int]&ClientWriteState) {
	state := client_write_states[client_fd] or { return }
	should_close := state.should_close
	epollout_was_armed := state.epollout_armed
	free_write_state(server, client_fd, mut client_write_states)
	client_buffers.delete(client_fd)
	if server.is_shutting_down() || should_close {
		handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
			client_read_starts, mut closing_client_fds, mut client_write_states)
		return
	}
	// Keep-alive: if EPOLLOUT was registered for the write, drop the fd from
	// epoll and re-add it with the original EPOLLIN|EPOLLET mask. The re-add
	// causes the kernel to fire a fresh edge for any pipelined request bytes
	// that piled up in the recv buffer while we were write-blocked; a plain
	// EPOLL_CTL_MOD would not generate that edge.
	if epollout_was_armed {
		remove_fd_from_epoll(epoll_fd, client_fd)
		if add_fd_to_epoll(epoll_fd, client_fd, u32(C.EPOLLIN | C.EPOLLET)) == -1 {
			handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
		}
	}
}

// handle_write resumes a blocked write when EPOLLOUT fires for `client_fd`.
fn handle_write(server &Server, epoll_fd int, client_fd int, mut client_fds map[int]bool, mut client_buffers map[int][]u8, mut client_read_starts map[int]i64, mut closing_client_fds map[int]bool, mut client_write_states map[int]&ClientWriteState) {
	mut state := client_write_states[client_fd] or { return }
	status := try_drain_write(client_fd, mut state)
	match status {
		.done {
			complete_write(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
		}
		.blocked {
			// Still blocked -- stay armed for the next EPOLLOUT.
		}
		.failed {
			handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
		}
	}
}

fn process_request(server &Server, epoll_fd int, client_fd int, request_buffer []u8, mut client_fds map[int]bool, mut client_buffers map[int][]u8, mut client_read_starts map[int]i64, mut closing_client_fds map[int]bool, mut client_write_states map[int]&ClientWriteState) {
	mut request_arena := voidptr(unsafe { nil })
	$if prealloc {
		request_arena = unsafe { prealloc_scope_begin() }
	}
	client_read_starts.delete(client_fd)
	server.begin_request()
	mut request_active := true

	mut decoded_http_request := decode_http_request(request_buffer) or {
		eprintln('Error decoding request ${err}')
		C.send(client_fd, tiny_bad_request_response.data, tiny_bad_request_response.len,
			C.MSG_NOSIGNAL)
		end_request_arena_current_thread(request_arena)
		if request_active {
			server.end_request()
		}
		handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
			client_read_starts, mut closing_client_fds, mut client_write_states)
		return
	}
	$if trace_prealloc ? {
		unsafe { prealloc_scope_checkpoint(c'fasthttp decoded request') }
	}
	decoded_http_request.client_conn_fd = client_fd
	decoded_http_request.user_data = server.user_data
	mut response := server.request_handler(decoded_http_request) or {
		eprintln('Error handling request ${err}')
		C.send(client_fd, tiny_bad_request_response.data, tiny_bad_request_response.len,
			C.MSG_NOSIGNAL)
		end_request_arena_current_thread(request_arena)
		if request_active {
			server.end_request()
		}
		handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
			client_read_starts, mut closing_client_fds, mut client_write_states)
		return
	}
	$if trace_prealloc ? {
		unsafe { prealloc_scope_checkpoint(c'fasthttp handler returned') }
	}
	response.attach_request_arena_if_empty(request_arena)

	match response.takeover_mode {
		.manual {
			// The handler has taken ownership of the connection.
			// Remove from epoll and tracking, but do NOT close the fd.
			client_fds.delete(client_fd)
			client_buffers.delete(client_fd)
			client_read_starts.delete(client_fd)
			closing_client_fds.delete(client_fd)
			remove_fd_from_epoll(epoll_fd, client_fd)
			response.abandon_request_arena_current_thread()
			response.free_owned_content()
			if request_active {
				server.end_request()
			}
			return
		}
		.reusable {
			set_blocking(client_fd, false)
			client_buffers.delete(client_fd)
			response.free_owned_content()
			response.end_request_arena_current_thread()
			if request_active {
				server.end_request()
			}
			if server.is_shutting_down() || response.should_close {
				handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut
					client_buffers, mut client_read_starts, mut closing_client_fds, mut
					client_write_states)
			}
			return
		}
		.none {}
	}

	// Open the file (if any) before constructing the state so we can fail fast
	// while we still own the request arena via response.
	mut file_fd := -1
	mut file_len := i64(0)
	if response.file_path != '' {
		fd := C.open(response.file_path.str, C.O_RDONLY, 0)
		if fd == -1 {
			eprintln('ERROR: open file failed')
			response.free_owned_content()
			response.end_request_arena_current_thread()
			if request_active {
				server.end_request()
			}
			handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
			return
		}
		mut st := C.stat{}
		if C.fstat(fd, &st) != 0 {
			eprintln('ERROR: fstat failed')
			C.close(fd)
			response.free_owned_content()
			response.end_request_arena_current_thread()
			if request_active {
				server.end_request()
			}
			handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
			return
		}
		file_fd = fd
		file_len = i64(st.st_size)
	}

	// Move content + arena ownership into the per-fd state. After this point we
	// must not touch the response's arena via end/free (it will be released by
	// free_write_state once the drain completes).
	content_bytes := response.take_or_clone_content()
	arena_ptr := response.take_request_arena()
	leave_request_arena_current_thread(arena_ptr)

	mut state := &ClientWriteState{
		content:        content_bytes
		content_pos:    0
		content_owned:  true
		file_fd:        file_fd
		file_len:       file_len
		file_pos:       0
		should_close:   response.should_close
		request_arena:  arena_ptr
		request_active: request_active
		start_ns:       time.sys_mono_now()
	}
	// From here on, the state owns end_request bookkeeping.
	request_active = false
	client_write_states[client_fd] = state

	status := try_drain_write(client_fd, mut state)
	match status {
		.done {
			complete_write(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
		}
		.blocked {
			// Kernel send buffer is full. Park the state and resume on EPOLLOUT.
			client_buffers.delete(client_fd)
			client_read_starts.delete(client_fd)
			if arm_epollout(epoll_fd, client_fd) == -1 {
				eprintln('ERROR: epoll_ctl(MOD, EPOLLOUT) failed errno=${C.errno}')
				handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut
					client_buffers, mut client_read_starts, mut closing_client_fds, mut
					client_write_states)
			} else {
				state.epollout_armed = true
			}
		}
		.failed {
			handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
		}
	}
}

fn process_events(server &Server, epoll_fd int, listen_fd int) {
	mut events := [max_connection_size]C.epoll_event{}
	mut request_buffer := []u8{len: server.max_request_buffer_size, cap: server.max_request_buffer_size}
	mut client_fds := map[int]bool{}
	mut client_buffers := map[int][]u8{}
	mut client_read_starts := map[int]i64{}
	mut closing_client_fds := map[int]bool{}
	mut client_write_states := map[int]&ClientWriteState{}
	unsafe {
		request_buffer.flags.set(.noslices | .nogrow | .noshrink)
	}
	for {
		if server.is_shutting_down() && server.active_request_count() == 0 {
			close_worker_clients(server, epoll_fd, mut client_fds, mut client_buffers, mut
				client_read_starts, mut closing_client_fds, mut client_write_states)
			return
		}
		num_events := C.epoll_wait(epoll_fd, &events[0], max_connection_size, epoll_wait_timeout_ms)
		if num_events < 0 {
			if C.errno == C.EINTR {
				continue
			}
			if server.is_shutting_down() {
				continue
			}
			eprintln('ERROR: epoll_wait() failed with errno=${C.errno}')
			continue
		}
		for i := 0; i < num_events; i++ {
			client_fd := unsafe { events[i].data.fd }
			// Accept new connections when the listening socket is readable
			if client_fd == listen_fd {
				if server.is_shutting_down() {
					continue
				}
				handle_accept_loop(epoll_fd, listen_fd, mut client_fds)
				continue
			}

			if events[i].events & u32((C.EPOLLHUP | C.EPOLLERR)) != 0 {
				if client_fd == listen_fd {
					eprintln('ERROR: listen fd had HUP/ERR')
					continue
				}
				if client_fd > 0 {
					// Don't bother sending 444 if there is an in-flight write -- the
					// peer is already disconnected and we want to tear down promptly.
					if client_fd !in client_write_states {
						C.send(client_fd, status_444_response.data, status_444_response.len,
							C.MSG_NOSIGNAL)
					}
					handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut
						client_buffers, mut client_read_starts, mut closing_client_fds, mut
						client_write_states)
				} else {
					eprintln('ERROR: Invalid FD from epoll: ${client_fd}')
				}
				continue
			}
			if events[i].events & u32(C.EPOLLOUT) != 0 {
				handle_write(server, epoll_fd, client_fd, mut client_fds, mut client_buffers, mut
					client_read_starts, mut closing_client_fds, mut client_write_states)
				// If both EPOLLIN and EPOLLOUT fired, we still want to drain the
				// socket -- fall through to the EPOLLIN branch below.
				if client_fd !in client_fds {
					continue
				}
			}
			if events[i].events & u32(C.EPOLLIN) != 0 {
				if server.is_shutting_down() {
					handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut
						client_buffers, mut client_read_starts, mut closing_client_fds, mut
						client_write_states)
					continue
				}
				if closing_client_fds[client_fd] or { false } {
					drain_closing_client(server, epoll_fd, client_fd, mut client_fds, mut
						client_buffers, mut client_read_starts, mut closing_client_fds, mut
						client_write_states)
					continue
				}
				if client_fd in client_write_states {
					// A response is still being drained for this fd. Reading a new
					// request now would overwrite the in-flight ClientWriteState and
					// silently drop the response/file transfer. Defer until the write
					// completes -- on keep-alive completion we re-arm EPOLLIN below so
					// any bytes that arrived during the write get delivered as a fresh
					// edge.
					continue
				}
				// Read all available data from the socket
				mut total_bytes_read := 0
				mut readed_request_buffer := client_buffers[client_fd] or {
					[]u8{cap: server.max_request_buffer_size}
				}
				mut headers_complete := false
				mut header_too_large := false
				mut header_end_pos := -1
				mut request_complete := false
				mut peer_closed := false
				mut recv_error := false

				for {
					bytes_read := C.recv(client_fd, unsafe { &request_buffer[0] },
						server.max_request_buffer_size - 1, 0)
					if bytes_read < 0 {
						if C.errno == C.EAGAIN || C.errno == C.EWOULDBLOCK {
							// No more data available right now
							break
						}
						// Error occurred
						eprintln('ERROR: recv() failed with errno=${C.errno}')
						recv_error = true
						break
					} else if bytes_read == 0 {
						// Connection closed by client
						peer_closed = true
						break
					}

					unsafe {
						readed_request_buffer.push_many(&request_buffer[0], bytes_read)
					}
					total_bytes_read += bytes_read
					if client_fd !in client_read_starts {
						client_read_starts[client_fd] = time.sys_mono_now()
					}

					// Enforce the configured limit on request headers, not on the whole body.
					buffer_len := readed_request_buffer.len
					if !headers_complete && buffer_len >= 4 {
						header_end_pos = find_header_end_in_buf(readed_request_buffer.data,
							buffer_len)
						if header_end_pos == -1 {
							if buffer_len >= server.max_request_buffer_size {
								header_too_large = true
								break
							}
						} else {
							headers_complete = true
							if header_end_pos > server.max_request_buffer_size {
								header_too_large = true
								break
							}
						}
					}

					if headers_complete && has_complete_body(readed_request_buffer.data, buffer_len) {
						request_complete = true
						break
					}
				}

				if header_too_large {
					send_terminal_response_and_drain(client_fd, status_413_response, mut
						client_buffers, mut client_read_starts, mut closing_client_fds)
					continue
				}
				if request_complete {
					process_request(server, epoll_fd, client_fd, readed_request_buffer, mut
						client_fds, mut client_buffers, mut client_read_starts, mut
						closing_client_fds, mut client_write_states)
				} else if recv_error {
					// Unexpected recv error - send 444 No Response
					C.send(client_fd, status_444_response.data, status_444_response.len,
						C.MSG_NOSIGNAL)
					handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut
						client_buffers, mut client_read_starts, mut closing_client_fds, mut
						client_write_states)
				} else if peer_closed || (total_bytes_read == 0 && readed_request_buffer.len == 0) {
					// Normal client closure (FIN received)
					handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut
						client_buffers, mut client_read_starts, mut closing_client_fds, mut
						client_write_states)
				} else if readed_request_buffer.len > 0 {
					client_buffers[client_fd] = readed_request_buffer
				}
			}
		}
		if server.timeout_in_seconds > 0 {
			now := time.sys_mono_now()
			timeout_ns := i64(server.timeout_in_seconds) * 1_000_000_000
			for client_fd in client_read_starts.keys() {
				started := client_read_starts[client_fd] or { continue }
				if now - started >= timeout_ns {
					send_terminal_response_and_drain(client_fd, status_408_response, mut
						client_buffers, mut client_read_starts, mut closing_client_fds)
				}
			}
			// Sweep write-side stalls: a client that armed EPOLLOUT but never
			// drains within `timeout_in_seconds` gets the connection torn down so
			// the worker isn't pinned waiting for a dead peer.
			for client_fd in client_write_states.keys() {
				state := client_write_states[client_fd] or { continue }
				if state.start_ns > 0 && now - state.start_ns >= timeout_ns {
					handle_client_closure(server, epoll_fd, client_fd, mut client_fds, mut
						client_buffers, mut client_read_starts, mut closing_client_fds, mut
						client_write_states)
				}
			}
		}
	}
}

fn (mut server Server) stop_accepting() {
	for i := 0; i < max_thread_pool_size; i++ {
		if server.listen_fds[i] < 0 {
			continue
		}
		if server.epoll_fds[i] >= 0 {
			remove_fd_from_epoll(server.epoll_fds[i], server.listen_fds[i])
		}
		close_socket(server.listen_fds[i])
		server.listen_fds[i] = -1
	}
}

// run starts the server and begins listening for incoming connections.
pub fn (mut server Server) run() ! {
	$if windows {
		eprintln('Windows is not supported yet')
		return
	}
	for i := 0; i < max_thread_pool_size; i++ {
		server.listen_fds[i] = create_server_socket(server)
		if server.listen_fds[i] < 0 {
			return
		}

		server.epoll_fds[i] = C.epoll_create1(0)
		if server.epoll_fds[i] < 0 {
			C.perror(c'epoll_create1 failed')
			close_socket(server.listen_fds[i])
			return
		}

		// Register the listening socket with each worker epoll for distributed accepts (edge-triggered)
		if add_fd_to_epoll(server.epoll_fds[i], server.listen_fds[i], u32(C.EPOLLIN | C.EPOLLET)) == -1 {
			close_socket(server.listen_fds[i])
			close_socket(server.epoll_fds[i])
			return
		}

		server.threads[i] = spawn process_events(server, server.epoll_fds[i], server.listen_fds[i])
	}

	server.mark_running()
	println('listening on http://0.0.0.0:${server.port}/')
	// Main thread waits for workers; accepts are handled in worker epoll loops
	for i in 0 .. max_thread_pool_size {
		server.threads[i].wait()
		if server.epoll_fds[i] >= 0 {
			close_socket(server.epoll_fds[i])
			server.epoll_fds[i] = -1
		}
	}
	server.mark_stopped()
}