saddle-framework 0.2.0

use std::time::Duration;

use saddle_runtime::compiled_route::{
    ClassifiedCompiledRouteAdapter, OfficialHttpRouteCoordinator, OfficialHttpTcpOutcome,
    OfficialTcpAttemptProfile, WaitingHttpTcp,
};
use saddle_service::internal::{RegisteredRouteExecutionProof, RegistryError};
use tokio::{io::AsyncReadExt, net::TcpStream, time::timeout};

use super::{
    bootstrap::GeneratedCompiledAdapter,
    framing::{Http1Framing, Http1ResponsePlan},
    parser::{FrozenHead, HeadParser, ParserError},
};

/// One framework-owned lookup surface. The production implementation is the
/// frozen Service capability; this trait is private so business code cannot
/// forge route proofs or invoke the entry coordinator.
pub(crate) trait FrozenRouteLookup {
    fn lookup(
        &self,
        route: &[u8],
        declared_length: usize,
    ) -> Result<RegisteredRouteExecutionProof, RegistryError>;
}

impl<A> FrozenRouteLookup for A
where
    A: GeneratedCompiledAdapter,
{
    fn lookup(
        &self,
        route: &[u8],
        declared_length: usize,
    ) -> Result<RegisteredRouteExecutionProof, RegistryError> {
        self.lookup_generated(route, declared_length)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum Http1EntryError {
    HeadDeadline,
    Disconnect,
    Parse(ParserError),
    UnknownRoute,
}

pub(crate) enum Http1EntryOutcome<P, C> {
    Runtime(OfficialHttpTcpOutcome<P, C, Http1ResponsePlan>),
    Rejected(Http1EntryError, TcpStream),
}

/// Reads exactly through CRLFCRLF, one byte at a time. Consequently the
/// socket handed to Runtime still owns every body byte and Transport has no
/// user-space body/prebuffer to retain while Admission returns Registered.
async fn read_frozen_head(
    socket: &mut TcpStream,
    deadline: Duration,
) -> Result<FrozenHead, Http1EntryError> {
    timeout(deadline, async {
        let mut parser = HeadParser::new();
        loop {
            let mut byte = [0_u8; 1];
            socket
                .read_exact(&mut byte)
                .await
                .map_err(|_| Http1EntryError::Disconnect)?;
            if let Some(head) = parser.push_byte(byte[0]).map_err(Http1EntryError::Parse)? {
                return Ok(head);
            }
        }
    })
    .await
    .map_err(|_| Http1EntryError::HeadDeadline)?
}

/// The only production-oriented head -> Service proof -> Runtime attempt
/// assembly. Transport never receives an envelope, request account, DB permit,
/// ManagedBytes or business Future.
pub(crate) async fn attempt_connection<A, L>(
    coordinator: &mut OfficialHttpRouteCoordinator<A, Http1Framing>,
    lookup: &L,
    context: A::Context,
    mut socket: TcpStream,
    head_deadline: Duration,
    mut profile: OfficialTcpAttemptProfile,
) -> Http1EntryOutcome<A::Proof, A::Context>
where
    A: ClassifiedCompiledRouteAdapter<Proof = RegisteredRouteExecutionProof>,
    A::Context: Copy,
    L: FrozenRouteLookup,
{
    let head = match read_frozen_head(&mut socket, head_deadline).await {
        Ok(head) => head,
        Err(error) => return Http1EntryOutcome::Rejected(error, socket),
    };
    let proof = match lookup.lookup(head.route(), head.content_length()) {
        Ok(proof) => proof,
        Err(_) => return Http1EntryOutcome::Rejected(Http1EntryError::UnknownRoute, socket),
    };
    profile.expected_body_bytes = head.content_length();
    Http1EntryOutcome::Runtime(coordinator.attempt(
        proof,
        context,
        Http1ResponsePlan::connection_close(),
        socket,
        profile,
    ))
}

/// Retry preserves the exact proof/context/framing plan/socket retained by the
/// Runtime registration. It cannot parse, look up, or replace any input.
pub(crate) fn retry_waiting<A>(
    coordinator: &mut OfficialHttpRouteCoordinator<A, Http1Framing>,
    waiting: WaitingHttpTcp<A::Proof, A::Context, Http1ResponsePlan>,
    profile: OfficialTcpAttemptProfile,
) -> Option<OfficialHttpTcpOutcome<A::Proof, A::Context, Http1ResponsePlan>>
where
    A: ClassifiedCompiledRouteAdapter,
    A::Context: Copy,
{
    coordinator
        .claim_retry()
        .map(|token| coordinator.retry(waiting, token, profile))
}

/// Shutdown/cancellation consumes the one Waiting owner and closes its socket
/// at this boundary. The returned boolean is Runtime's generation removal
/// evidence, not a Transport-maintained shadow counter.
pub(crate) fn cancel_waiting<A>(
    coordinator: &OfficialHttpRouteCoordinator<A, Http1Framing>,
    waiting: WaitingHttpTcp<A::Proof, A::Context, Http1ResponsePlan>,
) -> bool
where
    A: ClassifiedCompiledRouteAdapter,
    A::Context: Copy,
{
    let (socket, _proof, _context, _plan, removed) = coordinator.cancel(waiting);
    drop(socket);
    removed
}

#[cfg(test)]
mod tests {
    use tokio::{io::AsyncWriteExt, net::TcpListener};

    use super::*;

    async fn socket_pair() -> (TcpStream, TcpStream) {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let client = TcpStream::connect(address);
        let (client, server) = tokio::join!(client, listener.accept());
        (client.unwrap(), server.unwrap().0)
    }

    #[tokio::test]
    async fn real_socket_head_freeze_does_not_materialize_body() {
        let (mut client, mut server) = socket_pair().await;
        client
            .write_all(
                b"POST /plain HTTP/1.1\r\nHost: api.example:443\r\nContent-Length: 4\r\nConnection: close\r\n\r\nbody",
            )
            .await
            .unwrap();

        let head = read_frozen_head(&mut server, Duration::from_secs(1))
            .await
            .unwrap();
        assert_eq!(head.route(), b"/plain");
        assert_eq!(head.content_length(), 4);

        let mut body = [0_u8; 4];
        server.read_exact(&mut body).await.unwrap();
        assert_eq!(&body, b"body");
    }

    #[tokio::test]
    async fn head_deadline_releases_without_body_read() {
        let (_client, mut server) = socket_pair().await;
        assert_eq!(
            read_frozen_head(&mut server, Duration::from_millis(10)).await,
            Err(Http1EntryError::HeadDeadline)
        );
    }
}