liminal-sdk 0.16.0

use super::*;
use liminal::protocol::{
    CausalContext, DELIVER_MESSAGE_ID_VERSION, DeliveredId, MessageEnvelope, MessageId,
};
use std::net::TcpListener;

/// Encodes one frame into a fresh byte vector for the fake server below.
fn encode_frame(frame: &Frame) -> Result<Vec<u8>, SdkError> {
    let len = encoded_len(frame).map_err(|error| protocol_error(&error))?;
    let mut bytes = vec![0_u8; len];
    let written = encode(frame, &mut bytes).map_err(|error| protocol_error(&error))?;
    bytes.truncate(written);
    Ok(bytes)
}

/// Blocks reading `socket` until one complete frame decodes, discarding it. Used
/// by the fake server to consume the client's `Connect`/`Subscribe` frames.
fn read_and_discard_one(socket: &mut TcpStream, buffer: &mut Vec<u8>) -> Result<(), SdkError> {
    loop {
        match decode(buffer) {
            Ok((_, consumed)) => {
                buffer.drain(..consumed);
                return Ok(());
            }
            Err(
                ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
            ) => {
                let mut chunk = [0_u8; 512];
                let read = socket
                    .read(&mut chunk)
                    .map_err(|source| SdkError::Connection {
                        description: format!("fake server read failed: {source}"),
                    })?;
                if read == 0 {
                    return Err(SdkError::Connection {
                        description: "fake server: client closed before a full frame".to_string(),
                    });
                }
                buffer.extend_from_slice(chunk.get(..read).unwrap_or(&[]));
            }
            Err(error) => return Err(protocol_error(&error)),
        }
    }
}

#[test]
fn delivered_message_exposes_seq_and_payload() {
    let message = DeliveredMessage {
        delivery_seq: 3,
        schema_id: SchemaId::new([1; SchemaId::WIRE_LEN]),
        payload: vec![9, 8, 7],
        message: None,
    };
    assert_eq!(message.delivery_seq(), 3);
    assert_eq!(message.payload(), &[9, 8, 7]);
    assert_eq!(message.schema_id(), SchemaId::new([1; SchemaId::WIRE_LEN]));
    assert_eq!(message.into_payload(), vec![9, 8, 7]);
}

#[test]
fn deliver_frame_round_trips_through_codec() -> Result<(), SdkError> {
    // The exact frame the reader decodes: a Deliver carrying delivery_seq and a
    // MessageEnvelope whose payload the reader surfaces verbatim.
    let envelope = MessageEnvelope::new(
        SchemaId::new([2; SchemaId::WIRE_LEN]),
        CausalContext::independent(),
        vec![4, 5, 6],
    );
    let frame = Frame::new_deliver(SUBSCRIPTION_STREAM_ID, 1, envelope)
        .map_err(|error| protocol_error(&error))?;
    let len = encoded_len(&frame).map_err(|error| protocol_error(&error))?;
    let mut bytes = vec![0_u8; len];
    let written = encode(&frame, &mut bytes).map_err(|error| protocol_error(&error))?;
    let (decoded, consumed) = decode(&bytes[..written]).map_err(|error| protocol_error(&error))?;
    assert_eq!(consumed, written);
    let Frame::Deliver {
        delivery_seq,
        envelope,
        ..
    } = decoded
    else {
        return Err(SdkError::Protocol {
            description: "expected a Deliver frame".to_string(),
        });
    };
    assert_eq!(delivery_seq, 1);
    assert_eq!(envelope.payload, vec![4, 5, 6]);
    Ok(())
}

struct ScriptedServer {
    address: String,
    endpoint: std::net::SocketAddr,
    accepted: std::sync::Arc<std::sync::atomic::AtomicBool>,
    released: std::sync::Arc<std::sync::atomic::AtomicBool>,
    thread: std::thread::JoinHandle<Result<(), SdkError>>,
}

struct ListenerRelease<'a> {
    endpoint: std::net::SocketAddr,
    accepted: &'a std::sync::atomic::AtomicBool,
    released: &'a std::sync::atomic::AtomicBool,
    error: &'a mut Option<SdkError>,
}

impl Drop for ListenerRelease<'_> {
    fn drop(&mut self) {
        self.released
            .store(true, std::sync::atomic::Ordering::Release);
        if !self.accepted.load(std::sync::atomic::Ordering::Acquire) {
            match TcpStream::connect(self.endpoint) {
                Ok(socket) => drop(socket),
                Err(source) => {
                    *self.error = Some(SdkError::Connection {
                        description: format!("failed to release fake server accept: {source}"),
                    });
                }
            }
        }
    }
}

impl ScriptedServer {
    fn spawn(
        serve: impl FnOnce(TcpStream) -> Result<(), SdkError> + Send + 'static,
    ) -> Result<Self, SdkError> {
        let listener = TcpListener::bind("127.0.0.1:0").map_err(|source| SdkError::Connection {
            description: format!("failed to bind fake server: {source}"),
        })?;
        let endpoint = listener
            .local_addr()
            .map_err(|source| SdkError::Connection {
                description: format!("failed to read fake server address: {source}"),
            })?;
        let accepted = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let released = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let thread_accepted = std::sync::Arc::clone(&accepted);
        let thread_released = std::sync::Arc::clone(&released);
        let thread = std::thread::spawn(move || {
            let (socket, _) = listener.accept().map_err(|source| SdkError::Connection {
                description: format!("fake server accept failed: {source}"),
            })?;
            thread_accepted.store(true, std::sync::atomic::Ordering::Release);
            if thread_released.load(std::sync::atomic::Ordering::Acquire) {
                return Ok(());
            }
            serve(socket)
        });
        Ok(Self {
            address: endpoint.to_string(),
            endpoint,
            accepted,
            released,
            thread,
        })
    }

    fn run<T>(self, consume: impl FnOnce(&str) -> Result<T, SdkError>) -> Result<T, SdkError> {
        let mut release_error = None;
        let release = ListenerRelease {
            endpoint: self.endpoint,
            accepted: &self.accepted,
            released: &self.released,
            error: &mut release_error,
        };
        let consumption =
            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| consume(&self.address)))
                .unwrap_or_else(|payload| {
                    Err(SdkError::Connection {
                        description: format!(
                            "consumer assertion panicked: {}",
                            panic_description(payload.as_ref())
                        ),
                    })
                });
        drop(release);
        let server_result = self.thread.join().unwrap_or_else(|payload| {
            Err(SdkError::Connection {
                description: format!(
                    "fake server panicked: {}",
                    panic_description(payload.as_ref())
                ),
            })
        });
        let mut errors = Vec::with_capacity(3);
        if let Err(error) = &consumption {
            errors.push(format!("consumer failed: {error}"));
        }
        if let Err(error) = server_result {
            errors.push(format!("fake server failed: {error}"));
        }
        if let Some(error) = release_error {
            errors.push(format!("listener release failed: {error}"));
        }
        if !errors.is_empty() {
            return Err(SdkError::Connection {
                description: errors.join("; "),
            });
        }
        consumption
    }
}

fn panic_description(payload: &(dyn std::any::Any + Send)) -> String {
    payload.downcast_ref::<&str>().map_or_else(
        || {
            payload
                .downcast_ref::<String>()
                .cloned()
                .unwrap_or_else(|| {
                    format!("non-text panic payload of type {:?}", payload.type_id())
                })
        },
        |message| (*message).to_string(),
    )
}

fn wait_for_client_close(reader: &mut impl std::io::Read) -> Result<(), SdkError> {
    let mut scratch = [0_u8; 512];
    loop {
        let read = reader
            .read(&mut scratch)
            .map_err(|source| SdkError::Connection {
                description: format!("fake server client-close read failed: {source}"),
            })?;
        if read == 0 {
            return Ok(());
        }
    }
}

#[test]
fn assertion_panic_still_reports_the_server_result() -> Result<(), SdkError> {
    let (ready_tx, ready_rx) = std::sync::mpsc::channel::<()>();
    let server = ScriptedServer::spawn(move |socket| {
        drop(socket);
        ready_tx.send(()).map_err(|source| SdkError::Connection {
            description: format!("failed to signal forced server refusal: {source}"),
        })?;
        Err(SdkError::Connection {
            description: "forced server refusal".to_string(),
        })
    })?;
    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        server.run(|address| {
            let socket = TcpStream::connect(address).map_err(|source| SdkError::Connection {
                description: format!("failed to connect assertion fixture: {source}"),
            })?;
            ready_rx.recv().map_err(|source| SdkError::Connection {
                description: format!("server refusal signal closed: {source}"),
            })?;
            drop(socket);
            assert_eq!(1, 2, "forced consumer assertion");
            Ok(())
        })
    }));
    let Ok(Err(SdkError::Connection { description })) = result else {
        return Err(SdkError::Connection {
            description: "assertion panic left the server result unread".to_string(),
        });
    };
    assert!(
        description.contains("forced consumer assertion"),
        "consumer assertion cause was lost: {description}"
    );
    assert!(
        description.contains("forced server refusal"),
        "server result was not reported: {description}"
    );
    Ok(())
}

#[test]
fn pre_connect_refusal_releases_and_joins_the_server() -> Result<(), SdkError> {
    let handler_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let observed = std::sync::Arc::clone(&handler_calls);
    let server = ScriptedServer::spawn(move |socket| {
        drop(socket);
        observed.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        Ok(())
    })?;
    let result = server.run::<()>(|address| {
        Err(SdkError::Connection {
            description: format!("forced pre-connect refusal at {address}"),
        })
    });
    let Err(SdkError::Connection { description }) = result else {
        return Err(SdkError::Connection {
            description: "pre-connect refusal was not reported".to_string(),
        });
    };
    assert!(description.contains("forced pre-connect refusal"));
    assert_eq!(handler_calls.load(std::sync::atomic::Ordering::SeqCst), 0);
    Ok(())
}

fn assert_server_panic_is_reported(
    panic_server: impl FnOnce() + Send + 'static,
    payload: &str,
) -> Result<(), SdkError> {
    let (ready_tx, ready_rx) = std::sync::mpsc::channel::<()>();
    let server = ScriptedServer::spawn(move |socket| {
        drop(socket);
        ready_tx.send(()).map_err(|source| SdkError::Connection {
            description: format!("failed to signal forced server panic: {source}"),
        })?;
        panic_server();
        Ok(())
    })?;
    let result = server.run(|address| {
        let socket = TcpStream::connect(address).map_err(|source| SdkError::Connection {
            description: format!("failed to connect panic fixture: {source}"),
        })?;
        ready_rx.recv().map_err(|source| SdkError::Connection {
            description: format!("server panic signal closed: {source}"),
        })?;
        drop(socket);
        Ok(())
    });
    let Err(SdkError::Connection { description }) = result else {
        return Err(SdkError::Connection {
            description: "server panic was not reported".to_string(),
        });
    };
    assert!(
        description.contains(payload),
        "server panic payload was lost: {description}"
    );
    Ok(())
}

#[test]
fn server_string_slice_panic_payload_is_preserved() -> Result<(), SdkError> {
    assert_server_panic_is_reported(
        || std::panic::resume_unwind(Box::new("forced slice panic payload")),
        "forced slice panic payload",
    )
}

#[test]
fn server_owned_string_panic_payload_is_preserved() -> Result<(), SdkError> {
    assert_server_panic_is_reported(
        || std::panic::resume_unwind(Box::new(String::from("forced owned panic payload"))),
        "forced owned panic payload",
    )
}

#[test]
fn client_close_read_failure_is_reported() -> Result<(), SdkError> {
    struct RefusingReader;
    impl std::io::Read for RefusingReader {
        fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
            if buffer.is_empty() {
                return Ok(0);
            }
            Err(std::io::Error::other("forced client-close read failure"))
        }
    }
    let result = wait_for_client_close(&mut RefusingReader);
    let Err(SdkError::Connection { description }) = result else {
        return Err(SdkError::Connection {
            description: "client-close read failure was silently treated as EOF".to_string(),
        });
    };
    assert!(description.contains("forced client-close read failure"));
    Ok(())
}

/// Repro for the setup-residue drop: a server that coalesces `SubscribeAck` with
/// the first `Deliver` frames into one TCP segment must not lose those
/// deliveries. Before the fix, the throwaway subscribe buffer discarded the bytes
/// read past the ack and the reader started on an empty buffer, so both deliveries
/// vanished; now the residue threads into the reader and surfaces.
#[test]
fn open_preserves_deliveries_coalesced_with_the_subscribe_ack() -> Result<(), SdkError> {
    let (consumed_tx, consumed_rx) = std::sync::mpsc::channel::<()>();

    let server = ScriptedServer::spawn(move |mut socket| {
        let mut buffer = Vec::new();
        // Consume the client's Connect, then ack the handshake on its own write.
        read_and_discard_one(&mut socket, &mut buffer)?;
        write_frame(
            &mut socket,
            &Frame::ConnectAck {
                flags: 0,
                selected_version: CLIENT_MAX_VERSION,
                capabilities: 0,
            },
        )?;
        // Consume the client's Subscribe, then coalesce the SubscribeAck and two
        // Deliver frames into ONE segment — the exact hot-channel repro.
        read_and_discard_one(&mut socket, &mut buffer)?;
        let schema = SchemaId::new([7; SchemaId::WIRE_LEN]);
        let ack = Frame::SubscribeAck {
            flags: 0,
            stream_id: SUBSCRIPTION_STREAM_ID,
            subscription_id: 42,
            selected_schema: schema,
        };
        let first = Frame::new_deliver(
            SUBSCRIPTION_STREAM_ID,
            1,
            MessageEnvelope::new(schema, CausalContext::independent(), vec![1, 1, 1]),
        )
        .map_err(|error| protocol_error(&error))?;
        let second = Frame::new_deliver(
            SUBSCRIPTION_STREAM_ID,
            2,
            MessageEnvelope::new(schema, CausalContext::independent(), vec![2, 2, 2]),
        )
        .map_err(|error| protocol_error(&error))?;
        let mut segment = Vec::new();
        segment.extend_from_slice(&encode_frame(&ack)?);
        segment.extend_from_slice(&encode_frame(&first)?);
        segment.extend_from_slice(&encode_frame(&second)?);
        socket
            .write_all(&segment)
            .map_err(|source| SdkError::Connection {
                description: format!("fake server write failed: {source}"),
            })?;
        socket.flush().map_err(|source| SdkError::Connection {
            description: format!("fake server flush failed: {source}"),
        })?;
        // Keep the socket open until the consumer has asserted both deliveries.
        // Dropping the sender also releases this wait on a consumer failure.
        consumed_rx.recv().map_err(|source| SdkError::Connection {
            description: format!("fake server consumer signal closed: {source}"),
        })?;
        Ok(())
    })?;

    server.run(move |address| {
        let subscription = SubscriptionStream::open(address, "orders", Vec::new())?;
        assert_eq!(subscription.subscription_id(), 42);
        let first = subscription.recv_timeout(Duration::from_secs(2))?;
        assert_eq!(first.delivery_seq(), 1);
        assert_eq!(first.payload(), &[1, 1, 1]);
        let second = subscription.recv_timeout(Duration::from_secs(2))?;
        assert_eq!(second.delivery_seq(), 2);
        assert_eq!(second.payload(), &[2, 2, 2]);
        consumed_tx
            .send(())
            .map_err(|source| SdkError::Connection {
                description: format!("failed to release fake server after deliveries: {source}"),
            })?;
        drop(subscription);
        Ok(())
    })
}

/// The named setup deadline SDK-010 installs: 5 s, the estate's already-ratified
/// constant generalized, never a new default.
const NAMED_SETUP_DEADLINE: Duration = Duration::from_secs(5);

/// A control-frame reply that is slow but well-behaved: past the retired 100 ms
/// `READER_POLL_TIMEOUT` cadence, far inside the named 5 s deadline. The delay is
/// the fixture's INTENDED slow reply, not a proof device.
// The fixed delay is the slow reply being tested, not fixture synchronisation.
// Readiness and teardown use their own signals independently.
const SLOW_BUT_ANSWERED: Duration = Duration::from_millis(250);

/// Runs a fake server that answers the handshake and the subscribe, optionally
/// delaying each control reply, then holds the socket until the client tears it
/// down. The owned server is released after the consumer finishes.
fn spawn_setup_server(reply_delay: Duration) -> Result<ScriptedServer, SdkError> {
    ScriptedServer::spawn(move |mut socket| {
        let mut buffer = Vec::new();
        read_and_discard_one(&mut socket, &mut buffer)?;
        std::thread::sleep(reply_delay);
        write_frame(
            &mut socket,
            &Frame::ConnectAck {
                flags: 0,
                selected_version: CLIENT_MAX_VERSION,
                capabilities: 0,
            },
        )?;
        read_and_discard_one(&mut socket, &mut buffer)?;
        std::thread::sleep(reply_delay);
        write_frame(
            &mut socket,
            &Frame::SubscribeAck {
                flags: 0,
                stream_id: SUBSCRIPTION_STREAM_ID,
                subscription_id: 42,
                selected_schema: SchemaId::new([7; SchemaId::WIRE_LEN]),
            },
        )?;
        // Read to the client's teardown so the socket outlives the assertions.
        wait_for_client_close(&mut socket)
    })
}

/// RED PIN (SDK-010 R2, direction (b)) — the disarm, pinned behaviorally.
///
/// `connect_socket` arms the retired 100 ms `READER_POLL_TIMEOUT` on this socket
/// and never takes it off, so the steady-state reader wakes ten times a second
/// forever on a quiet subscription — family F9 in the wiring ledger.
/// `read_timeout()` reads the live `SO_RCVTIMEO` off the very kernel socket the
/// reader thread blocks on (the writer handle and the reader's handle are
/// `try_clone` siblings), so this observes behaviour, not source.
///
/// The assertion is `None`, not "longer than before": a leaked setup deadline
/// would report `Some(5s)` and fail exactly as `Some(100ms)` does.
#[test]
fn the_subscription_reader_carries_no_read_window_in_steady_state() -> Result<(), SdkError> {
    let server = spawn_setup_server(Duration::ZERO)?;
    server.run(|address| {
        let subscription = SubscriptionStream::open(address, "orders", Vec::new())?;
        let observed =
            subscription
                .writer
                .read_timeout()
                .map_err(|source| SdkError::Connection {
                    description: format!(
                        "failed to read the subscription socket read timeout: {source}"
                    ),
                })?;
        assert_eq!(
            observed, None,
            "the subscription reader must block with no read window once the control \
             exchange is over; a Some(_) here is a cadence, whatever its period"
        );
        drop(subscription);
        Ok(())
    })
}

/// PARITY PIN (SDK-010 R3) — the same slow-reply fixture the push reader is red
/// against, aimed at the subscription reader.
///
/// This one is GREEN before the change as well as after: `subscription.rs`'s
/// `read_one_frame` already samples the 100 ms cadence against a 5 s
/// `SETUP_TIMEOUT` total deadline instead of dying on the first tick, so a
/// 250 ms reply already survives here. It is pinned so the two TCP readers are
/// held to ONE answer after the generalization, and so a regression toward the
/// push reader's fatal-on-first-timeout shape is caught.
#[test]
fn open_survives_control_replies_slower_than_the_retired_poll_cadence() -> Result<(), SdkError> {
    let server = spawn_setup_server(SLOW_BUT_ANSWERED)?;
    server.run(|address| {
        let subscription = SubscriptionStream::open(address, "orders", Vec::new())?;
        assert_eq!(subscription.subscription_id(), 42);
        assert!(
            NAMED_SETUP_DEADLINE > SLOW_BUT_ANSWERED,
            "the fixture's slow reply must sit inside the named deadline"
        );
        drop(subscription);
        Ok(())
    })
}

/// TOMBSTONE (SDK-010 R5) — the retired reader poll family must not reappear in
/// this reader's production source.
#[test]
fn subscription_source_has_no_retired_reader_poll_family() {
    const SOURCE: &str = include_str!("subscription.rs");
    let production = SOURCE.split("#[cfg(test)]").next().unwrap_or(SOURCE);
    for forbidden in [
        "READER_POLL_TIMEOUT",
        "AtomicBool",
        "stop.load",
        "stop.store",
        "re-check the stop flag",
        "poll the stop flag",
    ] {
        assert!(
            !production.contains(forbidden),
            "retired subscription-reader poll-family source `{forbidden}` reappeared"
        );
    }
}

/// The boot and id a 1.2 server gives a delivery (Fable 4c).
const DELIVERED: DeliveredId = DeliveredId {
    boot: 1_791_328_512_345,
    id: 42,
};

/// 4c: a delivery that carries the server's id names its reply's parent as
/// `<boot>:<id>`; one that carries none names no parent, never an invented one.
#[test]
fn a_delivered_message_names_its_reply_parent_only_when_it_carries_an_id() {
    let with_id = DeliveredMessage {
        delivery_seq: 1,
        schema_id: SchemaId::new([1; SchemaId::WIRE_LEN]),
        payload: Vec::new(),
        message: Some(DELIVERED),
    };
    assert_eq!(with_id.message(), Some(DELIVERED));
    assert_eq!(
        with_id.reply_parent(),
        Some(MessageId::from("1791328512345:42"))
    );
    let without = DeliveredMessage {
        message: None,
        ..with_id
    };
    assert_eq!(without.message(), None);
    assert_eq!(without.reply_parent(), None);
}

/// 4c: the subscription offers 1.2, so a 1.2 server may send the id.
#[test]
fn the_subscription_offers_the_delivered_id_version() {
    assert_eq!(CLIENT_MAX_VERSION, DELIVER_MESSAGE_ID_VERSION);
}

/// 4c: the reader keeps the id a Deliver carried.
#[test]
fn the_reader_keeps_the_delivered_id() -> Result<(), SdkError> {
    let server = ScriptedServer::spawn(move |mut socket| {
        let mut buffer = Vec::new();
        read_and_discard_one(&mut socket, &mut buffer)?;
        write_frame(
            &mut socket,
            &Frame::ConnectAck {
                flags: 0,
                selected_version: DELIVER_MESSAGE_ID_VERSION,
                capabilities: 0,
            },
        )?;
        read_and_discard_one(&mut socket, &mut buffer)?;
        let schema = SchemaId::new([7; SchemaId::WIRE_LEN]);
        write_frame(
            &mut socket,
            &Frame::SubscribeAck {
                flags: 0,
                stream_id: SUBSCRIPTION_STREAM_ID,
                subscription_id: 42,
                selected_schema: schema,
            },
        )?;
        write_frame(
            &mut socket,
            &Frame::Deliver {
                flags: 0,
                stream_id: SUBSCRIPTION_STREAM_ID,
                delivery_seq: 1,
                envelope: MessageEnvelope::new(schema, CausalContext::independent(), vec![1]),
                message: Some(DELIVERED),
            },
        )?;
        wait_for_client_close(&mut socket)
    })?;

    server.run(|address| {
        let subscription = SubscriptionStream::open(address, "orders", Vec::new())?;
        let delivered = subscription.recv_timeout(Duration::from_secs(2))?;
        assert_eq!(delivered.message(), Some(DELIVERED));
        drop(subscription);
        Ok(())
    })
}