tailsurf-cli 0.1.0

//! End-to-end CLI tests against in-process HTTP and WebSocket fixtures.

use std::{
    collections::HashMap,
    fs,
    process::{Command, Stdio},
    sync::{Arc, Mutex},
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use axum::{
    Json, Router,
    extract::{
        Path, Query, State, WebSocketUpgrade,
        ws::{CloseFrame, Message, WebSocket},
    },
    http::{HeaderMap, StatusCode},
    response::{IntoResponse, Response},
    routing::{get, post},
};
use bytes::Bytes;
use secrecy::ExposeSecret;
use tailsurf::{
    BearerToken, StreamId, TokenId, TokenPermissions, TsfClient, TsfProducerConfig, WriterId,
    protocol::{
        rest::{
            CreateStreamRequest, CreateStreamResponse, IssueTokenRequest, IssueTokenResponse,
            IssuedStreamToken, ListTokensResponse, RevokeTokenRequest, StreamInfoResponse,
            StreamTailResponse, StreamTokenStatus, StreamTokenSummary, UpdateStreamRequest,
            Visibility,
        },
        ws::{
            ReadStart, ReadStreamOptions,
            frame::{
                ClientFrame, MAX_RECORD_BYTES, PartHeader, ReadRecord, RecordFormat, ServerFrame,
                TSF_V1, TSF_WS_PROTOCOL,
            },
        },
    },
    stream_url::StreamLocator,
};
use tokio::{io::AsyncReadExt, net::TcpListener, process::Command as TokioCommand, time::timeout};
use url::Url;

#[tokio::test]
async fn new_outputs_json_and_token_files() {
    let server = TestServer::start().await;
    let tmp = std::env::temp_dir().join(format!(
        "tsf-cli-test-{}-{}",
        std::process::id(),
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("time")
            .as_nanos()
    ));
    let _ = fs::remove_dir_all(&tmp);
    fs::create_dir_all(&tmp).expect("tmp dir");
    let owner_file = tmp.join("owner.token");
    let read_file = tmp.join("read.token");
    let write_file = tmp.join("write.token");

    let output = run_tsf(
        &server,
        [
            "new",
            "--format",
            "json",
            "--owner-token-file",
            owner_file.to_str().expect("owner path"),
            "--read-token-file",
            read_file.to_str().expect("read path"),
            "--write-token-file",
            write_file.to_str().expect("write path"),
        ],
        None,
    )
    .await;
    assert!(output.status.success(), "stderr={}", output.stderr);
    let json: serde_json::Value = serde_json::from_str(&output.stdout).expect("json output");
    assert!(json["stream_id"].as_str().is_some());
    assert_eq!(json["retention_secs"], 864_000);
    assert!(json["urls"]["o"].as_str().is_some());
    assert!(json["urls"]["r"].as_str().is_some());
    assert!(json["urls"]["w"].as_str().is_some());
    assert!(
        !fs::read_to_string(owner_file)
            .expect("owner token")
            .is_empty()
    );
    assert!(
        !fs::read_to_string(read_file)
            .expect("read token")
            .is_empty()
    );
    assert!(
        !fs::read_to_string(write_file)
            .expect("write token")
            .is_empty()
    );

    fs::remove_dir_all(tmp).expect("cleanup");
    server.abort();
}

#[tokio::test]
async fn new_text_output_covers_visibility_and_explicit_tokens() {
    let server = TestServer::start().await;

    let private = run_tsf(&server, ["new"], None).await;
    assert!(private.status.success(), "stderr={}", private.stderr);
    assert_eq!(
        normalize_created_stream_output(&private.stdout),
        "stream_id=<stream_id>\nretention_secs=<retention_secs>\no=<url>\nw=<url>\nr=<url>\n"
    );
    assert_created_output_urls_parse(&private.stdout, &["o", "w", "r"]);

    let public = run_tsf(&server, ["new", "--public"], None).await;
    assert!(public.status.success(), "stderr={}", public.stderr);
    assert_eq!(
        normalize_created_stream_output(&public.stdout),
        "stream_id=<stream_id>\nretention_secs=<retention_secs>\no=<url>\nw=<url>\n"
    );
    assert_created_output_urls_parse(&public.stdout, &["o", "w"]);

    let explicit = run_tsf(&server, ["new", "--token", "rw", "--token", "r"], None).await;
    assert!(explicit.status.success(), "stderr={}", explicit.stderr);
    assert_eq!(
        normalize_created_stream_output(&explicit.stdout),
        "stream_id=<stream_id>\nretention_secs=<retention_secs>\nrw=<url>\nr=<url>\n"
    );
    assert_created_output_urls_parse(&explicit.stdout, &["rw", "r"]);

    server.abort();
}

#[tokio::test]
async fn write_new_then_replay_round_trips_command_output() {
    let server = TestServer::start().await;
    let output = run_tsf(
        &server,
        ["write", "--new"],
        Some("hello from cli integration\n"),
    )
    .await;
    assert!(output.status.success(), "stderr={}", output.stderr);
    assert_eq!(output.stdout, "");
    assert_eq!(
        normalize_created_stream_output(&output.stderr),
        "stream_id=<stream_id>\nretention_secs=<retention_secs>\no=<url>\nw=<url>\nr=<url>\n"
    );
    let read_url = output
        .stderr
        .lines()
        .find_map(|line| line.strip_prefix("r="))
        .expect("read url");
    StreamLocator::parse(read_url).expect("valid read URL");

    let replay = run_tsf(&server, ["replay", read_url], None).await;
    assert!(replay.status.success(), "stderr={}", replay.stderr);
    assert_eq!(replay.stdout, "hello from cli integration\n");

    server.abort();
}

#[tokio::test]
async fn write_new_command_streams_output_and_propagates_exit_status() {
    let server = TestServer::start().await;
    let output = run_tsf(
        &server,
        [
            "write",
            "--new",
            "--",
            "sh",
            "-c",
            "printf out; printf err >&2; exit 7",
        ],
        None,
    )
    .await;
    assert_eq!(output.status.code(), Some(7), "stderr={}", output.stderr);
    let read_url = output
        .stderr
        .lines()
        .find_map(|line| line.strip_prefix("r="))
        .expect("read url");

    let replay = run_tsf(&server, ["replay", read_url], None).await;
    assert!(replay.status.success(), "stderr={}", replay.stderr);
    assert!(replay.stdout.contains("out"), "stdout={}", replay.stdout);
    assert!(replay.stdout.contains("err"), "stdout={}", replay.stdout);

    server.abort();
}

#[tokio::test]
async fn write_defaults_to_lines_and_splits_large_records() {
    let server = TestServer::start().await;
    let mut input = "x".repeat(MAX_RECORD_BYTES + 10);
    input.push('\n');
    input.push_str("tail\n");

    let output = run_tsf(&server, ["write", "--new"], Some(input.as_str())).await;
    assert!(output.status.success(), "stderr={}", output.stderr);
    let read_url = output
        .stderr
        .lines()
        .find_map(|line| line.strip_prefix("r="))
        .expect("read url");
    let locator = StreamLocator::parse(read_url).expect("valid read URL");
    let read_token = locator
        .token_with(TokenPermissions::allows_read)
        .expect("read token");
    let client = TsfClient::with_api_base_url(server.api_url.clone());
    let mut request = ReadStreamOptions::new(locator.stream_id).with_stream_token(read_token);
    request.start = Some(ReadStart::SeqNum(0));
    request.count = Some(3);
    let mut reader = client.connect_reader(request).await.expect("reader");

    let mut records = Vec::new();
    while records.len() < 3 {
        match reader.next_record().await.expect("event") {
            Some(record) => records.push(record),
            None => panic!("reader closed before expected records"),
        }
    }

    assert_eq!(records[0].writer_seq_num, 0);
    assert_eq!(records[0].part, PartHeader::new(0, false).expect("part"));
    assert_eq!(records[0].format, RecordFormat::Transcript);
    assert_eq!(records[0].data.len(), MAX_RECORD_BYTES);
    assert_eq!(records[1].writer_seq_num, 1);
    assert_eq!(records[1].part, PartHeader::new(1, true).expect("part"));
    assert_eq!(records[1].format, RecordFormat::Transcript);
    assert_eq!(records[1].data.len(), 11);
    assert_eq!(records[1].data.last(), Some(&b'\n'));
    assert_eq!(records[2].writer_seq_num, 2);
    assert_eq!(records[2].part, PartHeader::unsplit());
    assert_eq!(records[2].format, RecordFormat::Transcript);
    assert_eq!(records[2].data.as_ref(), b"tail\n");

    server.abort();
}

#[tokio::test]
async fn write_raw_preserves_large_input_across_flush_boundaries() {
    let server = TestServer::start().await;
    let input = "x".repeat(MAX_RECORD_BYTES + 10);

    let output = run_tsf(&server, ["write", "--new", "--raw"], Some(input.as_str())).await;
    assert!(output.status.success(), "stderr={}", output.stderr);
    let read_url = output
        .stderr
        .lines()
        .find_map(|line| line.strip_prefix("r="))
        .expect("read url");
    let locator = StreamLocator::parse(read_url).expect("valid read URL");
    let read_token = locator
        .token_with(TokenPermissions::allows_read)
        .expect("read token");
    let client = TsfClient::with_api_base_url(server.api_url.clone());
    let mut request = ReadStreamOptions::new(locator.stream_id).with_stream_token(read_token);
    request.start = Some(ReadStart::SeqNum(0));
    request.count = Some(16);
    let mut reader = client.connect_reader(request).await.expect("reader");

    let mut records = Vec::new();
    let mut output = Vec::new();
    while output.len() < input.len() {
        match reader.next_record().await.expect("event") {
            Some(record) => {
                output.extend_from_slice(&record.data);
                records.push(record);
            }
            None => panic!("reader closed before expected records"),
        }
    }

    assert_eq!(output, input.as_bytes());
    assert!(
        records
            .iter()
            .all(|record| record.part == PartHeader::unsplit())
    );
    assert!(
        records
            .iter()
            .all(|record| record.format == RecordFormat::Bytes)
    );
    assert!(
        records
            .iter()
            .all(|record| record.data.len() <= MAX_RECORD_BYTES)
    );
    for (index, record) in records.iter().enumerate() {
        assert_eq!(record.writer_seq_num, index as u64);
    }

    server.abort();
}

#[tokio::test]
async fn write_raw_flushes_on_linger() {
    let server = TestServer::start().await;

    let output = run_tsf(
        &server,
        [
            "write",
            "--new",
            "--raw",
            "--",
            "sh",
            "-c",
            "printf a; sleep 0.1; printf b",
        ],
        None,
    )
    .await;
    assert!(output.status.success(), "stderr={}", output.stderr);
    let read_url = output
        .stderr
        .lines()
        .find_map(|line| line.strip_prefix("r="))
        .expect("read url");
    let locator = StreamLocator::parse(read_url).expect("valid read URL");
    let read_token = locator
        .token_with(TokenPermissions::allows_read)
        .expect("read token");
    let client = TsfClient::with_api_base_url(server.api_url.clone());
    let mut request = ReadStreamOptions::new(locator.stream_id).with_stream_token(read_token);
    request.start = Some(ReadStart::SeqNum(0));
    request.count = Some(2);
    let mut reader = client.connect_reader(request).await.expect("reader");

    let mut data = Vec::new();
    while data.len() < 2 {
        match reader.next_record().await.expect("event") {
            Some(record) => {
                assert_eq!(record.format, RecordFormat::Bytes);
                data.push(record.data);
            }
            None => panic!("reader closed before expected records"),
        }
    }

    assert_eq!(data[0].as_ref(), b"a");
    assert_eq!(data[1].as_ref(), b"b");

    server.abort();
}

#[tokio::test]
async fn write_reconnect_reuses_writer_identity_and_unacked_sequence() {
    let server = FakeWriteServer::start().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let write_url = format!("http://localhost:3000/s/{stream_id}#w=write-secret");

    let output = run_tsf_with_api_url(
        server.api_url.clone(),
        ["write", write_url.as_str()],
        Some("retry me\n"),
    )
    .await;

    assert!(output.status.success(), "stderr={}", output.stderr);
    let attempts = server.append_attempts();
    assert_eq!(attempts.len(), 2);
    assert_eq!(attempts[0].writer_id, attempts[1].writer_id);
    assert_eq!(attempts[0].bearer_token, "write-secret");
    assert_eq!(attempts[1].bearer_token, "write-secret");
    assert_eq!(attempts[0].writer_seq_num, 0);
    assert_eq!(attempts[1].writer_seq_num, 0);
    assert_eq!(attempts[0].data.as_ref(), b"retry me\n");
    assert_eq!(attempts[1].data.as_ref(), b"retry me\n");
    assert_eq!(attempts[0].part, PartHeader::unsplit());
    assert_eq!(attempts[1].part, PartHeader::unsplit());
    assert_eq!(attempts[0].format, RecordFormat::Transcript);
    assert_eq!(attempts[1].format, RecordFormat::Transcript);

    server.abort();
}

#[tokio::test]
async fn producer_close_is_not_blocked_by_an_unused_reservation() {
    let server = FakeWriteServer::start().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let client = TsfClient::with_api_base_url(server.api_url.clone());
    let producer = client
        .connect_producer_with_config(
            tailsurf::protocol::ws::WriteStreamOptions::new(
                stream_id,
                WriterId::new_random(),
                "write-secret",
            ),
            TsfProducerConfig {
                max_unacked_bytes: 1,
                max_unacked_records: 1,
                max_reconnect_attempts: 0,
            },
        )
        .await
        .expect("producer");
    let _permit = producer.reserve(1).await.expect("reservation");

    timeout(Duration::from_secs(1), producer.close())
        .await
        .expect("producer close must not wait for reservation")
        .expect("producer close");

    server.abort();
}

#[tokio::test]
async fn tail_reconnect_resumes_after_last_s2_sequence() {
    let server = FakeReadServer::start().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let output = run_tsf_until_stdout_contains(
        server.api_url.clone(),
        ["tail", read_url.as_str()],
        b"first\nsecond\n",
        Duration::from_secs(5),
    )
    .await;

    assert_eq!(output.stdout, "first\nsecond\n");
    assert_eq!(output.stderr, "");
    let attempts = server.read_attempts();
    assert_eq!(attempts.len(), 2);
    assert_eq!(attempts[0].bearer_token, "read-secret");
    assert_eq!(attempts[1].bearer_token, "read-secret");
    assert_eq!(
        attempts[0].query.get("tail_offset").map(String::as_str),
        Some("0")
    );
    assert_eq!(attempts[0].query.get("seq_num"), None);
    assert_eq!(
        attempts[1].query.get("seq_num").map(String::as_str),
        Some("1")
    );
    assert_eq!(attempts[1].query.get("tail_offset"), None);

    server.abort();
}

#[tokio::test]
async fn tail_broken_websocket_resumes_after_last_s2_sequence() {
    let server = FakeReadServer::start_broken_read().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let output = run_tsf_until_stdout_contains(
        server.api_url.clone(),
        ["tail", read_url.as_str()],
        b"first\nsecond\n",
        Duration::from_secs(5),
    )
    .await;

    assert_eq!(output.stdout, "first\nsecond\n");
    assert_eq!(output.stderr, "");
    let attempts = server.read_attempts();
    assert_eq!(attempts.len(), 2);
    assert_eq!(attempts[0].bearer_token, "read-secret");
    assert_eq!(attempts[1].bearer_token, "read-secret");
    assert_eq!(
        attempts[0].query.get("tail_offset").map(String::as_str),
        Some("0")
    );
    assert_eq!(attempts[0].query.get("seq_num"), None);
    assert_eq!(
        attempts[1].query.get("seq_num").map(String::as_str),
        Some("1")
    );
    assert_eq!(attempts[1].query.get("tail_offset"), None);

    server.abort();
}

#[tokio::test]
async fn tail_server_shutdown_resumes_after_last_s2_sequence() {
    let server = FakeReadServer::start_server_shutdown().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let output = run_tsf_until_stdout_contains(
        server.api_url.clone(),
        ["tail", read_url.as_str()],
        b"first\nsecond\n",
        Duration::from_secs(5),
    )
    .await;

    assert_eq!(output.stdout, "first\nsecond\n");
    assert_eq!(output.stderr, "");
    let attempts = server.read_attempts();
    assert_eq!(attempts.len(), 2);
    assert_eq!(
        attempts[1].query.get("seq_num").map(String::as_str),
        Some("1")
    );

    server.abort();
}

#[tokio::test]
async fn tail_selector_flags_are_sent_as_read_query() {
    let tail_offset_server = FakeReadServer::start().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let tail_offset_output = run_tsf_until_stdout_contains(
        tail_offset_server.api_url.clone(),
        ["tail", "-n", "25", "--count", "7", read_url.as_str()],
        b"first\n",
        Duration::from_secs(5),
    )
    .await;

    assert_eq!(tail_offset_output.stdout, "first\n");
    assert_eq!(tail_offset_output.stderr, "");
    let attempts = tail_offset_server.read_attempts();
    assert_eq!(attempts.len(), 1);
    assert_eq!(
        attempts[0].query.get("tail_offset").map(String::as_str),
        Some("25")
    );
    assert_eq!(
        attempts[0].query.get("count").map(String::as_str),
        Some("7")
    );
    assert_eq!(attempts[0].query.get("seq_num"), None);
    assert_eq!(attempts[0].query.get("timestamp"), None);
    tail_offset_server.abort();

    let seq_server = FakeReadServer::start().await;
    let seq_output = run_tsf_until_stdout_contains(
        seq_server.api_url.clone(),
        ["tail", "--seq-num", "42", "--count", "3", read_url.as_str()],
        b"first\n",
        Duration::from_secs(5),
    )
    .await;

    assert_eq!(seq_output.stdout, "first\n");
    assert_eq!(seq_output.stderr, "");
    let attempts = seq_server.read_attempts();
    assert_eq!(attempts.len(), 1);
    assert_eq!(
        attempts[0].query.get("seq_num").map(String::as_str),
        Some("42")
    );
    assert_eq!(
        attempts[0].query.get("count").map(String::as_str),
        Some("3")
    );
    assert_eq!(attempts[0].query.get("tail_offset"), None);
    assert_eq!(attempts[0].query.get("timestamp"), None);
    seq_server.abort();

    let timestamp_server = FakeReadServer::start().await;
    let timestamp_output = run_tsf_until_stdout_contains(
        timestamp_server.api_url.clone(),
        ["tail", "--timestamp", "1781717406000", read_url.as_str()],
        b"first\n",
        Duration::from_secs(5),
    )
    .await;

    assert_eq!(timestamp_output.stdout, "first\n");
    assert_eq!(timestamp_output.stderr, "");
    let attempts = timestamp_server.read_attempts();
    assert_eq!(attempts.len(), 1);
    assert_eq!(
        attempts[0].query.get("timestamp").map(String::as_str),
        Some("1781717406000")
    );
    assert_eq!(attempts[0].query.get("tail_offset"), None);
    assert_eq!(attempts[0].query.get("seq_num"), None);
    timestamp_server.abort();
}

#[tokio::test]
async fn tail_rejects_ambiguous_start_selectors_before_connecting() {
    let server = FakeReadServer::start().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let output = run_tsf_with_api_url(
        server.api_url.clone(),
        ["tail", "-n", "10", "--seq-num", "5", read_url.as_str()],
        None,
    )
    .await;

    assert!(!output.status.success(), "stdout={}", output.stdout);
    assert!(
        output.stderr.contains("cannot be used with"),
        "stderr={}",
        output.stderr
    );
    assert!(server.read_attempts().is_empty());
    server.abort();
}

#[tokio::test]
async fn cli_reports_url_errors_before_opening_sockets() {
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");

    let read_server = FakeReadServer::start().await;
    let bad_tail = run_tsf_with_api_url(
        read_server.api_url.clone(),
        ["tail", "http://localhost:3000/not-a-stream"],
        None,
    )
    .await;
    assert!(!bad_tail.status.success(), "stdout={}", bad_tail.stdout);
    assert!(
        bad_tail.stderr.contains("invalid stream URL"),
        "stderr={}",
        bad_tail.stderr
    );
    assert!(read_server.read_attempts().is_empty());

    let bad_replay = run_tsf_with_api_url(
        read_server.api_url.clone(),
        ["replay", "http://localhost:3000/not-a-stream"],
        None,
    )
    .await;
    assert!(!bad_replay.status.success(), "stdout={}", bad_replay.stdout);
    assert!(
        bad_replay.stderr.contains("invalid stream URL"),
        "stderr={}",
        bad_replay.stderr
    );
    assert!(read_server.read_attempts().is_empty());
    read_server.abort();

    let write_server = FakeWriteServer::start().await;
    let read_only_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");
    let missing_write_token = run_tsf_with_api_url(
        write_server.api_url.clone(),
        ["write", &read_only_url],
        Some("data"),
    )
    .await;
    assert!(
        !missing_write_token.status.success(),
        "stdout={}",
        missing_write_token.stdout
    );
    assert!(
        missing_write_token
            .stderr
            .contains("stream URL does not contain a write token"),
        "stderr={}",
        missing_write_token.stderr
    );
    assert!(write_server.append_attempts().is_empty());
    write_server.abort();

    let read_only_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");
    let missing_owner_token = run_tsf_with_api_url(
        Url::parse("http://127.0.0.1:1").expect("api URL"),
        ["visibility", &read_only_url, "public"],
        None,
    )
    .await;
    assert!(
        !missing_owner_token.status.success(),
        "stdout={}",
        missing_owner_token.stdout
    );
    assert!(
        missing_owner_token
            .stderr
            .contains("stream URL does not contain an owner token"),
        "stderr={}",
        missing_owner_token.stderr
    );
}

#[tokio::test]
async fn cli_reports_rest_errors_without_raw_json_body() {
    let server = TestServer::start().await;

    let public = run_tsf(&server, ["new", "--public"], None).await;
    assert!(public.status.success(), "stderr={}", public.stderr);
    let lines = public
        .stdout
        .lines()
        .filter_map(|line| line.split_once('='))
        .collect::<HashMap<_, _>>();
    let owner_url = lines.get("o").expect("owner URL");
    let bad_owner_url = owner_url
        .split_once("#o=")
        .map(|(prefix, _token)| format!("{prefix}#o=bad-owner-secret"))
        .expect("owner fragment");

    let output = run_tsf(&server, ["visibility", &bad_owner_url, "private"], None).await;

    assert!(!output.status.success(), "stdout={}", output.stdout);
    assert!(
        output.stderr.contains("forbidden: owner token required"),
        "stderr={}",
        output.stderr
    );
    assert!(
        !output.stderr.contains(r#""error""#),
        "stderr={}",
        output.stderr
    );
    server.abort();
}

#[tokio::test]
async fn replay_suppresses_reused_writer_sequences() {
    let server = FakeReadServer::start_replay_transcript().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let output =
        run_tsf_with_api_url(server.api_url.clone(), ["replay", read_url.as_str()], None).await;

    assert!(output.status.success(), "stderr={}", output.stderr);
    assert_eq!(output.stdout, "dedupe\nstable\n");
    let attempts = server.read_attempts();
    assert_eq!(attempts.len(), 1);
    assert_eq!(attempts[0].bearer_token, "read-secret");
    assert_eq!(
        attempts[0].query.get("seq_num").map(String::as_str),
        Some("0")
    );
    assert_eq!(
        attempts[0].query.get("until").map(String::as_str),
        Some("3")
    );
    assert_eq!(attempts[0].query.get("tail_offset"), None);

    server.abort();
}

#[tokio::test]
async fn replay_rejects_logical_records_above_configured_limit() {
    let server = FakeReadServer::start_replay_split_record().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let output = run_tsf_with_api_url(
        server.api_url.clone(),
        [
            "replay",
            "--max-logical-record-bytes",
            "4",
            read_url.as_str(),
        ],
        None,
    )
    .await;

    assert!(!output.status.success(), "stdout={}", output.stdout);
    assert!(
        output
            .stderr
            .contains("failed to assemble transcript record"),
        "stderr={}",
        output.stderr
    );
    assert!(
        output
            .stderr
            .contains("logical record is 5 bytes; maximum is 4"),
        "stderr={}",
        output.stderr
    );
    server.abort();
}

#[tokio::test]
async fn replay_selector_flags_are_sent_as_bounded_read_query() {
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let seq_server = FakeReadServer::start_replay_transcript().await;
    let seq_output = run_tsf_with_api_url(
        seq_server.api_url.clone(),
        ["replay", "--seq-num", "2", read_url.as_str()],
        None,
    )
    .await;

    assert!(seq_output.status.success(), "stderr={}", seq_output.stderr);
    let attempts = seq_server.read_attempts();
    assert_eq!(attempts.len(), 1);
    assert_eq!(
        attempts[0].query.get("seq_num").map(String::as_str),
        Some("2")
    );
    assert_eq!(
        attempts[0].query.get("until").map(String::as_str),
        Some("3")
    );
    assert_eq!(
        attempts[0].query.get("count").map(String::as_str),
        Some("2")
    );
    assert_eq!(attempts[0].query.get("timestamp"), None);
    seq_server.abort();

    let timestamp_server = FakeReadServer::start_replay_transcript().await;
    let timestamp_output = run_tsf_with_api_url(
        timestamp_server.api_url.clone(),
        ["replay", "--timestamp", "1781717406000", read_url.as_str()],
        None,
    )
    .await;

    assert!(
        timestamp_output.status.success(),
        "stderr={}",
        timestamp_output.stderr
    );
    let attempts = timestamp_server.read_attempts();
    assert_eq!(attempts.len(), 1);
    assert_eq!(
        attempts[0].query.get("timestamp").map(String::as_str),
        Some("1781717406000")
    );
    assert_eq!(
        attempts[0].query.get("until").map(String::as_str),
        Some("3")
    );
    assert_eq!(attempts[0].query.get("count"), None);
    assert_eq!(attempts[0].query.get("seq_num"), None);
    timestamp_server.abort();

    let count_server = FakeReadServer::start_replay_binary().await;
    let count_output = run_tsf_bytes_with_api_url(
        count_server.api_url.clone(),
        ["replay", "--count", "1", read_url.as_str()],
    )
    .await;

    assert!(
        count_output.status.success(),
        "stderr={:?}",
        count_output.stderr
    );
    let attempts = count_server.read_attempts();
    assert_eq!(attempts.len(), 1);
    assert_eq!(
        attempts[0].query.get("seq_num").map(String::as_str),
        Some("0")
    );
    assert_eq!(
        attempts[0].query.get("until").map(String::as_str),
        Some("1")
    );
    assert_eq!(
        attempts[0].query.get("count").map(String::as_str),
        Some("1")
    );
    assert_eq!(attempts[0].query.get("timestamp"), None);
    count_server.abort();
}

#[tokio::test]
async fn replay_rejects_ambiguous_start_selectors_before_connecting() {
    let server = FakeReadServer::start_replay_transcript().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let output = run_tsf_with_api_url(
        server.api_url.clone(),
        [
            "replay",
            "--seq-num",
            "5",
            "--timestamp",
            "1781717406000",
            read_url.as_str(),
        ],
        None,
    )
    .await;

    assert!(!output.status.success(), "stdout={}", output.stdout);
    assert!(
        output.stderr.contains("cannot be used with"),
        "stderr={}",
        output.stderr
    );
    assert!(server.read_attempts().is_empty());
    server.abort();
}

#[tokio::test]
async fn replay_preserves_non_utf8_stdout_bytes() {
    let server = FakeReadServer::start_replay_binary().await;
    let stream_id = "0123456789abcdefghjkmnpqrstvwxyz"
        .parse::<StreamId>()
        .expect("stream id");
    let read_url = format!("http://localhost:3000/s/{stream_id}#r=read-secret");

    let output =
        run_tsf_bytes_with_api_url(server.api_url.clone(), ["replay", read_url.as_str()]).await;

    assert!(output.status.success(), "stderr={:?}", output.stderr);
    assert_eq!(
        output.stdout,
        vec![0x00, 0xff, b'b', b'i', b'n', b'\n', 0xf0, 0x28, 0x8c, 0x28]
    );
    assert_eq!(output.stderr, Vec::<u8>::new());
    let attempts = server.read_attempts();
    assert_eq!(attempts.len(), 1);
    assert_eq!(
        attempts[0].query.get("until").map(String::as_str),
        Some("1")
    );

    server.abort();
}

#[tokio::test]
async fn owner_commands_manage_visibility_tokens_and_deletion() {
    let server = TestServer::start().await;
    let created = run_tsf(&server, ["new", "--format", "json"], None).await;
    assert!(created.status.success(), "stderr={}", created.stderr);
    let created_json: serde_json::Value =
        serde_json::from_str(&created.stdout).expect("create output");
    let owner_url = created_json["urls"]["o"].as_str().expect("owner URL");

    let visibility = run_tsf(
        &server,
        ["visibility", owner_url, "public", "--format", "json"],
        None,
    )
    .await;
    assert!(visibility.status.success(), "stderr={}", visibility.stderr);
    let visibility_json: serde_json::Value =
        serde_json::from_str(&visibility.stdout).expect("visibility output");
    assert_eq!(visibility_json["visibility"], "public");

    let issued = run_tsf(
        &server,
        [
            "token", "issue", owner_url, "--token", "r", "--format", "json",
        ],
        None,
    )
    .await;
    assert!(issued.status.success(), "stderr={}", issued.stderr);
    let issued_json: serde_json::Value =
        serde_json::from_str(&issued.stdout).expect("issue output");
    let issued_url = issued_json["url"].as_str().expect("issued URL");
    StreamLocator::parse(issued_url).expect("issued URL parses");
    let token_id = issued_json["token_id"]
        .as_str()
        .expect("token id")
        .to_owned();

    let listed = run_tsf(
        &server,
        ["token", "list", owner_url, "--format", "json"],
        None,
    )
    .await;
    assert!(listed.status.success(), "stderr={}", listed.stderr);
    let listed_json: serde_json::Value =
        serde_json::from_str(&listed.stdout).expect("token list output");
    assert_eq!(listed_json["tokens"].as_array().map(Vec::len), Some(4));
    assert_eq!(
        listed_json["tokens"]
            .as_array()
            .and_then(|tokens| tokens.iter().find(|token| token["token_id"] == token_id))
            .map(|token| &token["status"]),
        Some(&serde_json::Value::String("active".to_owned()))
    );

    let revoked = run_tsf(
        &server,
        ["token", "revoke", owner_url, token_id.as_str()],
        None,
    )
    .await;
    assert!(revoked.status.success(), "stderr={}", revoked.stderr);
    assert_eq!(revoked.stdout, "");

    let listed = run_tsf(
        &server,
        ["token", "list", owner_url, "--format", "json"],
        None,
    )
    .await;
    let listed_json: serde_json::Value =
        serde_json::from_str(&listed.stdout).expect("token list output");
    assert_eq!(
        listed_json["tokens"]
            .as_array()
            .and_then(|tokens| tokens.iter().find(|token| token["token_id"] == token_id))
            .map(|token| &token["status"]),
        Some(&serde_json::Value::String("revoked".to_owned()))
    );

    let deleted = run_tsf(&server, ["delete", owner_url], None).await;
    assert!(deleted.status.success(), "stderr={}", deleted.stderr);
    assert_eq!(deleted.stdout, "");

    let after_delete = run_tsf(&server, ["visibility", owner_url, "private"], None).await;
    assert!(
        !after_delete.status.success(),
        "visibility update unexpectedly succeeded after delete"
    );

    server.abort();
}

struct TestServer {
    api_url: Url,
    task: tokio::task::JoinHandle<()>,
}

impl TestServer {
    async fn start() -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr");
        let state = Arc::new(TestApiState::default());
        let router = Router::new()
            .route("/api/v1/streams", post(test_create_stream))
            .route(
                "/api/v1/streams/{stream_id}",
                get(test_get_stream)
                    .patch(test_update_stream)
                    .delete(test_delete_stream),
            )
            .route(
                "/api/v1/streams/{stream_id}/tail",
                get(test_get_stream_tail),
            )
            .route(
                "/api/v1/streams/{stream_id}/tokens",
                get(test_list_tokens)
                    .post(test_issue_token)
                    .delete(test_revoke_token),
            )
            .route("/api/v1/streams/{stream_id}/write", get(test_write_socket))
            .route("/api/v1/streams/{stream_id}/read", get(test_read_socket))
            .with_state(state);
        let task = tokio::spawn(async move {
            axum::serve(listener, router).await.expect("server");
        });
        Self {
            api_url: Url::parse(&format!("http://{addr}")).expect("api URL"),
            task,
        }
    }

    fn abort(self) {
        self.task.abort();
    }
}

#[derive(Default)]
struct TestApiState {
    next_stream: Mutex<u64>,
    next_token: Mutex<u64>,
    streams: Mutex<HashMap<String, TestStream>>,
}

struct TestStream {
    stream_id: StreamId,
    visibility: Visibility,
    deleted: bool,
    tokens: Vec<TestToken>,
    records: Vec<TestRecord>,
}

#[derive(Clone)]
struct TestToken {
    token_id: TokenId,
    permissions: TokenPermissions,
    token: BearerToken,
    active: bool,
}

#[derive(Clone)]
struct TestRecord {
    s2_seq_num: u64,
    writer_id: WriterId,
    writer_seq_num: u64,
    part: PartHeader,
    format: RecordFormat,
    data: Bytes,
}

async fn test_create_stream(
    State(state): State<Arc<TestApiState>>,
    Json(request): Json<CreateStreamRequest>,
) -> Json<CreateStreamResponse> {
    let stream_id = {
        let mut next_stream = state.next_stream.lock().expect("next stream lock");
        let stream_id = format!("{:032x}", *next_stream)
            .parse::<StreamId>()
            .expect("stream id");
        *next_stream += 1;
        stream_id
    };
    let requested_tokens = request.issue_tokens.unwrap_or_else(|| {
        if request.visibility == Visibility::Public {
            vec![TokenPermissions::owner(), TokenPermissions::write()]
        } else {
            vec![
                TokenPermissions::owner(),
                TokenPermissions::write(),
                TokenPermissions::read(),
            ]
        }
    });
    let tokens = requested_tokens
        .into_iter()
        .map(|permissions| test_issue_stream_token(&state, permissions))
        .collect::<Vec<_>>();
    let response_tokens = tokens
        .iter()
        .map(|token| IssuedStreamToken {
            token_id: token.token_id,
            permissions: token.permissions,
            token: token.token.clone(),
        })
        .collect::<Vec<_>>();
    let mut streams = state.streams.lock().expect("streams lock");
    streams.insert(
        stream_id.to_string(),
        TestStream {
            stream_id,
            visibility: request.visibility,
            deleted: false,
            tokens,
            records: Vec::new(),
        },
    );

    Json(CreateStreamResponse {
        stream_id,
        visibility: request.visibility,
        retention_secs: request.retention_secs.unwrap_or(864_000),
        tokens: response_tokens,
    })
}

async fn test_get_stream(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
) -> Response {
    let streams = state.streams.lock().expect("streams lock");
    let Some(stream) = streams.get(&stream_id) else {
        return test_error(StatusCode::NOT_FOUND, "not_found", "stream not found");
    };
    if stream.deleted {
        return test_error(StatusCode::CONFLICT, "conflict", "stream is deleted");
    }
    Json(test_get_stream_response(stream)).into_response()
}

async fn test_update_stream(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
    headers: HeaderMap,
    Json(request): Json<UpdateStreamRequest>,
) -> Response {
    let mut streams = state.streams.lock().expect("streams lock");
    let Some(stream) = streams.get_mut(&stream_id) else {
        return test_error(StatusCode::NOT_FOUND, "not_found", "stream not found");
    };
    if stream.deleted {
        return test_error(StatusCode::CONFLICT, "conflict", "stream is deleted");
    }
    if !test_authorized(stream, &headers, TokenPermissions::allows_owner) {
        return test_error(StatusCode::FORBIDDEN, "forbidden", "owner token required");
    }
    if let Some(visibility) = request.visibility {
        stream.visibility = visibility;
    }
    Json(test_get_stream_response(stream)).into_response()
}

async fn test_delete_stream(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
    headers: HeaderMap,
) -> Response {
    let mut streams = state.streams.lock().expect("streams lock");
    let Some(stream) = streams.get_mut(&stream_id) else {
        return test_error(StatusCode::NOT_FOUND, "not_found", "stream not found");
    };
    if stream.deleted {
        return test_error(StatusCode::CONFLICT, "conflict", "stream is deleted");
    }
    if !test_authorized(stream, &headers, TokenPermissions::allows_owner) {
        return test_error(StatusCode::FORBIDDEN, "forbidden", "owner token required");
    }
    stream.deleted = true;
    StatusCode::NO_CONTENT.into_response()
}

async fn test_get_stream_tail(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
) -> Response {
    let streams = state.streams.lock().expect("streams lock");
    let Some(stream) = streams.get(&stream_id) else {
        return test_error(StatusCode::NOT_FOUND, "not_found", "stream not found");
    };
    if stream.deleted {
        return test_error(StatusCode::CONFLICT, "conflict", "stream is deleted");
    }
    Json(StreamTailResponse {
        stream_id: stream.stream_id,
        next_s2_seq_num: stream.records.len() as u64,
        last_timestamp_ms: stream.records.last().map(|_| 1_781_717_406_000),
    })
    .into_response()
}

async fn test_issue_token(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
    headers: HeaderMap,
    Json(request): Json<IssueTokenRequest>,
) -> Response {
    let mut streams = state.streams.lock().expect("streams lock");
    let Some(stream) = streams.get_mut(&stream_id) else {
        return test_error(StatusCode::NOT_FOUND, "not_found", "stream not found");
    };
    if stream.deleted {
        return test_error(StatusCode::CONFLICT, "conflict", "stream is deleted");
    }
    if !test_authorized(stream, &headers, TokenPermissions::allows_owner) {
        return test_error(StatusCode::FORBIDDEN, "forbidden", "owner token required");
    }
    let token = test_issue_stream_token(&state, request.permissions);
    let response = IssueTokenResponse {
        token_id: token.token_id,
        permissions: token.permissions,
        token: token.token.clone(),
    };
    stream.tokens.push(token);
    Json(response).into_response()
}

async fn test_list_tokens(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
    headers: HeaderMap,
) -> Response {
    let streams = state.streams.lock().expect("streams lock");
    let Some(stream) = streams.get(&stream_id) else {
        return test_error(StatusCode::NOT_FOUND, "not_found", "stream not found");
    };
    if stream.deleted {
        return test_error(StatusCode::CONFLICT, "conflict", "stream is deleted");
    }
    if !test_authorized(stream, &headers, TokenPermissions::allows_owner) {
        return test_error(StatusCode::FORBIDDEN, "forbidden", "owner token required");
    }
    Json(ListTokensResponse {
        tokens: stream
            .tokens
            .iter()
            .map(|token| StreamTokenSummary {
                token_id: token.token_id,
                permissions: token.permissions,
                status: if token.active {
                    StreamTokenStatus::Active
                } else {
                    StreamTokenStatus::Revoked
                },
                issued_at: "2026-08-07T12:00:00.000Z".to_owned(),
                expires_at: None,
                revoked_at: (!token.active).then(|| "2026-08-07T12:01:00.000Z".to_owned()),
                is_current: token.active && token.permissions.allows_owner(),
            })
            .collect(),
    })
    .into_response()
}

async fn test_revoke_token(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
    headers: HeaderMap,
    Json(request): Json<RevokeTokenRequest>,
) -> Response {
    let mut streams = state.streams.lock().expect("streams lock");
    let Some(stream) = streams.get_mut(&stream_id) else {
        return test_error(StatusCode::NOT_FOUND, "not_found", "stream not found");
    };
    if stream.deleted {
        return test_error(StatusCode::CONFLICT, "conflict", "stream is deleted");
    }
    if !test_authorized(stream, &headers, TokenPermissions::allows_owner) {
        return test_error(StatusCode::FORBIDDEN, "forbidden", "owner token required");
    }
    for token in &mut stream.tokens {
        if token.token_id == request.token_id {
            token.active = false;
        }
    }
    StatusCode::NO_CONTENT.into_response()
}

async fn test_write_socket(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
    ws: WebSocketUpgrade,
) -> Response {
    ws.protocols([TSF_WS_PROTOCOL])
        .on_upgrade(move |socket| test_write_flow(state, stream_id, socket))
}

async fn test_write_flow(state: Arc<TestApiState>, stream_id: String, mut socket: WebSocket) {
    let Some(Ok(Message::Binary(auth))) = socket.recv().await else {
        return;
    };
    let Ok(ClientFrame::AuthWrite {
        writer_id,
        bearer_token,
    }) = ClientFrame::decode_bytes(auth)
    else {
        return;
    };
    {
        let streams = state.streams.lock().expect("streams lock");
        let Some(stream) = streams.get(&stream_id) else {
            return;
        };
        if stream.deleted
            || !stream.tokens.iter().any(|token| {
                token.active
                    && token.token.expose_secret() == bearer_token.expose_secret()
                    && token.permissions.allows_write()
            })
        {
            return;
        }
    }
    socket
        .send(Message::Binary(
            ServerFrame::Hello { version: TSF_V1 }
                .encode()
                .expect("hello"),
        ))
        .await
        .expect("send hello");

    while let Some(Ok(Message::Binary(append))) = socket.recv().await {
        let Ok(ClientFrame::AppendRecord {
            writer_seq_num,
            part,
            format,
            data,
        }) = ClientFrame::decode_bytes(append)
        else {
            return;
        };
        let s2_seq_num = {
            let mut streams = state.streams.lock().expect("streams lock");
            let Some(stream) = streams.get_mut(&stream_id) else {
                return;
            };
            let s2_seq_num = stream.records.len() as u64;
            stream.records.push(TestRecord {
                s2_seq_num,
                writer_id,
                writer_seq_num,
                part,
                format,
                data,
            });
            s2_seq_num
        };
        socket
            .send(Message::Binary(
                ServerFrame::Ack {
                    writer_seq_start: writer_seq_num,
                    writer_seq_end: writer_seq_num,
                    s2_seq_start: s2_seq_num,
                    s2_seq_end: s2_seq_num,
                }
                .encode()
                .expect("ack"),
            ))
            .await
            .expect("send ack");
    }
}

async fn test_read_socket(
    State(state): State<Arc<TestApiState>>,
    Path(stream_id): Path<String>,
    Query(query): Query<HashMap<String, String>>,
    ws: WebSocketUpgrade,
) -> Response {
    ws.protocols([TSF_WS_PROTOCOL])
        .on_upgrade(move |socket| test_read_flow(state, stream_id, query, socket))
}

async fn test_read_flow(
    state: Arc<TestApiState>,
    stream_id: String,
    query: HashMap<String, String>,
    mut socket: WebSocket,
) {
    socket
        .send(Message::Binary(
            ServerFrame::AuthRequired.encode().expect("auth required"),
        ))
        .await
        .expect("send auth required");
    let Some(Ok(Message::Binary(auth))) = socket.recv().await else {
        return;
    };
    let Ok(ClientFrame::AuthRead { bearer_token }) = ClientFrame::decode_bytes(auth) else {
        return;
    };
    let records = {
        let streams = state.streams.lock().expect("streams lock");
        let Some(stream) = streams.get(&stream_id) else {
            return;
        };
        if stream.deleted
            || !stream.tokens.iter().any(|token| {
                token.active
                    && token.token.expose_secret() == bearer_token.expose_secret()
                    && token.permissions.allows_read()
            })
        {
            return;
        }
        test_select_records(stream, &query)
    };
    socket
        .send(Message::Binary(
            ServerFrame::Hello { version: TSF_V1 }
                .encode()
                .expect("hello"),
        ))
        .await
        .expect("send hello");
    for record in records {
        socket
            .send(Message::Binary(
                ServerFrame::ReadRecord(ReadRecord {
                    s2_seq_num: record.s2_seq_num,
                    writer_id: record.writer_id,
                    writer_seq_num: record.writer_seq_num,
                    part: record.part,
                    format: record.format,
                    data: record.data,
                })
                .encode()
                .expect("read record"),
            ))
            .await
            .expect("send record");
    }
    socket
        .send(Message::Close(None))
        .await
        .expect("close read socket");
}

fn test_issue_stream_token(state: &TestApiState, permissions: TokenPermissions) -> TestToken {
    let mut next_token = state.next_token.lock().expect("next token lock");
    let token_id = format!("{:024x}", *next_token)
        .parse::<TokenId>()
        .expect("token id");
    let token = BearerToken::from(format!("secret-{:024}", *next_token));
    *next_token += 1;
    TestToken {
        token_id,
        permissions,
        token,
        active: true,
    }
}

fn test_get_stream_response(stream: &TestStream) -> StreamInfoResponse {
    StreamInfoResponse {
        stream_id: stream.stream_id,
        basin: "test-basin".to_owned(),
        visibility: stream.visibility,
        state: if stream.deleted { "deleted" } else { "active" }.to_owned(),
        retention_secs: 864_000,
        active_token_count: stream.tokens.iter().filter(|token| token.active).count(),
    }
}

fn test_authorized(
    stream: &TestStream,
    headers: &HeaderMap,
    required: impl Fn(TokenPermissions) -> bool,
) -> bool {
    let Some(bearer_token) = test_bearer_token(headers) else {
        return false;
    };
    stream.tokens.iter().any(|token| {
        token.active && token.token.expose_secret() == bearer_token && required(token.permissions)
    })
}

fn test_bearer_token(headers: &HeaderMap) -> Option<&str> {
    headers
        .get(axum::http::header::AUTHORIZATION)?
        .to_str()
        .ok()?
        .strip_prefix("Bearer ")
}

fn test_error(status: StatusCode, code: &str, message: &str) -> Response {
    (
        status,
        Json(serde_json::json!({
            "error": {
                "code": code,
                "message": message
            }
        })),
    )
        .into_response()
}

fn test_select_records(stream: &TestStream, query: &HashMap<String, String>) -> Vec<TestRecord> {
    let mut records = stream.records.clone();
    if let Some(seq_num) = query
        .get("seq_num")
        .and_then(|value| value.parse::<u64>().ok())
    {
        records.retain(|record| record.s2_seq_num >= seq_num);
    } else if let Some(tail_offset) = query
        .get("tail_offset")
        .and_then(|value| value.parse::<usize>().ok())
    {
        let start = records.len().saturating_sub(tail_offset);
        records = records[start..].to_vec();
    }
    if let Some(until) = query
        .get("until")
        .and_then(|value| value.parse::<u64>().ok())
    {
        records.retain(|record| record.s2_seq_num <= until);
    }
    if let Some(count) = query
        .get("count")
        .and_then(|value| value.parse::<usize>().ok())
    {
        records.truncate(count);
    }
    records
}

struct CommandOutput {
    status: std::process::ExitStatus,
    stdout: String,
    stderr: String,
}

async fn run_tsf<const N: usize>(
    server: &TestServer,
    args: [&str; N],
    stdin: Option<&str>,
) -> CommandOutput {
    run_tsf_with_api_url(server.api_url.clone(), args, stdin).await
}

async fn run_tsf_with_api_url<const N: usize>(
    api_url: Url,
    args: [&str; N],
    stdin: Option<&str>,
) -> CommandOutput {
    let api_url = api_url.to_string();
    let args = args.map(str::to_owned).to_vec();
    let stdin = stdin.map(str::to_owned);
    tokio::task::spawn_blocking(move || run_tsf_blocking(api_url, args, stdin))
        .await
        .expect("tsf runner")
}

fn run_tsf_blocking(api_url: String, args: Vec<String>, stdin: Option<String>) -> CommandOutput {
    let mut command = Command::new(env!("CARGO_BIN_EXE_tsf"));
    command
        .arg("--api-url")
        .arg(api_url)
        .arg("--web-url")
        .arg("http://localhost:3000")
        .args(args)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());

    if stdin.is_some() {
        command.stdin(Stdio::piped());
    }

    let mut child = command.spawn().expect("spawn tsf");
    if let Some(stdin) = stdin {
        use std::io::Write as _;
        child
            .stdin
            .as_mut()
            .expect("stdin")
            .write_all(stdin.as_bytes())
            .expect("write stdin");
    }
    let output = child.wait_with_output().expect("tsf output");
    CommandOutput {
        status: output.status,
        stdout: String::from_utf8(output.stdout).expect("stdout utf8"),
        stderr: String::from_utf8(output.stderr).expect("stderr utf8"),
    }
}

struct CommandOutputBytes {
    status: std::process::ExitStatus,
    stdout: Vec<u8>,
    stderr: Vec<u8>,
}

async fn run_tsf_bytes_with_api_url<const N: usize>(
    api_url: Url,
    args: [&str; N],
) -> CommandOutputBytes {
    let api_url = api_url.to_string();
    let args = args.map(str::to_owned).to_vec();
    tokio::task::spawn_blocking(move || run_tsf_bytes_blocking(api_url, args))
        .await
        .expect("tsf byte runner")
}

fn run_tsf_bytes_blocking(api_url: String, args: Vec<String>) -> CommandOutputBytes {
    let output = Command::new(env!("CARGO_BIN_EXE_tsf"))
        .arg("--api-url")
        .arg(api_url)
        .arg("--web-url")
        .arg("http://localhost:3000")
        .args(args)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .output()
        .expect("tsf output");
    CommandOutputBytes {
        status: output.status,
        stdout: output.stdout,
        stderr: output.stderr,
    }
}

async fn run_tsf_until_stdout_contains<const N: usize>(
    api_url: Url,
    args: [&str; N],
    needle: &[u8],
    wait_for: Duration,
) -> CommandOutput {
    let mut command = TokioCommand::new(env!("CARGO_BIN_EXE_tsf"));
    command
        .arg("--api-url")
        .arg(api_url.to_string())
        .arg("--web-url")
        .arg("http://localhost:3000")
        .args(args)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let mut child = command.spawn().expect("spawn tsf");
    let mut stdout = child.stdout.take().expect("stdout");
    let mut stderr = child.stderr.take().expect("stderr");
    let needle = needle.to_vec();
    let stdout_task = tokio::spawn(async move {
        let mut output = Vec::new();
        let mut byte = [0_u8; 1];
        loop {
            let read = stdout.read(&mut byte).await.expect("read stdout");
            if read == 0 {
                return (output, false);
            }
            output.extend_from_slice(&byte[..read]);
            if output.windows(needle.len()).any(|window| window == needle) {
                return (output, true);
            }
        }
    });
    let stderr_task = tokio::spawn(async move {
        let mut output = Vec::new();
        stderr.read_to_end(&mut output).await.expect("read stderr");
        output
    });

    let (stdout, found) = match timeout(wait_for, stdout_task).await {
        Ok(result) => result.expect("stdout task"),
        Err(_) => {
            child.kill().await.expect("kill timed out tsf");
            let status = child.wait().await.expect("wait for timed out tsf");
            let stdout = Vec::new();
            let stderr = timeout(Duration::from_secs(1), stderr_task)
                .await
                .expect("timed out waiting for stderr")
                .expect("stderr task");
            panic!(
                "timed out waiting for stdout; status={status}; stdout={}; stderr={}",
                String::from_utf8_lossy(&stdout),
                String::from_utf8_lossy(&stderr)
            );
        }
    };
    assert!(
        found,
        "process exited before stdout contained expected bytes: {}",
        String::from_utf8_lossy(&stdout)
    );
    child.kill().await.expect("kill tsf tail");
    let status = child.wait().await.expect("wait for tsf tail");
    let stderr = timeout(Duration::from_secs(1), stderr_task)
        .await
        .expect("timed out waiting for stderr")
        .expect("stderr task");

    CommandOutput {
        status,
        stdout: String::from_utf8(stdout).expect("stdout utf8"),
        stderr: String::from_utf8(stderr).expect("stderr utf8"),
    }
}

fn normalize_created_stream_output(output: &str) -> String {
    output
        .lines()
        .map(|line| {
            if line.starts_with("stream_id=") {
                "stream_id=<stream_id>".to_owned()
            } else if line.starts_with("retention_secs=") {
                "retention_secs=<retention_secs>".to_owned()
            } else if let Some((permission, _url)) = line.split_once('=')
                && matches!(permission, "o" | "w" | "r" | "rw")
            {
                format!("{permission}=<url>")
            } else {
                line.to_owned()
            }
        })
        .collect::<Vec<_>>()
        .join("\n")
        + "\n"
}

fn assert_created_output_urls_parse(output: &str, expected_permissions: &[&str]) {
    let lines = output
        .lines()
        .filter_map(|line| line.split_once('='))
        .collect::<HashMap<_, _>>();
    let stream_id = lines.get("stream_id").expect("stream_id line");
    for permission in expected_permissions {
        let url = lines.get(permission).expect("permission URL line");
        let locator = StreamLocator::parse(url).expect("stream URL parses");
        assert_eq!(locator.stream_id.to_string(), *stream_id);
        let permissions = permission
            .parse::<TokenPermissions>()
            .expect("expected permission parses");
        assert!(
            locator
                .token_with(|token_permissions| token_permissions == permissions)
                .is_some(),
            "URL for {permission} did not contain a matching token"
        );
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct AppendAttempt {
    writer_id: WriterId,
    bearer_token: String,
    writer_seq_num: u64,
    part: PartHeader,
    format: RecordFormat,
    data: Bytes,
}

#[derive(Default)]
struct FakeWriteState {
    append_attempts: Mutex<Vec<AppendAttempt>>,
}

struct FakeWriteServer {
    api_url: Url,
    state: Arc<FakeWriteState>,
    task: tokio::task::JoinHandle<()>,
}

impl FakeWriteServer {
    async fn start() -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr");
        let state = Arc::new(FakeWriteState::default());
        let router = Router::new()
            .route("/api/v1/streams/{stream_id}/write", get(fake_write_socket))
            .with_state(state.clone());
        let task = tokio::spawn(async move {
            axum::serve(listener, router).await.expect("fake server");
        });
        Self {
            api_url: Url::parse(&format!("http://{addr}")).expect("api URL"),
            state,
            task,
        }
    }

    fn append_attempts(&self) -> Vec<AppendAttempt> {
        self.state
            .append_attempts
            .lock()
            .expect("append attempts lock")
            .clone()
    }

    fn abort(self) {
        self.task.abort();
    }
}

async fn fake_write_socket(
    State(state): State<Arc<FakeWriteState>>,
    ws: WebSocketUpgrade,
) -> Response {
    ws.protocols([TSF_WS_PROTOCOL])
        .on_upgrade(move |socket| fake_write_flow(state, socket))
}

async fn fake_write_flow(state: Arc<FakeWriteState>, mut socket: WebSocket) {
    let Some(Ok(Message::Binary(auth))) = socket.recv().await else {
        return;
    };
    let ClientFrame::AuthWrite {
        writer_id,
        bearer_token,
    } = ClientFrame::decode_bytes(auth).expect("auth write")
    else {
        return;
    };
    socket
        .send(Message::Binary(
            ServerFrame::Hello { version: TSF_V1 }
                .encode()
                .expect("hello"),
        ))
        .await
        .expect("send hello");

    let Some(Ok(Message::Binary(append))) = socket.recv().await else {
        return;
    };
    let ClientFrame::AppendRecord {
        writer_seq_num,
        part,
        format,
        data,
    } = ClientFrame::decode_bytes(append).expect("append")
    else {
        return;
    };
    let attempt_count = {
        let mut attempts = state.append_attempts.lock().expect("append attempts lock");
        attempts.push(AppendAttempt {
            writer_id,
            bearer_token: bearer_token.expose_secret().to_owned(),
            writer_seq_num,
            part,
            format,
            data,
        });
        attempts.len()
    };

    if attempt_count == 1 {
        socket
            .send(Message::Close(None))
            .await
            .expect("close first attempt");
        return;
    }

    socket
        .send(Message::Binary(
            ServerFrame::Ack {
                writer_seq_start: writer_seq_num,
                writer_seq_end: writer_seq_num,
                s2_seq_start: 0,
                s2_seq_end: 0,
            }
            .encode()
            .expect("ack"),
        ))
        .await
        .expect("send ack");
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct ReadAttempt {
    bearer_token: String,
    query: HashMap<String, String>,
}

#[derive(Default)]
struct FakeReadState {
    read_attempts: Mutex<Vec<ReadAttempt>>,
    mode: FakeReadMode,
}

#[derive(Clone, Copy, Default)]
enum FakeReadMode {
    #[default]
    Reconnect,
    BrokenRead,
    ServerShutdown,
    ReplayTranscript,
    ReplayBinary,
    ReplaySplitRecord,
}

struct FakeReadServer {
    api_url: Url,
    state: Arc<FakeReadState>,
    task: tokio::task::JoinHandle<()>,
}

impl FakeReadServer {
    async fn start() -> Self {
        Self::start_with_mode(FakeReadMode::Reconnect).await
    }

    async fn start_broken_read() -> Self {
        Self::start_with_mode(FakeReadMode::BrokenRead).await
    }

    async fn start_server_shutdown() -> Self {
        Self::start_with_mode(FakeReadMode::ServerShutdown).await
    }

    async fn start_replay_transcript() -> Self {
        Self::start_with_mode(FakeReadMode::ReplayTranscript).await
    }

    async fn start_replay_binary() -> Self {
        Self::start_with_mode(FakeReadMode::ReplayBinary).await
    }

    async fn start_replay_split_record() -> Self {
        Self::start_with_mode(FakeReadMode::ReplaySplitRecord).await
    }

    async fn start_with_mode(mode: FakeReadMode) -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr");
        let state = Arc::new(FakeReadState {
            read_attempts: Mutex::new(Vec::new()),
            mode,
        });
        let router = Router::new()
            .route("/api/v1/streams/{stream_id}/read", get(fake_read_socket))
            .route("/api/v1/streams/{stream_id}/tail", get(fake_read_tail))
            .with_state(state.clone());
        let task = tokio::spawn(async move {
            axum::serve(listener, router).await.expect("fake server");
        });
        Self {
            api_url: Url::parse(&format!("http://{addr}")).expect("api URL"),
            state,
            task,
        }
    }

    fn read_attempts(&self) -> Vec<ReadAttempt> {
        self.state
            .read_attempts
            .lock()
            .expect("read attempts lock")
            .clone()
    }

    fn abort(self) {
        self.task.abort();
    }
}

async fn fake_read_tail(
    State(state): State<Arc<FakeReadState>>,
    Path(stream_id): Path<String>,
) -> Json<serde_json::Value> {
    let next_s2_seq_num = match state.mode {
        FakeReadMode::Reconnect | FakeReadMode::BrokenRead | FakeReadMode::ServerShutdown => 2,
        FakeReadMode::ReplayTranscript => 4,
        FakeReadMode::ReplayBinary => 2,
        FakeReadMode::ReplaySplitRecord => 2,
    };
    Json(serde_json::json!({
        "stream_id": stream_id,
        "next_s2_seq_num": next_s2_seq_num,
        "last_timestamp_ms": 1781717406000_u64
    }))
}

async fn fake_read_socket(
    State(state): State<Arc<FakeReadState>>,
    Query(query): Query<HashMap<String, String>>,
    ws: WebSocketUpgrade,
) -> Response {
    ws.protocols([TSF_WS_PROTOCOL])
        .on_upgrade(move |socket| fake_read_flow(state, query, socket))
}

async fn fake_read_flow(
    state: Arc<FakeReadState>,
    query: HashMap<String, String>,
    mut socket: WebSocket,
) {
    socket
        .send(Message::Binary(
            ServerFrame::AuthRequired.encode().expect("auth required"),
        ))
        .await
        .expect("send auth required");
    let Some(Ok(Message::Binary(auth))) = socket.recv().await else {
        return;
    };
    let ClientFrame::AuthRead { bearer_token } =
        ClientFrame::decode_bytes(auth).expect("auth read")
    else {
        return;
    };
    let attempt_count = {
        let mut attempts = state.read_attempts.lock().expect("read attempts lock");
        attempts.push(ReadAttempt {
            bearer_token: bearer_token.expose_secret().to_owned(),
            query,
        });
        attempts.len()
    };
    socket
        .send(Message::Binary(
            ServerFrame::Hello { version: TSF_V1 }
                .encode()
                .expect("hello"),
        ))
        .await
        .expect("send hello");

    match state.mode {
        FakeReadMode::Reconnect => {
            if attempt_count == 1 {
                send_read_record(&mut socket, 0, 0, b"first\n").await;
                socket
                    .send(Message::Binary(
                        ServerFrame::ReconnectAdvised { deadline_secs: 0 }
                            .encode()
                            .expect("reconnect advised"),
                    ))
                    .await
                    .expect("send reconnect advised");
            } else {
                send_read_record(&mut socket, 1, 1, b"second\n").await;
            }
        }
        FakeReadMode::BrokenRead => {
            if attempt_count == 1 {
                send_read_record(&mut socket, 0, 0, b"first\n").await;
            } else {
                send_read_record(&mut socket, 1, 1, b"second\n").await;
            }
        }
        FakeReadMode::ServerShutdown => {
            if attempt_count == 1 {
                send_read_record(&mut socket, 0, 0, b"first\n").await;
                socket
                    .send(Message::Close(Some(CloseFrame {
                        code: 1001,
                        reason: "server_shutdown".into(),
                    })))
                    .await
                    .expect("close for server shutdown");
            } else {
                send_read_record(&mut socket, 1, 1, b"second\n").await;
            }
        }
        FakeReadMode::ReplayTranscript => {
            send_read_record(&mut socket, 0, 0, b"dedupe\n").await;
            send_read_record(&mut socket, 1, 0, b"dedupe\n").await;
            send_read_record(&mut socket, 2, 1, b"stable\n").await;
            send_read_record(&mut socket, 3, 1, b"changed\n").await;
            socket
                .send(Message::Close(None))
                .await
                .expect("close replay socket");
        }
        FakeReadMode::ReplayBinary => {
            send_read_record_with_format(
                &mut socket,
                0,
                0,
                RecordFormat::Bytes,
                &[0x00, 0xff, b'b', b'i', b'n', b'\n'],
            )
            .await;
            send_read_record_with_format(
                &mut socket,
                1,
                1,
                RecordFormat::Bytes,
                &[0xf0, 0x28, 0x8c, 0x28],
            )
            .await;
            socket
                .send(Message::Close(None))
                .await
                .expect("close binary replay socket");
        }
        FakeReadMode::ReplaySplitRecord => {
            send_read_part(
                &mut socket,
                0,
                0,
                PartHeader::new(0, false).expect("part"),
                b"hel",
            )
            .await;
            send_read_part(
                &mut socket,
                1,
                1,
                PartHeader::new(1, true).expect("part"),
                b"lo",
            )
            .await;
            socket
                .send(Message::Close(None))
                .await
                .expect("close split replay socket");
        }
    }
}

async fn send_read_record(
    socket: &mut WebSocket,
    s2_seq_num: u64,
    writer_seq_num: u64,
    data: &[u8],
) {
    send_read_record_with_format(
        socket,
        s2_seq_num,
        writer_seq_num,
        RecordFormat::Transcript,
        data,
    )
    .await
}

async fn send_read_record_with_format(
    socket: &mut WebSocket,
    s2_seq_num: u64,
    writer_seq_num: u64,
    format: RecordFormat,
    data: &[u8],
) {
    socket
        .send(Message::Binary(
            ServerFrame::ReadRecord(ReadRecord {
                s2_seq_num,
                writer_id: WriterId::from_bytes([7; WriterId::BYTE_LEN]),
                writer_seq_num,
                part: PartHeader::unsplit(),
                format,
                data: Bytes::copy_from_slice(data),
            })
            .encode()
            .expect("read record"),
        ))
        .await
        .expect("send read record");
}

async fn send_read_part(
    socket: &mut WebSocket,
    s2_seq_num: u64,
    writer_seq_num: u64,
    part: PartHeader,
    data: &[u8],
) {
    socket
        .send(Message::Binary(
            ServerFrame::ReadRecord(ReadRecord {
                s2_seq_num,
                writer_id: WriterId::from_bytes([7; WriterId::BYTE_LEN]),
                writer_seq_num,
                part,
                format: RecordFormat::Transcript,
                data: Bytes::copy_from_slice(data),
            })
            .encode()
            .expect("read record"),
        ))
        .await
        .expect("send read part");
}