use dambi::Bytes;

use crate::CodecLayer;
use crate::SlotId;
use crate::pool::client::state::CLIENT_IOV_CAP;
use crate::protocol::client::ClientProtocol;
use crate::ring::frame::FrameRing;
use crate::ring::send::SendRing;

pub struct Egress<'a, L: CodecLayer> {
    inner: &'a mut SendRing<CLIENT_IOV_CAP>,
    codec_state: &'a mut L::ConnState,
}

impl<'a, L: CodecLayer> Egress<'a, L> {
    #[inline(always)]
    pub fn push(&mut self, bytes: Bytes) {
        self.inner.push(bytes);
    }

    #[inline(always)]
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    /// Mutable access to the paired codec's connection state.
    ///
    /// The returned borrow is reborrowed from `&mut self`, so it must be
    /// released (use ends) before any other `Egress` method can be called.
    /// Typical usage is sequential: trigger a codec transition, then enqueue.
    ///
    /// For the [`Plain`](crate::Plain) layer, `L::ConnState = ()` and this
    /// returns a ZST `&mut ()` — zero overhead.
    #[inline(always)]
    pub fn codec_state(&mut self) -> &mut L::ConnState {
        self.codec_state
    }
}

pub struct ClientSessionGeneric<P: ClientProtocol, const ING: usize, const IOV: usize> {
    pub(crate) ingress: FrameRing<ING>,
    pub(crate) egress: SendRing<IOV>,
    pub(crate) state: P::ConnState,
}

impl<P: ClientProtocol, const ING: usize, const IOV: usize> Default
    for ClientSessionGeneric<P, ING, IOV>
{
    #[inline(always)]
    fn default() -> Self {
        Self {
            ingress: FrameRing::new(),
            egress: SendRing::new(),
            state: P::ConnState::default(),
        }
    }
}

impl<P: ClientProtocol, const ING: usize> ClientSessionGeneric<P, ING, CLIENT_IOV_CAP> {
    #[inline(always)]
    pub fn enqueue(&mut self, bytes: Bytes) {
        self.egress.push(bytes);
    }

    #[inline(always)]
    pub fn state(&self) -> &P::ConnState {
        &self.state
    }

    #[inline(always)]
    pub fn state_mut(&mut self) -> &mut P::ConnState {
        &mut self.state
    }

    #[inline(always)]
    pub fn egress_is_empty(&self) -> bool {
        self.egress.is_empty()
    }

    pub(crate) fn on_connect(
        &mut self,
        proto: &mut P,
        conn_id: SlotId,
        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
    ) {
        let Self { state, egress, .. } = self;
        proto.on_connect(
            conn_id,
            state,
            Egress {
                inner: egress,
                codec_state,
            },
        );
    }

    pub(crate) fn on_disconnect(
        &mut self,
        proto: &mut P,
        conn_id: SlotId,
        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
    ) {
        let Self { state, egress, .. } = self;
        proto.on_disconnect(
            conn_id,
            state,
            Egress {
                inner: egress,
                codec_state,
            },
        );
    }

    pub(crate) fn ingest(
        &mut self,
        src: &[u8],
        proto: &mut P,
        conn_id: SlotId,
        codec_state: &mut <P::CodecLayer as CodecLayer>::ConnState,
    ) {
        let _ = self.ingress.append(src);
        let Self {
            ingress,
            state,
            egress,
            ..
        } = self;
        ingress.drain(|buf| {
            let (head, consumed) = proto.parse(state, buf)?;
            proto.on_response(
                conn_id,
                state,
                head,
                Egress {
                    inner: &mut *egress,
                    codec_state: &mut *codec_state,
                },
            );
            Some(consumed)
        });
    }

    #[inline(always)]
    pub(crate) fn fill_msghdr(&mut self, bytes_cap: usize) -> Option<&libc::msghdr> {
        self.egress.fill_msghdr(bytes_cap)
    }

    #[inline(always)]
    pub(crate) fn ack_send(&mut self, n: u32) {
        self.egress.ack(n);
    }
}
