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()
}