//! The connection actor — one per accepted TCP connection.
//!
//! Runs the HTTP/1.1 request loop:
//!
//!   loop {
//!       read bytes → parse → build Conn
//!       pipeline.run(conn)               // inline; no spawn
//!       write response
//!       if !keep_alive { break }
//!   }
//!
//! Everything in here happens in one smarm green thread. The actor parks on
//! `wait_readable` between bytes and `wait_writable` during slow writes;
//! during those parks, other connection actors progress freely.

use crate::conn::{Body, Conn, HttpVersion, RespBody, StreamBody};
use crate::conn_registry::{Cast, ConnRegistry, DeregisterGuard};
use crate::net::OwnedFd;
use crate::parser::{self, ParseError};
use crate::plug::Pipeline;

use smarm::ServerRef;

use std::io::{self, ErrorKind};
use std::os::fd::RawFd;
use std::time::{Duration, Instant};

// ---------------------------------------------------------------------------
// Limits
// ---------------------------------------------------------------------------

/// Per-connection settings the connection actor needs to honour.
#[derive(Clone, Copy, Debug)]
pub struct ConnLimits {
    pub max_headers:        usize,
    pub initial_read_buf:   usize,
    /// Hard cap on the request head to bound buffer growth. 64 KiB is
    /// well over Apache's 8 KiB default; protects against pathological
    /// clients streaming headers forever.
    pub max_head_bytes:     usize,
    /// Hard cap on Content-Length we'll accept. 16 MiB is enough for a CRUD
    /// example; configurable in `Config`.
    pub max_body_bytes:     usize,
    /// Idle budget between requests: how long we'll park waiting for the
    /// FIRST byte of a request (including the first request on a fresh
    /// connection). Expiry closes the connection silently — nothing is
    /// owed to a client that isn't talking.
    pub keep_alive_timeout: Duration,
    /// Per-request wall-clock budget, measured from the first byte of a
    /// request until the request (head + body) is fully read. Expiry
    /// mid-head gets a best-effort 408; expiry mid-body just closes.
    /// Pipeline run time is NOT covered — that's the handler's business.
    /// Covers the READ phase only; the write phase has its own
    /// per-write budget (`write_timeout`) so a streaming response can
    /// legitimately outlive any whole-request clock.
    pub request_timeout:    Duration,
    /// Per-write budget for response bytes: every `write_all` (the fixed
    /// head+body, and each streamed chunk) must complete within this.
    /// A client that stops reading mid-response is dropped when its
    /// socket buffer fills and a write stalls past the budget.
    pub write_timeout:      Duration,
}

impl Default for ConnLimits {
    fn default() -> Self {
        Self {
            max_headers:        64,
            initial_read_buf:   8 * 1024,
            max_head_bytes:     64 * 1024,
            max_body_bytes:     16 * 1024 * 1024,
            keep_alive_timeout: Duration::from_secs(60),
            request_timeout:    Duration::from_secs(30),
            write_timeout:      Duration::from_secs(30),
        }
    }
}

// ---------------------------------------------------------------------------
// run_connection — entry point spawned by the listener actor.
// ---------------------------------------------------------------------------

pub fn run_connection(
    fd: OwnedFd,
    pipeline: Pipeline,
    limits: ConnLimits,
    registry: ServerRef<ConnRegistry>,
) {
    // The OwnedFd cleans up via Drop on any exit path (panic, error, or
    // normal close). No explicit close calls below.
    let raw = fd.as_raw();
    let mut buf: Vec<u8> = Vec::with_capacity(limits.initial_read_buf);

    // Self-register (initially idle: no request head parsed yet) and arm
    // the deregistration guard. Both casts come from this actor, so
    // Started always precedes Ended in the registry's inbox — see
    // conn_registry module docs for why the listener must not do this.
    let me = smarm::self_pid();
    let _ = registry.cast(Cast::ConnStarted(me));
    let _guard = DeregisterGuard::new(registry.clone(), me, Cast::ConnEnded);

    loop {
        // ----- 1. Read until we have a full request head. -----
        // We are idle until a head parses: stoppable by a draining
        // registry while parked here.
        let (parsed, request_deadline) = match read_head(raw, &mut buf, &limits) {
            Ok(p) => p,
            Err(ReadHeadErr::ClientClosed) => {
                // Clean EOF between requests (or before any request). Normal.
                return;
            }
            Err(ReadHeadErr::IdleTimeout) => {
                // keep_alive_timeout expired waiting for the first byte of
                // a request. Nothing is owed; close silently.
                return;
            }
            Err(ReadHeadErr::RequestTimeout) => {
                // request_timeout expired mid-head (slowloris and friends).
                // Best-effort 408 WITHOUT parking on writability — a client
                // that stalls reads must not defeat the timeout by making
                // the 408 write park forever.
                try_write_once(raw, b"HTTP/1.1 408 Request Timeout\r\ncontent-length: 0\r\nconnection: close\r\n\r\n");
                return;
            }
            Err(ReadHeadErr::Io(_)) => {
                // Network error. Best-effort close; we're done.
                return;
            }
            Err(ReadHeadErr::Parse(e)) => {
                emit_error_response(raw, &e, Instant::now() + limits.write_timeout);
                return;
            }
        };
        let _ = registry.cast(Cast::ConnBusy(me));

        // ----- 2. Read body. -----
        // Content-Length pre-check only applies to fixed bodies; a chunked
        // body is bounded incrementally by the decoder.
        let body_len = parsed.content_length.unwrap_or(0);
        if body_len > limits.max_body_bytes {
            let _ = write_all(
                raw,
                b"HTTP/1.1 413 Payload Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
                Instant::now() + limits.write_timeout,
            );
            return;
        }

        // If client sent `Expect: 100-continue`, emit it before reading the
        // body. RFC 7231 §5.1.1. We don't gate on app logic here; v1 always
        // accepts.
        if parsed.expect_100 {
            if write_all(raw, b"HTTP/1.1 100 Continue\r\n\r\n", Instant::now() + limits.write_timeout).is_err() {
                return;
            }
        }

        // `consumed_past_head`: how many RAW bytes of `buf` past the head
        // this request's body occupied — for chunked bodies that is framing
        // included, NOT the decoded length. The keep-alive drain at the
        // bottom of the loop must drop exactly this much to land on the
        // next pipelined request.
        let (body, consumed_past_head) = if parsed.chunked {
            match read_chunked_body(raw, &mut buf, parsed.head_len, &limits, request_deadline) {
                Ok(ok) => ok,
                Err(ChunkedBodyErr::TooLarge) => {
                    let _ = write_all(
                        raw,
                        b"HTTP/1.1 413 Payload Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
                        Instant::now() + limits.write_timeout,
                    );
                    return;
                }
                Err(ChunkedBodyErr::Malformed) => {
                    let _ = write_all(
                        raw,
                        b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
                        Instant::now() + limits.write_timeout,
                    );
                    return;
                }
                // Timeout mid-body (and any other io error) -> just close.
                Err(ChunkedBodyErr::Io(_)) => return,
            }
        } else {
            match read_body(raw, &mut buf, parsed.head_len, body_len, request_deadline) {
                Ok(b) => (b, body_len),
                // Timeout mid-body (and any other body io error) -> just
                // close; there's no point talking HTTP to a client this far
                // gone.
                Err(_) => return,
            }
        };

        let keep_alive = parsed.keep_alive;
        let version    = parsed.version;
        let head_len   = parsed.head_len;
        let conn       = parser::build_conn(parsed, Body::from_bytes(body));

        // ----- 3. Run the pipeline. -----
        // Catch panics at the actor boundary — a panicking handler should
        // not take down the whole connection silently with no response.
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            pipeline.run(conn)
        }));

        let mut response_conn = match result {
            Ok(c)  => c,
            Err(_) => {
                // Distinguish a genuine handler panic from smarm's stop
                // sentinel, which is also a panic payload and which this
                // catch_unwind would otherwise swallow — turning a
                // graceful stop into a 500-and-keep-running. The stop
                // flag is persistent (not consumed by raising the
                // sentinel), so if we were stopped this re-raises it
                // here, outside the catch, and we unwind properly (the
                // fd and registry guards clean up).
                smarm::preempt::check_cancelled();

                // Compose a 500 manually; the original Conn was moved into
                // the closure.
                let mut c = Conn::new();
                c.version = version;
                c.put_status(500).put_header("content-length", "0")
            }
        };

        // If no plug touched status, that's a configuration error (no router
        // matched, no default handler). Emit 404.
        if response_conn.status.is_none() {
            response_conn = response_conn.put_status(404)
                .put_body(RespBody::Empty);
        }

        // ----- 3.5 WebSocket upgrade short-circuit. -----
        // An accepted handshake (marker + 101) ends HTTP on this socket:
        // write the 101 head and leave the request loop. Chunk 3 (duplex
        // wiring) takes the socket over right here; until then leaving
        // the loop closes it via OwnedFd::drop — the 101 is still the
        // honest, testable wire artefact. The status check is defensive:
        // a post-handler plug that clobbered the 101 forfeits the
        // upgrade and falls through to plain HTTP.
        if response_conn.upgrade.is_some() && response_conn.status == Some(101) {
            let head = parser::serialise_response(&response_conn, true);
            let _ = write_all(raw, &head, Instant::now() + limits.write_timeout);
            return;
        }

        // ----- 4. Write the response. -----
        // A Stream body on HTTP/1.0 has no chunked framing: the body is
        // delimited by EOF, so keep-alive is forced off for this response
        // (and `connection: close` goes on the wire).
        let is_stream = matches!(response_conn.resp_body, RespBody::Stream(_));
        let keep_alive = keep_alive
            && !(is_stream && version == HttpVersion::Http10);

        let head_bytes = parser::serialise_response(&response_conn, keep_alive);
        if write_all(raw, &head_bytes, Instant::now() + limits.write_timeout).is_err() {
            return;
        }

        if let RespBody::Stream(stream) = response_conn.resp_body {
            let chunked = version == HttpVersion::Http11;
            if pump_stream(raw, stream, chunked, limits.write_timeout).is_err() {
                return;
            }
            if !chunked {
                // EOF delimits the HTTP/1.0 stream body.
                return;
            }
        }

        // ----- 5. Loop or close. -----
        if !keep_alive {
            return;
        }

        // Response is on the wire; nothing is owed. Going idle here makes
        // us stoppable by a draining registry while we park for the next
        // keep-alive request. (If the next request is already pipelined in
        // `buf`, the very next read_head parses it without parking and we
        // go Busy again — a draining registry's request_stop may still
        // catch us, which is acceptable: drain means no new work.)
        let _ = registry.cast(Cast::ConnIdle(me));

        // Drop the request bytes (head + raw body framing) from `buf`;
        // anything past them is the start of the next pipelined request.
        let consumed = head_len + consumed_past_head;
        buf.drain(..consumed);
    }
}

// ---------------------------------------------------------------------------
// read_head
// ---------------------------------------------------------------------------

#[allow(dead_code)] // io::Error is captured for future logging
enum ReadHeadErr {
    ClientClosed,
    /// keep_alive_timeout expired while waiting for the first byte of a
    /// request. Close silently.
    IdleTimeout,
    /// request_timeout expired after the request had started arriving.
    /// Best-effort 408.
    RequestTimeout,
    Io(io::Error),
    Parse(ParseError),
}

/// Read until `parse_head` succeeds or fails definitively. `buf` may already
/// contain leftover bytes from a previous keep-alive cycle; we try to parse
/// those before reading more from the socket.
///
/// Two wall-clock budgets govern the waits (each wait uses whichever budget
/// is currently active):
///
/// - while `buf` is empty and nothing has arrived, we are *idle* and the
///   wait is bounded by `keep_alive_timeout`;
/// - the instant the request has started (first byte read, or pipelined
///   bytes already in `buf` at entry), the *request* clock starts: an
///   `Instant` deadline of `request_timeout` from that moment, which also
///   covers body reads — it is returned alongside the parsed head so the
///   caller can thread it into `read_body`.
fn read_head(
    fd: RawFd,
    buf: &mut Vec<u8>,
    limits: &ConnLimits,
) -> Result<(parser::ParsedHead, Instant), ReadHeadErr> {
    let entry = Instant::now();
    let idle_deadline = entry + limits.keep_alive_timeout;
    // Pipelined leftovers count as a started request.
    let mut request_deadline: Option<Instant> = if buf.is_empty() {
        None
    } else {
        Some(entry + limits.request_timeout)
    };

    loop {
        // Try to parse what we already have. On the first iteration of a
        // fresh keep-alive cycle, `buf` may already hold the next request.
        if !buf.is_empty() {
            match parser::parse_head(buf, limits.max_headers) {
                Ok(h)  => {
                    let deadline = request_deadline
                        .unwrap_or_else(|| Instant::now() + limits.request_timeout);
                    return Ok((h, deadline));
                }
                Err(ParseError::Incomplete) => {} // need more bytes
                Err(e) => return Err(ReadHeadErr::Parse(e)),
            }
        }

        if buf.len() >= limits.max_head_bytes {
            return Err(ReadHeadErr::Parse(ParseError::TooManyHeaders));
        }

        // Read more, bounded by whichever budget is active.
        let deadline = request_deadline.unwrap_or(idle_deadline);
        match read_some(fd, buf, limits.initial_read_buf, deadline) {
            Ok(0)  => return Err(ReadHeadErr::ClientClosed),
            Ok(_)  => {
                if request_deadline.is_none() {
                    // First byte(s) of this request: the request clock
                    // starts now.
                    request_deadline = Some(Instant::now() + limits.request_timeout);
                }
            }
            Err(e) if e.kind() == ErrorKind::TimedOut => {
                return Err(if request_deadline.is_some() {
                    ReadHeadErr::RequestTimeout
                } else {
                    ReadHeadErr::IdleTimeout
                });
            }
            Err(e) => return Err(ReadHeadErr::Io(e)),
        }
    }
}

// ---------------------------------------------------------------------------
// read_body
// ---------------------------------------------------------------------------

fn read_body(
    fd: RawFd,
    buf: &mut Vec<u8>,
    head_len: usize,
    body_len: usize,
    deadline: Instant,
) -> io::Result<Vec<u8>> {
    // Bytes already in `buf` past the head belong to the body.
    let already = buf.len().saturating_sub(head_len);
    let need    = body_len.saturating_sub(already);

    if need == 0 {
        // We have the full body in `buf` already. Extract a copy; `buf` is
        // drained later in the connection loop.
        return Ok(buf[head_len..head_len + body_len].to_vec());
    }

    // Read until we have the rest, on the same request budget that the
    // head was read under.
    let mut total_read = already;
    while total_read < body_len {
        match read_some(fd, buf, 8 * 1024, deadline) {
            Ok(0)  => return Err(io::Error::new(ErrorKind::UnexpectedEof, "client closed during body")),
            Ok(n)  => total_read += n,
            Err(e) => return Err(e),
        }
    }
    Ok(buf[head_len..head_len + body_len].to_vec())
}

// ---------------------------------------------------------------------------
// read_chunked_body — incremental chunked transfer-decoding (request side).
// ---------------------------------------------------------------------------
//
// Decodes `Transfer-Encoding: chunked` from `buf[head_len..]`, reading more
// from the socket as needed on the SAME request deadline the head was read
// under. Returns (decoded_body, raw_bytes_consumed_past_head) — the raw
// count includes all framing and the trailer section, so the caller's
// keep-alive drain lands exactly on the next pipelined request.
//
// Bounds: the DECODED size is capped at max_body_bytes (-> TooLarge/413);
// a single size line (incl. chunk extensions, which are ignored) is capped
// at MAX_SIZE_LINE and the trailer section at MAX_TRAILER_BYTES (-> 
// Malformed/400) so framing spam can't grow `buf` unboundedly. Trailers
// are consumed and discarded — nothing in the pipeline wants them yet.

const MAX_SIZE_LINE:     usize = 128;
const MAX_TRAILER_BYTES: usize = 8 * 1024;

#[allow(dead_code)] // io::Error is captured for future logging
enum ChunkedBodyErr {
    Io(io::Error),
    Malformed,
    TooLarge,
}

fn read_chunked_body(
    fd: RawFd,
    buf: &mut Vec<u8>,
    head_len: usize,
    limits: &ConnLimits,
    deadline: Instant,
) -> Result<(Vec<u8>, usize), ChunkedBodyErr> {
    // Ensure `buf` holds at least `until` bytes, reading on the request
    // deadline. Io(TimedOut) on expiry, UnexpectedEof on early close.
    fn fill_to(
        fd: RawFd,
        buf: &mut Vec<u8>,
        until: usize,
        deadline: Instant,
    ) -> Result<(), ChunkedBodyErr> {
        while buf.len() < until {
            match read_some(fd, buf, 8 * 1024, deadline) {
                Ok(0) => {
                    return Err(ChunkedBodyErr::Io(io::Error::new(
                        ErrorKind::UnexpectedEof,
                        "client closed during chunked body",
                    )))
                }
                Ok(_) => {}
                Err(e) => return Err(ChunkedBodyErr::Io(e)),
            }
        }
        Ok(())
    }

    // Find "\r\n" in buf[from..], reading more as needed; the line may be
    // at most `max_line` bytes (terminator excluded). Returns the index of
    // the '\r'.
    fn find_crlf(
        fd: RawFd,
        buf: &mut Vec<u8>,
        from: usize,
        max_line: usize,
        deadline: Instant,
    ) -> Result<usize, ChunkedBodyErr> {
        let mut scan = from;
        loop {
            while scan + 1 < buf.len() {
                if buf[scan] == b'\r' && buf[scan + 1] == b'\n' {
                    return Ok(scan);
                }
                scan += 1;
                if scan - from > max_line {
                    return Err(ChunkedBodyErr::Malformed);
                }
            }
            fill_to(fd, buf, buf.len() + 1, deadline)?;
        }
    }

    let mut pos = head_len;
    let mut decoded: Vec<u8> = Vec::new();

    loop {
        // ----- size line: HEX[;extensions]\r\n -----
        let line_end = find_crlf(fd, buf, pos, MAX_SIZE_LINE, deadline)?;
        let line = &buf[pos..line_end];
        let size_str = match line.iter().position(|&b| b == b';') {
            Some(i) => &line[..i], // chunk extensions: ignored
            None    => line,
        };
        let size_str = std::str::from_utf8(size_str)
            .map_err(|_| ChunkedBodyErr::Malformed)?
            .trim();
        let size = usize::from_str_radix(size_str, 16)
            .map_err(|_| ChunkedBodyErr::Malformed)?;
        pos = line_end + 2;

        if size == 0 {
            // ----- trailer section: zero or more header lines, then CRLF -----
            let trailer_start = pos;
            loop {
                let t_end = find_crlf(fd, buf, pos, MAX_SIZE_LINE.max(1024), deadline)?;
                let empty = t_end == pos;
                pos = t_end + 2;
                if empty {
                    return Ok((decoded, pos - head_len));
                }
                if pos - trailer_start > MAX_TRAILER_BYTES {
                    return Err(ChunkedBodyErr::Malformed);
                }
            }
        }

        if decoded.len() + size > limits.max_body_bytes {
            return Err(ChunkedBodyErr::TooLarge);
        }

        // ----- chunk payload + trailing CRLF -----
        fill_to(fd, buf, pos + size + 2, deadline)?;
        decoded.extend_from_slice(&buf[pos..pos + size]);
        if &buf[pos + size..pos + size + 2] != b"\r\n" {
            return Err(ChunkedBodyErr::Malformed);
        }
        pos += size + 2;
    }
}

// ---------------------------------------------------------------------------
// read_some — single epoll-park + read loop, bounded by a deadline.
// ---------------------------------------------------------------------------
//
// Appends what it reads onto `buf`. Returns bytes read, 0 for EOF,
// `ErrorKind::TimedOut` when `deadline` passes before the fd turns
// readable, or the last io error.

fn read_some(
    fd: RawFd,
    buf: &mut Vec<u8>,
    chunk: usize,
    deadline: Instant,
) -> io::Result<usize> {
    // Loop to absorb EAGAIN: a readable wakeup followed by EAGAIN is
    // possible (signal race, etc). Re-park and retry rather than returning
    // 0 (which would be confused with EOF by callers). The deadline is an
    // Instant, so spurious wakes don't reset the budget.
    loop {
        let remaining = deadline.saturating_duration_since(Instant::now());
        if remaining.is_zero() {
            return Err(io::Error::new(ErrorKind::TimedOut, "read deadline elapsed"));
        }
        if !smarm::wait_readable_timeout(fd, remaining)? {
            return Err(io::Error::new(ErrorKind::TimedOut, "read deadline elapsed"));
        }

        let start = buf.len();
        buf.resize(start + chunk, 0);

        let n = unsafe {
            libc::read(fd, buf.as_mut_ptr().add(start) as *mut _, chunk)
        };

        if n < 0 {
            let err = io::Error::last_os_error();
            buf.truncate(start);
            if err.kind() == ErrorKind::WouldBlock || err.kind() == ErrorKind::Interrupted {
                continue;
            }
            return Err(err);
        }

        let n = n as usize;
        buf.truncate(start + n);
        return Ok(n); // n == 0 here is real EOF
    }
}

// ---------------------------------------------------------------------------
// try_write_once — single non-parking write attempt, result ignored.
// ---------------------------------------------------------------------------
//
// For best-effort farewells (the 408) to clients we've decided to drop:
// one non-blocking write syscall, no wait_writable park. A client that
// stalls its read side must not be able to keep this actor alive past its
// own timeout. The socket send buffer almost always has room for a
// one-liner, so in practice the 408 lands.

fn try_write_once(fd: RawFd, buf: &[u8]) {
    unsafe {
        let _ = libc::write(fd, buf.as_ptr() as *const _, buf.len());
    }
}

// ---------------------------------------------------------------------------
// write_all — robust write loop, bounded by a deadline.
// ---------------------------------------------------------------------------
//
// Mirrors read_some: each writability park is bounded by the remaining
// budget. `ErrorKind::TimedOut` when the deadline passes before the bytes
// are down — a client that stops reading must not pin this actor in
// wait_writable forever (the write-side twin of slowloris).

fn write_all(fd: RawFd, mut buf: &[u8], deadline: Instant) -> io::Result<()> {
    while !buf.is_empty() {
        // Park on writability before each syscall, bounded by the budget.
        let remaining = deadline.saturating_duration_since(Instant::now());
        if remaining.is_zero() {
            return Err(io::Error::new(ErrorKind::TimedOut, "write deadline elapsed"));
        }
        if !smarm::wait_writable_timeout(fd, remaining)? {
            return Err(io::Error::new(ErrorKind::TimedOut, "write deadline elapsed"));
        }

        let n = unsafe {
            libc::write(fd, buf.as_ptr() as *const _, buf.len())
        };
        if n < 0 {
            let err = io::Error::last_os_error();
            if err.kind() == ErrorKind::WouldBlock {
                continue; // spurious wake; retry
            }
            return Err(err);
        }
        if n == 0 {
            return Err(io::Error::new(ErrorKind::WriteZero, "write returned 0"));
        }
        buf = &buf[n as usize..];
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// pump_stream — drive a RespBody::Stream onto the wire.
// ---------------------------------------------------------------------------
//
// The pull side of the v0.3 streaming design: the handler's producer actor
// owns the Sender; this conn actor owns the socket and every write
// deadline. We park in `recv()` between chunks — that park is stoppable
// (a draining registry's `request_stop` unwinds us out of `park_current`
// via the stop sentinel; fd + registry guards clean up), so an infinite
// stream is force-stoppable at the drain deadline like any other in-flight
// request. End of stream is the channel closing: every Sender dropped.
//
// Each chunk gets a FRESH write_timeout budget — a stream is expected to
// outlive any whole-response clock; what is not tolerated is a single
// write stalling. On write failure we return Err: the conn loop drops the
// Receiver, and the producer's next `send` observes the closed channel and
// should exit (that is the documented producer contract).
//
// `chunked` selects HTTP/1.1 chunked framing (hex-length CRLF payload
// CRLF, terminated by a 0-chunk) vs HTTP/1.0 raw writes (EOF-delimited;
// caller closes). Empty chunks are skipped — a zero-length chunk would
// terminate the framing early.

fn pump_stream(
    fd: RawFd,
    stream: StreamBody,
    chunked: bool,
    write_timeout: Duration,
) -> io::Result<()> {
    let write_chunk = |payload: &[u8]| -> io::Result<()> {
        let deadline = Instant::now() + write_timeout;
        if chunked {
            let mut framed = Vec::with_capacity(payload.len() + 20);
            framed.extend_from_slice(format!("{:x}\r\n", payload.len()).as_bytes());
            framed.extend_from_slice(payload);
            framed.extend_from_slice(b"\r\n");
            write_all(fd, &framed, deadline)
        } else {
            write_all(fd, payload, deadline)
        }
    };

    loop {
        // With a heartbeat configured (SSE), the wait between chunks is the
        // heartbeat interval; expiry emits the ping and keeps waiting. A
        // ping write that stalls past write_timeout errors out below —
        // that is the dead-client detector. Without one, plain recv():
        // stoppable by the draining registry either way.
        let msg = match &stream.heartbeat {
            Some((interval, payload)) => match stream.rx.recv_timeout(*interval) {
                Ok(chunk) => Some(chunk),
                Err(smarm::RecvTimeoutError::Timeout) => {
                    write_chunk(payload)?;
                    continue;
                }
                Err(smarm::RecvTimeoutError::Disconnected) => None,
            },
            None => stream.rx.recv().ok(),
        };

        match msg {
            Some(chunk) => {
                if chunk.is_empty() {
                    continue;
                }
                write_chunk(&chunk)?;
            }
            None => {
                // All senders dropped: end of stream.
                if chunked {
                    write_all(fd, b"0\r\n\r\n", Instant::now() + write_timeout)?;
                }
                return Ok(());
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Error responses for unparseable / malformed requests.
// ---------------------------------------------------------------------------

fn emit_error_response(fd: RawFd, err: &ParseError, deadline: Instant) {
    let resp: &[u8] = match err {
        ParseError::TooManyHeaders =>
            b"HTTP/1.1 431 Request Header Fields Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
        ParseError::BadContentLength =>
            b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
        ParseError::Unsupported =>
            b"HTTP/1.1 411 Length Required\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
        // Incomplete and Malformed both lead here; Incomplete shouldn't
        // appear (read_head loops on it).
        _ =>
            b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
    };
    let _ = write_all(fd, resp, deadline);
}