//! platform readiness poller: epoll (linux) / kqueue (darwin, bsd).
//! level-triggered on both platforms — handlers drain until EAGAIN.
//! written against zig 0.16.0; syscalls via std.c (libc required).
const std = @import("std");
const builtin = @import("builtin");
const c = std.c;

pub const Interest = struct {
    read: bool = false,
    write: bool = false,
};

pub const Event = struct {
    token: u64,
    readable: bool,
    writable: bool,
    hup: bool,
    /// the fd this readiness is for. Delivery is destructive -- it clears the
    /// token and drops the registration -- so a caller that cannot consume
    /// the event needs the fd back in order to re-arm it.
    fd: c.fd_t = -1,
};

pub const Poller = switch (builtin.os.tag) {
    .linux => Epoll,
    .macos, .ios, .tvos, .watchos, .visionos, .freebsd, .netbsd, .openbsd, .dragonfly => Kqueue,
    else => @compileError("zio poller: unsupported os"),
};

pub const Epoll = struct {
    epfd: c.fd_t,
    /// diagnostics: arming is allowed to fail (fd closed, watch limit) and
    /// the failure is otherwise invisible from outside the process.
    arm_failures: u64 = 0,
    last_arm_errno: i32 = 0,
    allocator: std.mem.Allocator,
    /// epoll holds ONE registration per fd, so per-direction one-shot arms
    /// live here: independent read/write tokens, merged kernel interest.
    /// without this, a writer fiber arming an fd replaces a parked reader's
    /// interest and token — a lost wake (proven by the arm-collision test).
    armed: std.AutoHashMapUnmanaged(c.fd_t, Armed) = .empty,

    const Armed = struct { read_token: ?u64 = null, write_token: ?u64 = null };
    /// tags kernel data words that refer to the `armed` table (vs literal
    /// tokens from `add`, e.g. the wake pipe).
    const TAG: u64 = 1 << 63;

    const EPOLL = std.os.linux.EPOLL;

    pub fn init(allocator: std.mem.Allocator) !Epoll {
        const epfd = c.epoll_create1(EPOLL.CLOEXEC);
        if (epfd < 0) return error.PollerInitFailed;
        return .{ .epfd = epfd, .allocator = allocator };
    }

    /// remove `fd`'s registration and return whoever was parked on it.
    /// called when the fd is about to be closed: the kernel drops closed fds
    /// from the epoll set silently, so a fiber parked on one would never be
    /// woken again.
    pub fn takeWaiters(p: *Epoll, fd: c.fd_t) [2]?u64 {
        const e = p.armed.fetchRemove(fd) orelse return .{ null, null };
        _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null);
        return .{ e.value.read_token, e.value.write_token };
    }

    /// number of fds with at least one armed direction (diagnostic).
    pub fn registrationCount(p: *const Epoll) usize {
        return p.armed.count();
    }

    pub fn deinit(p: *Epoll) void {
        p.armed.deinit(p.allocator);
        _ = c.close(p.epfd);
    }

    pub fn add(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void {
        var ev: c.epoll_event = .{
            .events = eventsFrom(interest),
            .data = .{ .u64 = token },
        };
        if (c.epoll_ctl(p.epfd, EPOLL.CTL_ADD, fd, &ev) != 0) return error.PollerAddFailed;
    }

    pub fn del(p: *Epoll, fd: c.fd_t) void {
        _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null);
        _ = p.armed.remove(fd);
    }

    pub fn mod(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void {
        var ev: c.epoll_event = .{
            .events = eventsFrom(interest),
            .data = .{ .u64 = token },
        };
        if (c.epoll_ctl(p.epfd, EPOLL.CTL_MOD, fd, &ev) != 0) return error.PollerModFailed;
    }

    /// arm a single direction: fires once for that direction, then disarms
    /// until re-armed. the other direction's arm (if any) is preserved.
    pub fn armOneshot(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void {
        const gop = p.armed.getOrPut(p.allocator, fd) catch return error.PollerArmFailed;
        if (!gop.found_existing) gop.value_ptr.* = .{};
        if (interest.read) gop.value_ptr.read_token = token else gop.value_ptr.write_token = token;

        var ev: c.epoll_event = .{
            .events = mergedEvents(gop.value_ptr.*),
            .data = .{ .u64 = TAG | @as(u64, @intCast(@as(u32, @bitCast(fd)))) },
        };
        if (c.epoll_ctl(p.epfd, EPOLL.CTL_MOD, fd, &ev) == 0) return;
        // MOD failed: the fd was not registered — it is fresh (or was closed
        // and the number reused). any other-direction token in the table is
        // from a previous fd lifetime; drop it so a dead fiber's arm cannot
        // shadow this registration.
        gop.value_ptr.* = .{};
        if (interest.read) gop.value_ptr.read_token = token else gop.value_ptr.write_token = token;
        ev.events = mergedEvents(gop.value_ptr.*);
        if (c.epoll_ctl(p.epfd, EPOLL.CTL_ADD, fd, &ev) == 0) return;
        // both MOD and ADD failed. MOD=ENOENT means "not registered"; ADD
        // failing then is usually ENOSPC/ENOMEM — the epoll watch limit.
        // record it: a silent arm failure turns a parked fiber into a spin,
        // and a spinning accept loop starves the whole scheduler.
        p.last_arm_errno = @backingInt(c.errno(@as(isize, -1)));
        p.arm_failures +%= 1;
        _ = p.armed.remove(fd);
        return error.PollerArmFailed;
    }

    /// fills `out` with ready events; blocks up to timeout_ms. returns slice
    /// of out. a single fd can yield two events (read + write waiters), so
    /// callers size `out` at 2x their raw batch.
    pub fn wait(p: *Epoll, out: []Event, timeout_ms: i32) []Event {
        var raw: [max_batch]c.epoll_event = undefined;
        const cap: c_uint = @intCast(@min(out.len / 2, raw.len));
        const n = c.epoll_wait(p.epfd, &raw, cap, timeout_ms);
        if (n <= 0) return out[0..0];
        const count: usize = @intCast(n);
        var emitted: usize = 0;
        for (raw[0..count]) |r| {
            const readable = r.events & EPOLL.IN != 0;
            const writable = r.events & EPOLL.OUT != 0;
            const hup = r.events & (EPOLL.HUP | EPOLL.ERR | EPOLL.RDHUP) != 0;
            if (r.data.u64 & TAG == 0) {
                out[emitted] = .{ .token = r.data.u64, .readable = readable, .writable = writable, .hup = hup, .fd = -1 };
                emitted += 1;
                continue;
            }
            const fd: c.fd_t = @bitCast(@as(u32, @truncate(r.data.u64)));
            const entry = p.armed.getPtr(fd) orelse {
                // stale kernel registration with no table entry: remove it
                // so it cannot fire again.
                _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null);
                continue;
            };
            if ((readable or hup) and entry.read_token != null) {
                out[emitted] = .{ .token = entry.read_token.?, .readable = true, .writable = false, .hup = hup, .fd = fd };
                entry.read_token = null;
                emitted += 1;
            }
            if ((writable or hup) and entry.write_token != null) {
                out[emitted] = .{ .token = entry.write_token.?, .readable = false, .writable = true, .hup = hup, .fd = fd };
                entry.write_token = null;
                emitted += 1;
            }
            // userspace one-shot: narrow the interest, and DELETE the
            // registration once no direction is armed. an empty registration
            // left as RDHUP-only keeps firing for every dead socket
            // (level-triggered), and a few hundred churned connections turn
            // every epoll batch into token-less hangup noise that starves
            // real events — observed as the accept loop going deaf while
            // everything else stayed healthy.
            if (entry.read_token == null and entry.write_token == null) {
                _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null);
                _ = p.armed.remove(fd);
            } else {
                var ev: c.epoll_event = .{
                    .events = mergedEvents(entry.*),
                    .data = .{ .u64 = TAG | @as(u64, @intCast(@as(u32, @bitCast(fd)))) },
                };
                _ = c.epoll_ctl(p.epfd, EPOLL.CTL_MOD, fd, &ev);
            }
        }
        return out[0..emitted];
    }

    fn mergedEvents(a: Armed) u32 {
        var events: u32 = EPOLL.RDHUP;
        if (a.read_token != null) events |= EPOLL.IN;
        if (a.write_token != null) events |= EPOLL.OUT;
        return events;
    }

    fn eventsFrom(interest: Interest) u32 {
        var events: u32 = EPOLL.RDHUP;
        if (interest.read) events |= EPOLL.IN;
        if (interest.write) events |= EPOLL.OUT;
        return events;
    }
};

pub const Kqueue = struct {
    kq: c.fd_t,
    arm_failures: u64 = 0,
    last_arm_errno: i32 = 0,
    allocator: std.mem.Allocator,
    /// mirrors the kernel's per-direction one-shot arms. The kernel does not
    /// need it (filters are independent), but close-while-parked does:
    /// closing an fd silently removes its kevents, so without a userspace
    /// record of who is parked there the fiber is stranded forever — the
    /// same deafness class the epoll table already guards against.
    armed: std.AutoHashMapUnmanaged(c.fd_t, Armed) = .empty,

    const Armed = struct { read_token: ?u64 = null, write_token: ?u64 = null };

    pub fn init(allocator: std.mem.Allocator) !Kqueue {
        const kq = c.kqueue();
        if (kq < 0) return error.PollerInitFailed;
        return .{ .kq = kq, .allocator = allocator };
    }

    /// remove `fd`'s arms and return whoever was parked on it — called when
    /// the fd is about to be closed. The kernel removes the kevents itself
    /// on close; dropping the table entry here keeps the mirror honest.
    pub fn takeWaiters(p: *Kqueue, fd: c.fd_t) [2]?u64 {
        const e = p.armed.fetchRemove(fd) orelse return .{ null, null };
        return .{ e.value.read_token, e.value.write_token };
    }

    pub fn registrationCount(p: *const Kqueue) usize {
        return p.armed.count();
    }

    pub fn deinit(p: *Kqueue) void {
        p.armed.deinit(p.allocator);
        _ = c.close(p.kq);
    }

    pub fn add(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void {
        // both filters are always registered; interest toggles ENABLE/DISABLE.
        // kqueue EV_ADD on an existing (ident,filter) updates it, so mod == add.
        const changes = [2]c.Kevent{
            kev(fd, c.EVFILT.READ, interest.read, token),
            kev(fd, c.EVFILT.WRITE, interest.write, token),
        };
        var none: [0]c.Kevent = .{};
        const rc = c.kevent(p.kq, &changes, changes.len, &none, 0, null);
        if (rc < 0) return error.PollerAddFailed;
    }

    pub fn mod(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void {
        return p.add(fd, interest, token) catch error.PollerModFailed;
    }

    /// arm a single edge: fires once, then disarms until re-armed.
    pub fn armOneshot(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void {
        const gop = p.armed.getOrPut(p.allocator, fd) catch return error.PollerArmFailed;
        if (!gop.found_existing) gop.value_ptr.* = .{};
        if (interest.read) gop.value_ptr.read_token = token else gop.value_ptr.write_token = token;
        var changes: [1]c.Kevent = undefined;
        const filter: @FieldType(c.Kevent, "filter") = if (interest.read) c.EVFILT.READ else c.EVFILT.WRITE;
        changes[0] = .{
            .ident = @intCast(fd),
            .filter = filter,
            .flags = c.EV.ADD | c.EV.ENABLE | c.EV.ONESHOT,
            .fflags = 0,
            .data = 0,
            .udata = token,
        };
        var none: [0]c.Kevent = .{};
        if (c.kevent(p.kq, &changes, changes.len, &none, 0, null) < 0) {
            if (interest.read) gop.value_ptr.read_token = null else gop.value_ptr.write_token = null;
            if (gop.value_ptr.read_token == null and gop.value_ptr.write_token == null)
                _ = p.armed.remove(fd);
            return error.PollerArmFailed;
        }
    }

    pub fn del(p: *Kqueue, fd: c.fd_t) void {
        // closing the fd removes its kevents automatically; explicit EV_DELETE
        // would race with close-initiated removal, so rely on close semantics.
        _ = p;
        _ = fd;
    }

    pub fn wait(p: *Kqueue, out: []Event, timeout_ms: i32) []Event {
        var raw: [max_batch]c.Kevent = undefined;
        const cap: c_int = @intCast(@min(out.len, raw.len));
        const ts: c.timespec = .{
            .sec = @divTrunc(timeout_ms, 1000),
            .nsec = @as(@FieldType(c.timespec, "nsec"), @intCast(@mod(timeout_ms, 1000))) * 1_000_000,
        };
        const no_changes: [0]c.Kevent = .{};
        const n = c.kevent(p.kq, &no_changes, 0, &raw, cap, &ts);
        if (n <= 0) return out[0..0];
        const count: usize = @intCast(n);
        for (raw[0..count], out[0..count]) |r, *e| {
            const fd: c.fd_t = @intCast(r.ident);
            e.* = .{
                .token = r.udata,
                .readable = r.filter == c.EVFILT.READ,
                .writable = r.filter == c.EVFILT.WRITE,
                .hup = r.flags & c.EV.EOF != 0,
                .fd = fd,
            };
            // EV_ONESHOT consumed the kernel arm; mirror that here. Literal
            // tokens from add() (the wake pipe) are not in the table.
            if (p.armed.getPtr(fd)) |entry| {
                if (e.readable) entry.read_token = null else entry.write_token = null;
                if (entry.read_token == null and entry.write_token == null)
                    _ = p.armed.remove(fd);
            }
        }
        return out[0..count];
    }

    fn kev(fd: c.fd_t, filter: @FieldType(c.Kevent, "filter"), enabled: bool, token: u64) c.Kevent {
        return .{
            .ident = @intCast(fd),
            .filter = filter,
            .flags = c.EV.ADD | @as(@FieldType(c.Kevent, "flags"), if (enabled) c.EV.ENABLE else c.EV.DISABLE),
            .fflags = 0,
            .data = 0,
            .udata = token,
        };
    }
};

pub const max_batch = 256;

test "epoll: a delivered readiness event destroys the arm and is never redelivered" {
    if (builtin.os.tag != .linux) return error.SkipZigTest;
    var p = try Epoll.init(std.testing.allocator);
    defer p.deinit();

    var fds: [2]c.fd_t = undefined;
    try std.testing.expect(c.socketpair(c.AF.UNIX, c.SOCK.STREAM, 0, &fds) == 0);
    defer _ = c.close(fds[0]);
    defer _ = c.close(fds[1]);

    const token: u64 = 0xBEEF;
    try p.armOneshot(fds[0], .{ .read = true }, token);
    try std.testing.expectEqual(@as(usize, 1), p.registrationCount());

    // peer writes then closes: readable + RDHUP, the websocket close-frame case
    try std.testing.expectEqual(@as(isize, 3), c.write(fds[1], "bye", 3));
    _ = c.close(fds[1]);

    var out: [max_batch * 2]Event = undefined;
    const first = p.wait(&out, 50);
    try std.testing.expectEqual(@as(usize, 1), first.len);
    try std.testing.expectEqual(token, first[0].token);

    // delivery already destroyed the registration
    try std.testing.expectEqual(@as(usize, 0), p.registrationCount());

    // the caller drops this event (the fiber it named was not parked).
    // Nothing re-arms. The bytes are still unread and the peer is gone, but
    // the poller will never mention this fd again -- it is now invisible.
    const second = p.wait(&out, 50);
    try std.testing.expectEqual(@as(usize, 0), second.len);

    var pending: c_int = 0;
    _ = c.ioctl(fds[0], @as(c_int, 0x541B), &pending); // FIONREAD
    try std.testing.expectEqual(@as(c_int, 3), pending);
}