use super::super::Pool;
use super::super::slot::WRITE_ROOM_MIN;
use super::super::slot::{BufSlot, Conn, ConnFlags, CoreSlot};
use super::Server;
use super::state::StreamState;
use super::state::{ServerSlotState, SlotEgress, SlotIngress};
use super::wake::WakeQueue;
use crate::CodecLayer;
use crate::SlotId;
use crate::WriteBufStorage;
use crate::protocol::server::{
    HandlerAction, ParkAction, PendingFuture, PendingResponse, RequestBodyStream,
    RequestStreamAction, RouteRequestBodyKind, ServerProtocol,
};
use dambi::BytesMut;
use dope_core::driver::{DriverOps, PoolDriver, ProvidedRingOps};
use dope_core::profile::Profile;
use dope_transport::Transport;

type PendingFutureBody = dambi::InlineFuture<'static, PendingResponse, 64>;

struct ParseCtx<'a, H: ServerProtocol> {
    protocol: &'a mut H,
    conn: &'a mut Conn,
    buf: &'a mut BufSlot<<H::CodecLayer as CodecLayer>::WriteBuf>,
    user: &'a mut ServerSlotState<H>,
    slot_idx: usize,
}

enum ParseStep {
    Continue(usize),
    Stop(usize),
}

enum OutcomeStep {
    Step(ParseStep),
    Drop,
}

impl<'a, H: ServerProtocol> ParseCtx<'a, H> {
    #[cold]
    fn install_pending(&mut self, future: PendingFutureBody) {
        self.conn.flags.set(ConnFlags::PARKED, true);
        let id = SlotId::from_parts(self.slot_idx as u32, self.conn.generation);
        self.user.pending_future = Some(PendingFuture {
            generation: self.conn.generation,
            future,
        });
        WakeQueue::push(id);
    }

    #[inline(always)]
    fn clear_parse_phase(&mut self) {
        self.conn.parse_phase = 0;
        self.conn.body_needed = 0;
    }

    #[inline(always)]
    fn apply_send(
        &mut self,
        wh: usize,
        written: u16,
        close_after: bool,
        total: usize,
    ) -> OutcomeStep {
        self.conn.write_head = wh as u16 + written;
        if close_after {
            self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
        }
        OutcomeStep::Step(ParseStep::Continue(total))
    }

    #[inline(always)]
    fn apply_close_payload(
        &mut self,
        wh: usize,
        bytes: &'static [u8],
        total: usize,
    ) -> OutcomeStep {
        let write_slice = &mut self.buf.write_buf.as_mut_slice()[wh..];
        if write_slice.len() < bytes.len() {
            return OutcomeStep::Drop;
        }
        write_slice[..bytes.len()].copy_from_slice(bytes);
        self.conn.write_head = wh as u16 + bytes.len() as u16;
        self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
        OutcomeStep::Step(ParseStep::Continue(total))
    }

    #[inline(always)]
    fn apply_outcome(&mut self, outcome: HandlerAction, wh: usize, total: usize) -> OutcomeStep {
        match outcome {
            HandlerAction::Send {
                written,
                close_after,
            } => self.apply_send(wh, written, close_after, total),
            HandlerAction::SendStatic {
                hdr_written,
                body,
                close_after,
            } => {
                self.conn.write_head = (wh as u32 + hdr_written).min(u16::MAX as u32) as u16;
                self.conn.pending_body_ptr = body.as_ptr();
                self.conn.pending_body_len = body.len() as u32;
                if close_after {
                    self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
                }
                OutcomeStep::Step(ParseStep::Continue(total))
            }
            HandlerAction::SendStream {
                hdr_written,
                stream,
                close_after,
            } => {
                self.conn.write_head = (wh as u32 + hdr_written).min(u16::MAX as u32) as u16;
                self.user.egress = SlotEgress::Stream(Box::new(StreamState::new(stream)));
                if close_after {
                    self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
                }
                OutcomeStep::Step(ParseStep::Continue(total))
            }
            HandlerAction::Pending(future) => {
                self.install_pending(future);
                OutcomeStep::Step(ParseStep::Stop(total))
            }
            HandlerAction::Close(bytes) => self.apply_close_payload(wh, bytes, total),
            HandlerAction::SendVectored {
                iovs,
                iov_count,
                cookie,
                total_bytes,
                close_after,
            } => {
                let conn = &mut *self.conn;
                let buf = &mut *self.buf;
                if buf.staged_gather_cookie != cookie {
                    let n = (iov_count as usize).min(buf.pending_iovs.len());
                    for (i, iov) in iovs.iter().enumerate().take(n) {
                        buf.pending_iovs[i] = iov.as_libc();
                    }
                    buf.staged_gather_cookie = cookie;
                }
                conn.large_inflight = total_bytes;
                conn.large_tail = 0;
                self.user.egress = SlotEgress::Gather { iov_count };
                if close_after {
                    conn.flags.set(ConnFlags::CLOSE_AFTER, true);
                }
                OutcomeStep::Step(ParseStep::Continue(total))
            }
        }
    }

    #[inline(always)]
    fn apply_park_outcome(&mut self, outcome: ParkAction, wh: usize, total: usize) -> OutcomeStep {
        match outcome {
            ParkAction::Send {
                written,
                close_after,
            } => self.apply_send(wh, written, close_after, total),
            ParkAction::Park => {
                self.conn.flags.set(ConnFlags::PARKED, true);
                OutcomeStep::Step(ParseStep::Stop(total))
            }
            ParkAction::Close(bytes) => self.apply_close_payload(wh, bytes, total),
        }
    }

    #[inline(always)]
    fn parse_one(&mut self, view: &[u8]) -> ParseStep {
        let Some((head, head_len, body_len)) =
            self.protocol.parse(&mut self.user.protocol_state, view)
        else {
            return ParseStep::Stop(0);
        };
        let head_len = head_len as usize;
        let body_needed: usize = if self.conn.parse_phase == 1 {
            self.conn.body_needed as usize
        } else {
            body_len as usize
        };

        let route_kind = self.protocol.route_request_body_kind(&head);

        if body_needed > 0 && matches!(route_kind, RouteRequestBodyKind::Stream) {
            return self.dispatch_stream(view, head, head_len, body_needed);
        }

        if head_len.saturating_add(body_needed) > self.buf.ingress_buf.len() {
            return self.dispatch_oversized(view, head_len, body_needed, route_kind);
        }

        let total = head_len + body_needed;
        if view.len() < total {
            self.conn.parse_phase = 1;
            self.conn.body_needed = body_needed as u32;
            return ParseStep::Stop(0);
        }

        let wh = self.conn.write_head as usize;
        if self.buf.write_buf.as_slice().len().saturating_sub(wh) < WRITE_ROOM_MIN {
            return ParseStep::Stop(0);
        }
        if !self.conn.pending_body_ptr.is_null()
            || !self.user.egress.is_idle()
            || self.conn.flags.contains(ConnFlags::PARKED)
        {
            return ParseStep::Stop(0);
        }

        let req_bytes: &[u8] = &view[..total];
        let write_slice: &mut [u8] = &mut self.buf.write_buf.as_mut_slice()[wh..];
        if H::SUPPORTS_PARK {
            let conn_id = SlotId::from_parts(self.slot_idx as u32, self.conn.generation);
            let outcome = self.protocol.handle_park(
                conn_id,
                &mut self.user.protocol_state,
                req_bytes,
                head,
                write_slice,
            );
            return match self.apply_park_outcome(outcome, wh, total) {
                OutcomeStep::Step(step) => {
                    self.clear_parse_phase();
                    step
                }
                OutcomeStep::Drop => ParseStep::Stop(0),
            };
        }
        let outcome = self.protocol.handle(req_bytes, head, write_slice);
        match self.apply_outcome(outcome, wh, total) {
            OutcomeStep::Step(step) => {
                self.clear_parse_phase();
                step
            }
            OutcomeStep::Drop => ParseStep::Stop(0),
        }
    }

    #[cold]
    fn dispatch_stream<'v>(
        &mut self,
        view: &'v [u8],
        head: <H as ServerProtocol>::Head<'v>,
        head_len: usize,
        body_needed: usize,
    ) -> ParseStep {
        if !self.conn.pending_body_ptr.is_null()
            || !self.user.ingress.is_idle()
            || !self.user.egress.is_idle()
        {
            return ParseStep::Stop(0);
        }
        let wh = self.conn.write_head as usize;
        if self.buf.write_buf.as_slice().len().saturating_sub(wh) < WRITE_ROOM_MIN {
            return ParseStep::Stop(0);
        }

        let (receiver, handle) = RequestBodyStream::new(body_needed as u32);
        let pushed = handle.push_slice(&view[head_len..]);
        let eof = handle.is_complete();
        if !eof {
            self.user.ingress = SlotIngress::RequestStream(Box::new(handle));
        }

        let outcome = {
            let head_bytes: &[u8] = &view[..head_len];
            let write_slice: &mut [u8] = &mut self.buf.write_buf.as_mut_slice()[wh..];
            self.protocol
                .handle_request_stream(head_bytes, head, receiver, write_slice)
        };
        let consumed = head_len + pushed;
        self.clear_parse_phase();
        match outcome {
            RequestStreamAction::Pending(future) => self.install_pending(future),
            RequestStreamAction::Close(bytes) => {
                let write_slice = &mut self.buf.write_buf.as_mut_slice()[wh..];
                let n = bytes.len().min(write_slice.len());
                write_slice[..n].copy_from_slice(&bytes[..n]);
                self.conn.write_head = wh as u16 + n as u16;
                self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
            }
        }
        ParseStep::Stop(consumed)
    }

    #[cold]
    fn dispatch_oversized(
        &mut self,
        view: &[u8],
        head_len: usize,
        body_needed: usize,
        route_kind: RouteRequestBodyKind,
    ) -> ParseStep {
        if matches!(route_kind, RouteRequestBodyKind::Spilled) {
            let total = head_len + body_needed;
            let mut spill = BytesMut::with_capacity(total);
            let available = view.len().min(total);
            spill.extend_from_slice(&view[..available]);
            self.user.ingress = SlotIngress::Spill {
                buf: Box::new(spill),
                filled: available as u32,
                target: total as u32,
            };
            self.clear_parse_phase();
            return ParseStep::Stop(available);
        }
        let wh = self.conn.write_head as usize;
        let bytes = self.protocol.oversize_response();
        if self.buf.write_buf.as_slice().len().saturating_sub(wh) < bytes.len() {
            return ParseStep::Stop(0);
        }
        self.buf.write_buf.as_mut_slice()[wh..wh + bytes.len()].copy_from_slice(bytes);
        self.conn.write_head = wh as u16 + bytes.len() as u16;
        self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
        self.clear_parse_phase();
        ParseStep::Stop(view.len())
    }
}

impl<H: ServerProtocol, T: Transport, F: Profile, B: PoolDriver> super::super::PoolRecv<B>
    for Pool<Server<H, T>, F, B>
{
    #[inline(always)]
    fn on_recv_event(
        &mut self,
        driver: &mut B::Driver,
        token: B::Token,
        result: i32,
        more: bool,
        bid: Option<u16>,
    ) {
        match self.core.classify_recv(driver, token, result, more, bid) {
            super::super::core::RecvDispatch::Skipped => {}
            super::super::core::RecvDispatch::EofClose { slot_idx } => {
                self.core.free_slot(driver, slot_idx);
            }
            super::super::core::RecvDispatch::Errno {
                slot_idx,
                errno,
                more,
            } => {
                self.on_recv_errno(driver, slot_idx, errno, more);
            }
            super::super::core::RecvDispatch::Chunk {
                slot_idx,
                ptr,
                len,
                bid,
                more,
            } => {
                let src: &[u8] = unsafe { std::slice::from_raw_parts(ptr, len) };
                if <H::CodecLayer as CodecLayer>::IS_PASSTHROUGH {
                    self.ingest_plaintext(driver, slot_idx, src);
                } else {
                    self.ingest_codec(driver, slot_idx, src);
                }
                driver.provided_ring().defer(bid);
                if !self.core.slots[slot_idx]
                    .conn
                    .flags
                    .contains(ConnFlags::LIVE)
                {
                    return;
                }
                self.submit_slot_write(driver, slot_idx);
                if !more {
                    let _ = self.core.arm_recv(driver, slot_idx);
                }
            }
            super::super::core::RecvDispatch::DrainOnly { slot_idx, more } => {
                if !more {
                    let _ = self.core.arm_recv(driver, slot_idx);
                }
            }
        }
    }
}

impl<H: ServerProtocol, T: Transport, F: Profile, B: PoolDriver> Pool<Server<H, T>, F, B> {
    #[cold]
    #[inline(never)]
    fn on_recv_errno(&mut self, driver: &mut B::Driver, slot_idx: usize, errno: i32, more: bool) {
        if errno == libc::EINVAL {
            panic!("dope invariant violated: recv-multi unsupported on required uring platform");
        }
        if errno == libc::ECANCELED {
            if !more {
                self.core.slots[slot_idx].io.recv.armed = false;
            }
            return;
        }
        self.core.slots[slot_idx].io.recv.armed = false;
        self.core.free_slot(driver, slot_idx);
    }

    #[cold]
    fn ingest_codec(&mut self, driver: &mut B::Driver, slot_idx: usize, src: &[u8]) {
        let Some(plaintext) = self.core.process_codec_inbound(driver, slot_idx, src) else {
            return;
        };
        self.ingest_plaintext(driver, slot_idx, &plaintext);
        self.core.install_plaintext(slot_idx, plaintext);
    }

    fn ingest_plaintext(&mut self, driver: &mut B::Driver, slot_idx: usize, src: &[u8]) {
        let ingress_kind = match &self.core.slots[slot_idx].user.get().ingress {
            SlotIngress::Spill { .. } => 1u8,
            SlotIngress::RequestStream(_) => 2u8,
            SlotIngress::Idle => 0u8,
        };
        match ingress_kind {
            1 => {
                self.spill_push(slot_idx, src);
                return;
            }
            2 => {
                self.request_stream_push(driver, slot_idx, src);
                return;
            }
            _ => {}
        }

        let head = self.core.slots[slot_idx].conn.ingress_head as usize;
        let parked = self.core.slots[slot_idx]
            .conn
            .flags
            .contains(ConnFlags::PARKED);

        if head == 0 && !parked {
            let consumed = self.parse_and_dispatch_slice(slot_idx, src);
            let spill_in_progress = self.core.slots[slot_idx].user.get().ingress.is_spill();
            if !spill_in_progress {
                let remaining = &src[consumed..];
                let rl = remaining.len();
                if rl > 0 {
                    let entry = &mut self.core.slots[slot_idx];
                    let n = rl.min(entry.buf.ingress_buf.len());
                    entry.buf.ingress_buf[..n].copy_from_slice(&remaining[..n]);
                    entry.conn.ingress_head = n as u16;
                    entry.conn.ingress_tail = 0;
                }
            }
        } else {
            {
                let entry = &mut self.core.slots[slot_idx];
                let head = entry.conn.ingress_head as usize;
                let room = entry.buf.ingress_buf.len().saturating_sub(head);
                let n = src.len().min(room);
                if n > 0 {
                    entry.buf.ingress_buf[head..head + n].copy_from_slice(&src[..n]);
                    entry.conn.ingress_head = (head + n) as u16;
                }
            }
            self.parse_and_dispatch(slot_idx);
            self.try_dispatch_spilled(slot_idx);
        }
    }

    pub(super) fn parse_and_dispatch_slice(&mut self, slot_idx: usize, src: &[u8]) -> usize {
        if self.core.slots[slot_idx]
            .conn
            .flags
            .contains(ConnFlags::PARKED)
        {
            return 0;
        }
        let CoreSlot {
            conn, buf, user, ..
        } = &mut self.core.slots[slot_idx];
        let user = user.get_mut();
        let mut ctx = ParseCtx {
            protocol: &mut self.direction.protocol,
            conn,
            buf,
            user,
            slot_idx,
        };
        let mut cursor = 0usize;
        while cursor < src.len() {
            match ctx.parse_one(&src[cursor..]) {
                ParseStep::Continue(n) => cursor += n,
                ParseStep::Stop(n) => {
                    cursor += n;
                    break;
                }
            }
        }
        cursor
    }

    pub(super) fn parse_and_dispatch(&mut self, slot_idx: usize) {
        if self.core.slots[slot_idx]
            .conn
            .flags
            .contains(ConnFlags::PARKED)
        {
            return;
        }
        let CoreSlot {
            conn, buf, user, ..
        } = &mut self.core.slots[slot_idx];
        let user = user.get_mut();
        let ingress_ptr = buf.ingress_buf.as_ptr();
        let mut ctx = ParseCtx {
            protocol: &mut self.direction.protocol,
            conn,
            buf,
            user,
            slot_idx,
        };
        loop {
            let tail = ctx.conn.ingress_tail as usize;
            let head = ctx.conn.ingress_head as usize;
            if head <= tail {
                break;
            }
            let view: &[u8] =
                unsafe { std::slice::from_raw_parts(ingress_ptr.add(tail), head - tail) };
            match ctx.parse_one(view) {
                ParseStep::Continue(n) => ctx.conn.ingress_tail = (tail + n) as u16,
                ParseStep::Stop(n) => {
                    ctx.conn.ingress_tail = (tail + n) as u16;
                    break;
                }
            }
        }
        if ctx.conn.ingress_tail == ctx.conn.ingress_head {
            ctx.conn.ingress_head = 0;
            ctx.conn.ingress_tail = 0;
        }
    }

    #[cold]
    #[inline(never)]
    fn spill_push(&mut self, slot_idx: usize, src: &[u8]) {
        let trigger = {
            let user = self.core.slots[slot_idx].user.get_mut();
            let SlotIngress::Spill {
                buf,
                filled,
                target,
            } = &mut user.ingress
            else {
                return;
            };
            let needed = (*target as usize).saturating_sub(*filled as usize);
            let take = src.len().min(needed);
            buf.extend_from_slice(&src[..take]);
            *filled = filled.saturating_add(take as u32);
            *filled >= *target
        };
        if trigger {
            self.dispatch_spilled(slot_idx);
        }
    }

    #[cold]
    fn request_stream_push(&mut self, driver: &mut B::Driver, slot_idx: usize, src: &[u8]) {
        let (consumed, eof) = {
            let user = self.core.slots[slot_idx].user.get();
            match &user.ingress {
                SlotIngress::RequestStream(handle) => {
                    let consumed = handle.push_slice(src);
                    (consumed, handle.is_complete())
                }
                _ => return,
            }
        };
        if consumed == 0 && !src.is_empty() && !eof {
            self.core.free_slot(driver, slot_idx);
            return;
        }
        if !eof {
            return;
        }
        self.core.slots[slot_idx].user.get_mut().ingress = SlotIngress::Idle;
        let remaining = &src[consumed..];
        if remaining.is_empty() {
            return;
        }
        let entry = &mut self.core.slots[slot_idx];
        let head = entry.conn.ingress_head as usize;
        let room = entry.buf.ingress_buf.len().saturating_sub(head);
        let n = remaining.len().min(room);
        if n > 0 {
            entry.buf.ingress_buf[head..head + n].copy_from_slice(&remaining[..n]);
            entry.conn.ingress_head = (head + n) as u16;
        }
    }

    #[inline(always)]
    fn try_dispatch_spilled(&mut self, slot_idx: usize) {
        let user = self.core.slots[slot_idx].user.get();
        let SlotIngress::Spill { filled, target, .. } = &user.ingress else {
            return;
        };
        if *filled < *target {
            return;
        }
        self.dispatch_spilled(slot_idx);
    }

    #[cold]
    #[inline(never)]
    fn dispatch_spilled(&mut self, slot_idx: usize) {
        let spill_box = {
            let user = self.core.slots[slot_idx].user.get_mut();
            let prev = std::mem::replace(&mut user.ingress, SlotIngress::Idle);
            match prev {
                SlotIngress::Spill { buf, .. } => buf,
                other => {
                    user.ingress = other;
                    return;
                }
            }
        };
        let spill: BytesMut = *spill_box;
        let req_bytes: &[u8] = spill.as_ref();
        let parsed_head = {
            let state = &mut self.core.slots[slot_idx].user.get_mut().protocol_state;
            match self.direction.protocol.parse(state, req_bytes) {
                Some((head, _, _)) => head,
                None => return,
            }
        };

        let egress_busy = !self.core.slots[slot_idx].user.get().egress.is_idle();
        if !self.core.slots[slot_idx].conn.pending_body_ptr.is_null() || egress_busy {
            return;
        }
        let CoreSlot {
            conn, buf, user, ..
        } = &mut self.core.slots[slot_idx];
        let user = user.get_mut();
        let wh = conn.write_head as usize;
        if buf.write_buf.as_slice().len().saturating_sub(wh) < WRITE_ROOM_MIN {
            return;
        }
        let mut ctx = ParseCtx {
            protocol: &mut self.direction.protocol,
            conn,
            buf,
            user,
            slot_idx,
        };
        let write_slice: &mut [u8] = &mut ctx.buf.write_buf.as_mut_slice()[wh..];
        let outcome = ctx.protocol.handle(req_bytes, parsed_head, write_slice);
        let _ = ctx.apply_outcome(outcome, wh, req_bytes.len());
    }
}
