use serde_json::{json, Value};
use std::time::Duration;

const PDS_URL: &str = "http://127.0.0.1:2583";

async fn client() -> reqwest::Client {
    reqwest::Client::builder()
        .timeout(Duration::from_secs(5))
        .build()
        .unwrap()
}

async fn wait_for_pds() -> bool {
    let c = client().await;
    for _ in 0..20 {
        if let Ok(r) = c.get(format!("{}/healthz", PDS_URL)).send().await {
            if r.status().is_success() {
                return true;
            }
        }
        tokio::time::sleep(Duration::from_millis(250)).await;
    }
    false
}

#[tokio::test]
async fn describe_server() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let r: Value = c
        .get(format!("{}/xrpc/com.atproto.server.describeServer", PDS_URL))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert!(r["did"].is_string());
    assert!(r["available_user_domains"].is_array());
    assert_eq!(r["invite_code_required"], json!(false));
}

#[tokio::test]
async fn create_account_session_refresh_resolve() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;

    let handle = format!("itest_{}.maarcadetweet.local", uuid::Uuid::new_v4().simple());
    let pw = "hunter2hunter2";

    let r: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({
            "handle": handle,
            "email": "i@test.com",
            "password": pw,
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert!(r["did"].is_string(), "createAccount: {:?}", r);
    assert_eq!(r["handle"], json!(handle));
    assert!(r["access_jwt"].is_string());
    let did = r["did"].as_str().unwrap().to_string();
    let access = r["access_jwt"].as_str().unwrap().to_string();
    let refresh = r["refresh_jwt"].as_str().unwrap().to_string();

    let r2: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createSession", PDS_URL))
        .json(&json!({"identifier": handle, "password": pw}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(r2["did"], json!(did));
    assert!(r2["access_jwt"].is_string());

    let r3: Value = c
        .post(format!("{}/xrpc/com.atproto.server.refreshSession", PDS_URL))
        .json(&json!({"refresh_jwt": refresh}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert!(r3["access_jwt"].is_string());
    assert_eq!(r3["did"], json!(did));

    let r4: Value = c
        .post(format!("{}/xrpc/com.atproto.identity.resolveHandle", PDS_URL))
        .json(&json!({"handle": handle}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(r4["did"], json!(did));

    let _ = access;
}

#[tokio::test]
async fn create_record_enforces_160_chars() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let handle = format!("rec_{}.maarcadetweet.local", uuid::Uuid::new_v4().simple());

    let acc: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({"handle": handle, "password": "hunter2hunter2"}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let did = acc["did"].as_str().unwrap().to_string();
    let jwt = acc["access_jwt"].as_str().unwrap().to_string();

    let ok_resp = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {"text": "ok 140 chars total, fits.", "createdAt": "2026-07-01T12:00:00Z"},
        }))
        .send()
        .await
        .unwrap();
    let ok_body: Value = ok_resp.json().await.unwrap();
    assert!(ok_body["uri"].is_string(), "expected uri, got: {:?}", ok_body);

    let long = "x".repeat(161);
    let r401 = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {"text": long, "createdAt": "2026-07-01T12:00:00Z"},
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r401.status().as_u16(), 400);
    let body: Value = r401.json().await.unwrap();
    assert_eq!(body["error"], json!("InvalidRequest"));
    assert!(body["message"]
        .as_str()
        .unwrap()
        .contains("max length"));

    let r_400_empty = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {"text": "", "createdAt": "2026-07-01T12:00:00Z"},
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r_400_empty.status().as_u16(), 400);

    let r_400_missing = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {"text": "x"},
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r_400_missing.status().as_u16(), 400);

    let r_unknown = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.unknown.thing",
            "record": {"text": "x", "createdAt": "2026-07-01T12:00:00Z"},
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r_unknown.status().as_u16(), 400);

    let r_noauth = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {"text": "x", "createdAt": "2026-07-01T12:00:00Z"},
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r_noauth.status().as_u16(), 401);
}

#[tokio::test]
async fn rejects_duplicate_handle() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let handle = format!("dup_{}.maarcadetweet.local", uuid::Uuid::new_v4().simple());
    let r1: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({"handle": handle, "password": "hunter2hunter2"}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert!(r1["did"].is_string());
    let r2 = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({"handle": handle, "password": "hunter2hunter2"}))
        .send()
        .await
        .unwrap();
    assert_eq!(r2.status().as_u16(), 409);
}

#[tokio::test]
async fn rejects_short_password() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let r = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({
            "handle": format!("sp_{}.maarcadetweet.local", uuid::Uuid::new_v4().simple()),
            "password": "short"
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r.status().as_u16(), 400);
}

// -- sync endpoint tests ---------------------------------------------------

/// Minimal CAR v1 parser used to inspect what the server returned. This is
/// intentionally simple — it just extracts the header, root CIDs and (cid,
/// block) pairs. The point of the tests is to verify the on-the-wire format,
/// not to re-implement a full CAR library.
#[derive(Debug, Default)]
struct ParsedCar {
    version: u64,
    roots: Vec<String>,
    blocks: Vec<(String, Vec<u8>)>,
}

fn parse_car(bytes: &[u8]) -> ParsedCar {
    let mut out = ParsedCar::default();
    let mut p = 0usize;

    // Read a LEB128 varint.
    fn read_varint(bytes: &[u8], pos: &mut usize) -> u64 {
        let mut value: u64 = 0;
        let mut shift = 0u32;
        loop {
            let b = bytes[*pos];
            *pos += 1;
            value |= ((b & 0x7f) as u64) << shift;
            if b & 0x80 == 0 {
                return value;
            }
            shift += 7;
        }
    }

    // Read a CBOR "head": single byte for value <= 23, otherwise head +
    // 1/2/4/8 extra bytes for info 24/25/26/27. Returns (major, value,
    // bytes_consumed).
    fn read_cbor_head(bytes: &[u8], pos: usize) -> (u8, u64, usize) {
        let first = bytes[pos];
        let major = first >> 5;
        let info = first & 0x1f;
        let (value, extra) = match info {
            0..=23 => (info as u64, 0usize),
            24 => (bytes[pos + 1] as u64, 1),
            25 => (((bytes[pos + 1] as u64) << 8) | (bytes[pos + 2] as u64), 2),
            26 => (
                ((bytes[pos + 1] as u64) << 24)
                    | ((bytes[pos + 2] as u64) << 16)
                    | ((bytes[pos + 3] as u64) << 8)
                    | (bytes[pos + 4] as u64),
                4,
            ),
            27 => {
                let mut n = 0u64;
                for i in 0..8 {
                    n = (n << 8) | (bytes[pos + 1 + i] as u64);
                }
                (n, 8)
            }
            other => panic!("unsupported CBOR info {other}"),
        };
        (major, value, 1 + extra)
    }

    // Header
    let header_len = read_varint(bytes, &mut p) as usize;
    let _header_end = p + header_len;
    let (maj, n_items, consumed) = read_cbor_head(bytes, p);
    assert_eq!(maj, 5, "header must be a CBOR map");
    p += consumed;
    assert_eq!(n_items, 2, "header must have 2 keys");

    for _ in 0..2 {
        let (maj, n, c) = read_cbor_head(bytes, p);
        assert_eq!(maj, 3, "key must be a text string");
        p += c;
        let key = std::str::from_utf8(&bytes[p..p + n as usize])
            .unwrap()
            .to_string();
        p += n as usize;
        if key == "version" {
            let (maj, v, c) = read_cbor_head(bytes, p);
            assert_eq!(maj, 0, "version must be an unsigned int");
            p += c;
            out.version = v;
        } else if key == "roots" {
            let (maj, n, c) = read_cbor_head(bytes, p);
            assert_eq!(maj, 4, "roots must be a CBOR array");
            p += c;
            for _ in 0..n {
                let (maj, _, c) = read_cbor_head(bytes, p);
                assert_eq!(maj, 6, "root CID must be a tagged value");
                p += c;
                let (maj, ln, c) = read_cbor_head(bytes, p);
                assert_eq!(maj, 2, "root CID must be a byte string");
                p += c;
                let cid_bytes = &bytes[p..p + ln as usize];
                let cid_hex: String = cid_bytes
                    .iter()
                    .map(|b| format!("{:02x}", b))
                    .collect();
                p += ln as usize;
                out.roots.push(format!("raw:{}", cid_hex));
            }
        } else {
            panic!("unexpected header key {key}");
        }
    }
    // Body sections
    while p < bytes.len() {
        let section_len = read_varint(bytes, &mut p) as usize;
        let section_end = p + section_len;
        // CID = varint version + varint codec + (multihash = code + size + digest)
        let cid_start = p;
        let _v = read_varint(bytes, &mut p);
        let _c = read_varint(bytes, &mut p);
        // Multihash: read code, then size, then `size` bytes of digest.
        let _mh_code = read_varint(bytes, &mut p);
        let mh_size = read_varint(bytes, &mut p) as usize;
        let cid_end = p;
        p += mh_size;
        let cid_hex: String = bytes[cid_start..cid_end + mh_size]
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect();
        let data = bytes[p..section_end].to_vec();
        p = section_end;
        out.blocks.push((cid_hex, data));
    }
    out
}

/// Create a fresh user and a few records. Returns the http client, the
/// user's DID, the access JWT, and the list of record value CIDs.
async fn fresh_user_with_records() -> (reqwest::Client, String, String, Vec<String>) {
    let c = client().await;
    let handle = format!("sync_{}.maarcadetweet.local", uuid::Uuid::new_v4().simple());
    let acc: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({"handle": handle, "password": "hunter2hunter2"}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let did = acc["did"].as_str().unwrap().to_string();
    let jwt = acc["access_jwt"].as_str().unwrap().to_string();

    let mut record_cids = Vec::new();
    for i in 0..3 {
        let resp: Value = c
            .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
            .bearer_auth(&jwt)
            .json(&json!({
                "repo": did,
                "collection": "app.twi.post",
                "record": {
                    "text": format!("hello sync #{i}"),
                    "createdAt": "2026-07-01T12:00:00Z",
                },
            }))
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        assert!(resp["uri"].is_string(), "createRecord: {:?}", resp);
        record_cids.push(resp["cid"].as_str().unwrap().to_string());
    }
    (c, did, jwt, record_cids)
}

#[tokio::test]
async fn sync_get_repo_returns_valid_car() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, _jwt, record_cids) = fresh_user_with_records().await;

    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRepo?did={}",
            PDS_URL, did
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    assert_eq!(
        resp.headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .unwrap_or(""),
        "application/vnd.ipld.car"
    );
    let bytes = resp.bytes().await.unwrap();
    assert!(!bytes.is_empty());

    let parsed = parse_car(&bytes);
    assert_eq!(parsed.version, 1);
    assert_eq!(parsed.roots.len(), 1);

    // Every record CID we created should appear as a block in the CAR.
    for cid in &record_cids {
        // The CAR stores the raw CID bytes; we re-encode the multibase
        // string into the same raw form to compare.
        let cid_obj: cid::Cid = cid.parse().unwrap();
        let raw_hex: String = cid_obj
            .to_bytes()
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect();
        assert!(
            parsed.blocks.iter().any(|(c, _)| c == &raw_hex),
            "CAR should contain record {cid} (raw: {raw_hex})"
        );
    }

    // The CAR should also contain the MST root block (DAG-CBOR of a node)
    // and the head commit block. The MST root is a `tag(42) + cid_link`
    // node, and the head commit is a DAG-CBOR map with did/version/rev.
    assert!(
        parsed.blocks.len() >= 5,
        "CAR should have at least head commit + MST root + 3 records; got {}",
        parsed.blocks.len()
    );
}

#[tokio::test]
async fn sync_get_repo_missing_did_returns_400() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRepo?did=did:plc:nope%sinvalid",
            PDS_URL
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 400);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["error"], json!("RepoNotFound"));
}

#[tokio::test]
async fn sync_get_blocks_returns_requested_cids() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, _jwt, record_cids) = fresh_user_with_records().await;

    // First fetch getRepo so we can pick out arbitrary blocks (e.g. MST
    // nodes) in addition to record value blocks.
    let repo_bytes = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRepo?did={}",
            PDS_URL, did
        ))
        .send()
        .await
        .unwrap()
        .bytes()
        .await
        .unwrap();
    let _parsed = parse_car(&repo_bytes);

    // Ask getBlocks for just one record CID. We pick the first record we
    // created, which we know is in the repo.
    let some_record = &record_cids[0];

    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getBlocks?did={}&cids={}",
            PDS_URL, did, some_record
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    assert_eq!(
        resp.headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .unwrap_or(""),
        "application/vnd.ipld.car"
    );
    let bytes = resp.bytes().await.unwrap();
    let parsed = parse_car(&bytes);
    // The CAR should contain exactly the blocks we requested (modulo the
    // dedup behaviour — at minimum the record block).
    let target_record_hex = {
        let cid_obj: cid::Cid = some_record.parse().unwrap();
        cid_obj
            .to_bytes()
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect::<String>()
    };
    assert!(
        parsed
            .blocks
            .iter()
            .any(|(c, _)| c == &target_record_hex),
        "getBlocks should include the requested record; got blocks: {:?}",
        parsed.blocks.iter().map(|(c, _)| c).collect::<Vec<_>>()
    );
}

#[tokio::test]
async fn sync_get_blocks_invalid_cid_returns_400() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getBlocks?did=did:plc:anything&cids=not-a-cid",
            PDS_URL
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 400);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["error"], json!("InvalidRequest"));
}

#[tokio::test]
async fn sync_get_latest_commit_returns_json() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, _jwt, _cids) = fresh_user_with_records().await;

    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getLatestCommit?did={}",
            PDS_URL, did
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    let cid_str = body["cid"].as_str().expect("cid should be a string");
    let rev_str = body["rev"].as_str().expect("rev should be a string");
    let parsed: cid::Cid = cid_str.parse().expect("cid should parse");
    // The latest commit CID must be a CIDv1 DAG-CBOR block (0x71).
    assert_eq!(parsed.codec(), 0x71);
    assert!(!rev_str.is_empty());

    // The returned CID should match the one we see as the root of getRepo.
    let repo = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRepo?did={}",
            PDS_URL, did
        ))
        .send()
        .await
        .unwrap()
        .bytes()
        .await
        .unwrap();
    let parsed_repo = parse_car(&repo);
    let root_hex = parsed_repo.roots[0]
        .trim_start_matches("raw:")
        .to_string();
    let expected_hex: String = parsed
        .to_bytes()
        .iter()
        .map(|b| format!("{:02x}", b))
        .collect();
    assert_eq!(root_hex, expected_hex, "getLatestCommit cid must equal getRepo root");
}

#[tokio::test]
async fn sync_get_latest_commit_missing_did() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getLatestCommit?did=did:plc:no-such-did",
            PDS_URL
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 400);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["error"], json!("RepoNotFound"));
}

#[tokio::test]
async fn sync_get_record_returns_value_block() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, jwt, record_cids) = fresh_user_with_records().await;

    // We need an (rkey, value_cid) pair. Re-fetch the records: each
    // createRecord returns a uri+cid, and the rkey is the trailing tid.
    let first: Value = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {"text": "first post for sync getRecord", "createdAt": "2026-07-01T12:00:00Z"},
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let uri = first["uri"].as_str().unwrap();
    let value_cid = first["cid"].as_str().unwrap();
    let rkey = uri.rsplit('/').next().unwrap().to_string();

    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRecord?did={}&collection=app.twi.post&rkey={}",
            PDS_URL, did, rkey
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    assert_eq!(
        resp.headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .unwrap_or(""),
        "application/vnd.ipld.car"
    );
    let bytes = resp.bytes().await.unwrap();
    let parsed = parse_car(&bytes);
    let target_hex: String = value_cid
        .parse::<cid::Cid>()
        .unwrap()
        .to_bytes()
        .iter()
        .map(|b| format!("{:02x}", b))
        .collect();
    assert!(
        parsed.blocks.iter().any(|(c, _)| c == &target_hex),
        "getRecord CAR should include the value block {value_cid} (raw: {target_hex})"
    );
    // The record_cids list (from the helper) and our new first CID should
    // both be valid CIDs; the test really only checks the value block is
    // there, the other one is sanity.
    assert!(!record_cids.is_empty());
}

#[tokio::test]
async fn sync_get_record_missing_returns_404() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, _jwt, _cids) = fresh_user_with_records().await;
    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRecord?did={}&collection=app.twi.post&rkey=doesnotexist",
            PDS_URL, did
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 404);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["error"], json!("RecordNotFound"));
}

#[tokio::test]
async fn sync_list_repos_includes_recent_user() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, _jwt, _cids) = fresh_user_with_records().await;
    // Page through listRepos with a small limit until we see our DID.
    let mut cursor: Option<String> = None;
    let mut found = false;
    for _ in 0..50 {
        let url = match &cursor {
            Some(c) => format!(
                "{}/xrpc/com.atproto.sync.listRepos?limit=50&cursor={}",
                PDS_URL, c
            ),
            None => format!("{}/xrpc/com.atproto.sync.listRepos?limit=50", PDS_URL),
        };
        let resp = c.get(&url).send().await.unwrap();
        assert_eq!(resp.status().as_u16(), 200);
        let body: Value = resp.json().await.unwrap();
        let repos = body["repos"].as_array().expect("repos array");
        if repos.iter().any(|r| r["did"] == json!(did)) {
            found = true;
            // The matching entry should have a `head` (CID) and `rev`.
            let r = repos.iter().find(|r| r["did"] == json!(did)).unwrap();
            assert!(r["head"].is_string(), "head: {:?}", r);
            assert!(r["rev"].is_string(), "rev: {:?}", r);
            assert_eq!(r["active"], json!(true));
            // head must parse as a CID
            let _cid: cid::Cid = r["head"].as_str().unwrap().parse().unwrap();
            break;
        }
        match body["cursor"].as_str() {
            Some(c) => cursor = Some(c.to_string()),
            None => break,
        }
    }
    assert!(found, "listRepos should include the freshly created DID {did}");
}

// -- new Phase 2c tests ----------------------------------------------------

/// Create a single account with a deterministic handle, returning
/// (client, did, jwt).
async fn fresh_account(handle_suffix: &str) -> (reqwest::Client, String, String) {
    let c = client().await;
    let handle = format!(
        "lp{}_{}.maarcadetweet.local",
        handle_suffix,
        uuid::Uuid::new_v4().simple()
    );
    let acc: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({"handle": handle, "password": "hunter2hunter2"}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let did = acc["did"].as_str().unwrap().to_string();
    let jwt = acc["access_jwt"].as_str().unwrap().to_string();
    (c, did, jwt)
}

/// Recompute the CID for a CAR body + raw CID bytes and compare. Returns
/// true if they match.
fn cid_matches_block(cid_hex: &str, block: &[u8]) -> bool {
    let bytes = match hex::decode(cid_hex) {
        Ok(b) => b,
        Err(_) => return false,
    };
    let cid = match cid::Cid::read_bytes(bytes.as_slice()) {
        Ok(c) => c,
        Err(_) => return false,
    };
    let hash = cid.hash();
    let digest = hash.digest();
    let mut hasher = sha2::Sha256::new();
    use sha2::Digest;
    hasher.update(block);
    let out = hasher.finalize();
    let mut computed = [0u8; 32];
    computed.copy_from_slice(&out);
    digest == computed
}

#[tokio::test]
async fn sync_get_record_includes_mst_proof() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, jwt, _existing_cids) = fresh_user_with_records().await;

    let created: Value = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {
                "text": "proof please",
                "createdAt": "2026-07-01T12:00:00Z",
            },
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let value_cid = created["cid"].as_str().unwrap().to_string();
    let rkey = created["uri"]
        .as_str()
        .unwrap()
        .rsplit('/')
        .next()
        .unwrap()
        .to_string();

    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRecord?did={}&collection=app.twi.post&rkey={}",
            PDS_URL, did, rkey
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let bytes = resp.bytes().await.unwrap();
    let parsed = parse_car(&bytes);

    let target_hex: String = value_cid
        .parse::<cid::Cid>()
        .unwrap()
        .to_bytes()
        .iter()
        .map(|b| format!("{:02x}", b))
        .collect();
    assert!(
        parsed.blocks.iter().any(|(c, _)| c == &target_hex),
        "value block missing from CAR"
    );

    let mut dag_cbor_nodes = 0usize;
    for (cid_hex, data) in &parsed.blocks {
        assert!(
            cid_matches_block(cid_hex, data),
            "CAR block CID mismatch for {cid_hex}"
        );
        if let Ok(val) = ciborium::from_reader::<ciborium::value::Value, _>(data.as_slice()) {
            if let ciborium::value::Value::Map(_) = val {
                dag_cbor_nodes += 1;
            }
        }
    }

    assert!(
        parsed.blocks.len() >= 3,
        "expected head commit + value + at least one MST node; got {}",
        parsed.blocks.len()
    );
    assert!(
        dag_cbor_nodes >= 2,
        "expected at least 2 DAG-CBOR map blocks (head commit + MST nodes); got {}",
        dag_cbor_nodes
    );

    let repo_bytes = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRepo?did={}",
            PDS_URL, did
        ))
        .send()
        .await
        .unwrap()
        .bytes()
        .await
        .unwrap();
    let full = parse_car(&repo_bytes);
    let mut found_in_full = 0usize;
    for (cid_hex, _) in &parsed.blocks {
        if full.blocks.iter().any(|(c, _)| c == cid_hex) {
            found_in_full += 1;
        }
    }
    assert_eq!(
        found_in_full,
        parsed.blocks.len(),
        "every block in getRecord CAR must also appear in getRepo CAR"
    );
}

#[tokio::test]
async fn sync_list_repos_keyset_pagination() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let mut created_dids = Vec::new();
    for i in 0..5 {
        let (c, did, jwt) = fresh_account(&format!("page{i:02}")).await;
        let resp: Value = c
            .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
            .bearer_auth(&jwt)
            .json(&json!({
                "repo": did,
                "collection": "app.twi.post",
                "record": {
                    "text": format!("pagination seed {i}"),
                    "createdAt": "2026-07-01T12:00:00Z",
                },
            }))
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        assert!(resp["uri"].is_string(), "createRecord: {:?}", resp);
        created_dids.push(did);
    }
    let min_did = created_dids.iter().min().unwrap().clone();
    let start_cursor = did_cursor_lt(&min_did);

    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
    let mut cursor: Option<String> = Some(start_cursor);
    let mut pages = 0;
    loop {
        pages += 1;
        assert!(pages < 2000, "pagination did not terminate");
        let url = format!(
            "{}/xrpc/com.atproto.sync.listRepos?limit=2&cursor={}",
            PDS_URL,
            urlencode(cursor.as_deref().unwrap_or(""))
        );
        let resp = client().await.get(&url).send().await.unwrap();
        assert_eq!(resp.status().as_u16(), 200);
        let body: Value = resp.json().await.unwrap();
        let repos = body["repos"].as_array().expect("repos array");
        for r in repos {
            let did = r["did"].as_str().unwrap().to_string();
            assert!(
                seen.insert(did.clone()),
                "duplicate DID across pages: {did}"
            );
        }
        if created_dids.iter().all(|d| seen.contains(d)) {
            break;
        }
        match body["cursor"].as_str() {
            Some(c) => cursor = Some(c.to_string()),
            None => panic!(
                "pagination exhausted before all created DIDs were seen; missing {:?}",
                created_dids.iter().filter(|d| !seen.contains(*d)).collect::<Vec<_>>()
            ),
        }
    }
}

fn urlencode(s: &str) -> String {
    s.chars()
        .map(|c| match c {
            'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '_' | '.' | '~' => c.to_string(),
            other => format!("%{:02X}", other as u32),
        })
        .collect()
}

fn did_cursor_lt(did: &str) -> String {
    let bytes = did.as_bytes();
    let mut prefix = Vec::with_capacity(bytes.len());
    for &b in bytes {
        if b == 0 {
            prefix.push(b);
        } else {
            prefix.push(b - 1);
            break;
        }
    }
    if prefix.len() < bytes.len() {
        prefix.extend_from_slice(&bytes[prefix.len()..]);
    } else {
        prefix.push(b'_');
    }
    String::from_utf8(prefix).unwrap_or_else(|_| did.to_string())
}

#[test]
fn did_cursor_lt_is_strictly_less() {
    let s = "did:key:z16Dxyz";
    let lt = did_cursor_lt(s);
    assert!(lt.as_str() < s, "{lt} should be < {s}");
    let min = std::cmp::min(s, lt.as_str());
    assert_eq!(min, lt.as_str());
}

#[tokio::test]
async fn sync_list_repos_caps_limit() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.listRepos?limit=10000",
            PDS_URL
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    let repos = body["repos"].as_array().expect("repos array");
    assert!(
        repos.len() <= 1000,
        "limit=10000 should be capped at MAX_LIST_LIMIT=1000, got {}",
        repos.len()
    );
}

#[tokio::test]
async fn sync_get_record_returns_404_for_missing() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, _jwt, _cids) = fresh_user_with_records().await;
    let resp = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRecord?did={}&collection=app.twi.post&rkey=totallymissingkey",
            PDS_URL, did
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 404);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["error"], json!("RecordNotFound"));
}

// -- embed-carrying record tests ------------------------------------------

/// Post a record with an `app.bsky.embed.images` embed. The PDS should
/// accept it (the lexicon allows embed variants), persist the value
/// block, and return a 200 with a URI+CID. The AppView's ingest-commit
/// push is best-effort, so we don't depend on the AppView being up.
#[tokio::test]
async fn create_record_with_image_embed() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let handle = format!(
        "img_{}.maarcadetweet.local",
        uuid::Uuid::new_v4().simple()
    );
    let acc: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({
            "handle": handle,
            "password": "hunter2hunter2"
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let did = acc["did"].as_str().unwrap().to_string();
    let jwt = acc["access_jwt"].as_str().unwrap().to_string();

    let resp = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {
                "text": "check out this image",
                "createdAt": "2026-07-01T12:00:00Z",
                "embed": {
                    "$type": "app.bsky.embed.images",
                    "images": [
                        {
                            "alt": "a single image",
                            "image": {
                                "$type": "blob",
                                "ref": {"$link": "bafyreiblob1"},
                                "mimeType": "image/jpeg",
                                "size": 1024
                            },
                            "aspectRatio": {"width": 800, "height": 600}
                        }
                    ]
                }
            }
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200, "image embed post must succeed");
    let body: Value = resp.json().await.unwrap();
    assert!(body["uri"].is_string(), "createRecord: {:?}", body);
    let cid = body["cid"].as_str().unwrap();
    let uri = body["uri"].as_str().unwrap();
    assert!(cid.starts_with("bafy"), "cid should be a CID");
    assert!(uri.starts_with("at://"), "uri should be at://");

    // Round-trip via sync.getRecord to confirm the value block survived
    // CBOR encoding at the right CID. We don't decode the CBOR here —
    // the AppView's embed parsing test covers that — but we verify the
    // block is reachable from the repo so future reads succeed.
    let rkey = uri.rsplit('/').next().unwrap();
    let car = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRecord?did={}&collection=app.twi.post&rkey={}",
            PDS_URL, did, rkey
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(car.status().as_u16(), 200);
    let parsed = parse_car(&car.bytes().await.unwrap());
    let target_hex: String = cid
        .parse::<cid::Cid>()
        .unwrap()
        .to_bytes()
        .iter()
        .map(|b| format!("{:02x}", b))
        .collect();
    assert!(
        parsed.blocks.iter().any(|(c, _)| c == &target_hex),
        "value block {cid} must be in getRecord CAR"
    );
}

/// Post a record with an `app.bsky.embed.external` (link card).
#[tokio::test]
async fn create_record_with_external_embed() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let c = client().await;
    let handle = format!(
        "ext_{}.maarcadetweet.local",
        uuid::Uuid::new_v4().simple()
    );
    let acc: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({
            "handle": handle,
            "password": "hunter2hunter2"
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let did = acc["did"].as_str().unwrap().to_string();
    let jwt = acc["access_jwt"].as_str().unwrap().to_string();

    let resp = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {
                "text": "see link",
                "createdAt": "2026-07-01T12:00:00Z",
                "embed": {
                    "$type": "app.bsky.embed.external",
                    "external": {
                        "uri": "https://example.com/article",
                        "title": "An article",
                        "description": "Short description."
                    }
                }
            }
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    assert!(body["uri"].is_string(), "createRecord: {:?}", body);
}

// -- like / deleteRecord tests ---------------------------------------------

/// Helper: build a `(did, jwt)` pair against a fresh account. Reused
/// by the like/delete tests; the handle is unique-per-call so we
/// don't collide with parallel test runs.
async fn fresh_user(prefix: &str) -> (reqwest::Client, String, String) {
    let c = client().await;
    let handle = format!(
        "{}_{}.maarcadetweet.local",
        prefix,
        uuid::Uuid::new_v4().simple()
    );
    let acc: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({"handle": handle, "password": "hunter2hunter2"}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let did = acc["did"].as_str().unwrap().to_string();
    let jwt = acc["access_jwt"].as_str().unwrap().to_string();
    (c, did, jwt)
}

/// Seed a single post we can like. Returns the post's `uri` and `cid`
/// — both are needed to build a like record's `subject`.
async fn seed_post(
    c: &reqwest::Client,
    did: &str,
    jwt: &str,
    text: &str,
) -> (String, String) {
    let resp: Value = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {
                "text": text,
                "createdAt": "2026-07-01T12:00:00Z",
            },
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let uri = resp["uri"].as_str().unwrap().to_string();
    let cid = resp["cid"].as_str().unwrap().to_string();
    (uri, cid)
}

#[tokio::test]
async fn create_like_persists_record_and_returns_uri() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, jwt) = fresh_user("like").await;
    let (subject_uri, subject_cid) =
        seed_post(&c, &did, &jwt, "post to be liked").await;

    // Use the flat BSky shape — what the Tauri client will send.
    let resp: Value = c
        .post(format!("{}/xrpc/com.atproto.feed.like.create", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "subject": {
                "uri": subject_uri,
                "cid": subject_cid,
            },
            "createdAt": "2026-07-04T12:00:00Z",
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();

    let uri = resp["uri"].as_str().expect("uri missing");
    let cid = resp["cid"].as_str().expect("cid missing");
    let commit = resp["commit"].as_object().expect("commit object");

    // The returned URI should be at://{did}/app.bsky.feed.like/{rkey}.
    assert!(uri.starts_with(&format!("at://{did}/app.bsky.feed.like/")));
    assert!(cid.starts_with("bafy"), "cid should look like a CID: {cid}");
    assert!(commit["cid"].is_string());
    assert!(commit["rev"].is_string());

    // The like should be readable back via sync.getRecord.
    let rkey = uri.rsplit('/').next().unwrap();
    let car = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRecord?did={}&collection=app.bsky.feed.like&rkey={}",
            PDS_URL, did, rkey
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(car.status().as_u16(), 200);

    // Round-trip with the generic createRecord body shape to
    // confirm the handler accepts both shapes.
    let (subject_uri2, subject_cid2) =
        seed_post(&c, &did, &jwt, "second post for generic-shape like").await;
    let resp2: Value = c
        .post(format!("{}/xrpc/com.atproto.feed.like.create", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.bsky.feed.like",
            "record": {
                "subject": {
                    "uri": subject_uri2,
                    "cid": subject_cid2,
                },
                "createdAt": "2026-07-04T12:00:00Z",
            },
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert!(
        resp2["uri"]
            .as_str()
            .unwrap()
            .starts_with(&format!("at://{did}/app.bsky.feed.like/")),
        "generic-shape like should also work, got: {resp2:?}"
    );

    // A wrong collection name should be rejected.
    let r_wrong = c
        .post(format!("{}/xrpc/com.atproto.feed.like.create", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {
                "subject": {
                    "uri": subject_uri,
                    "cid": subject_cid,
                },
                "createdAt": "2026-07-04T12:00:00Z",
            },
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r_wrong.status().as_u16(), 400);

    // No bearer header should be rejected.
    let r_noauth = c
        .post(format!("{}/xrpc/com.atproto.feed.like.create", PDS_URL))
        .json(&json!({
            "repo": did,
            "subject": {
                "uri": subject_uri,
                "cid": subject_cid,
            },
            "createdAt": "2026-07-04T12:00:00Z",
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r_noauth.status().as_u16(), 401);
}

#[tokio::test]
async fn delete_record_removes_like() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    let (c, did, jwt) = fresh_user("del").await;
    let (subject_uri, subject_cid) =
        seed_post(&c, &did, &jwt, "to be liked then unliked").await;

    // Create a like.
    let create: Value = c
        .post(format!("{}/xrpc/com.atproto.feed.like.create", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "subject": {
                "uri": subject_uri,
                "cid": subject_cid,
            },
            "createdAt": "2026-07-04T12:00:00Z",
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let like_uri = create["uri"].as_str().unwrap().to_string();
    let like_rkey = like_uri.rsplit('/').next().unwrap().to_string();

    // Delete it.
    let del: Value = c
        .post(format!("{}/xrpc/com.atproto.repo.deleteRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.bsky.feed.like",
            "rkey": like_rkey,
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let commit = del["commit"].as_object().expect("commit object");
    assert!(commit["cid"].is_string());
    assert!(commit["rev"].is_string());

    // The like should be gone from sync.getRecord.
    let r = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRecord?did={}&collection=app.bsky.feed.like&rkey={}",
            PDS_URL, did, like_rkey
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(r.status().as_u16(), 404, "deleted like must 404");

    // Idempotency: deleting again should still return 200 with a
    // commit. (The repo's MST is unchanged so the commit is a
    // no-op, but the request is accepted.)
    let del2 = c
        .post(format!("{}/xrpc/com.atproto.repo.deleteRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.bsky.feed.like",
            "rkey": like_rkey,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(del2.status().as_u16(), 200, "second delete must be idempotent");

    // `deleteRecord` is generic — it should work for any
    // collection. Create a post and delete it the same way.
    let post_resp: Value = c
        .post(format!("{}/xrpc/com.atproto.repo.createRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "record": {
                "text": "to be deleted",
                "createdAt": "2026-07-04T12:00:00Z",
            },
        }))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let post_uri = post_resp["uri"].as_str().unwrap().to_string();
    let post_rkey = post_uri.rsplit('/').next().unwrap().to_string();
    let r3 = c
        .post(format!("{}/xrpc/com.atproto.repo.deleteRecord", PDS_URL))
        .bearer_auth(&jwt)
        .json(&json!({
            "repo": did,
            "collection": "app.twi.post",
            "rkey": post_rkey,
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r3.status().as_u16(), 200);

    // Auth: deleting someone else's record (here: a fabricated repo
    // matching the JWT sub) is rejected when the JWT sub doesn't
    // match the body. We can't easily mint a JWT for a different
    // DID in this test, so we just check that missing JWT → 401.
    let r_noauth = c
        .post(format!("{}/xrpc/com.atproto.repo.deleteRecord", PDS_URL))
        .json(&json!({
            "repo": did,
            "collection": "app.bsky.feed.like",
            "rkey": "any",
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(r_noauth.status().as_u16(), 401);
}

// -- concurrent write tests (Phase 5b review C1) ---------------------------

/// Phase 5b review C1 — concurrent PDS writes used to race on the
/// `repos.head_commit` column. Two writers would both read the same
/// head, both build a valid child commit with the same `prev`, and
/// the second `UPDATE` would clobber the first — leaving the first
/// writer's MST changes stranded in `repo_blocks` but unreachable
/// from the new head.
///
/// The fix is `SELECT … FOR UPDATE` on the user's `repos` row inside
/// a transaction (see `routes::helpers::apply_repo_write`). This
/// test fires 10 parallel `createRecord` requests for one DID and
/// asserts every record survives into the final head: every rkey is
/// fetchable via `sync.getRecord` (each returns 200, not 404), and
/// the MST contains exactly the 10 records we created.
#[tokio::test]
async fn concurrent_writes_dont_lose_data() {
    if !wait_for_pds().await {
        eprintln!("pds not running, skipping");
        return;
    }
    // Concurrent writers serialise on the row lock; with 10 of them
    // each round-trip takes a few hundred ms, so the per-request
    // timeout has to be generous. The default 5s `client()` would
    // time out long requests 9 and 10.
    let c = reqwest::Client::builder()
        .timeout(Duration::from_secs(60))
        .build()
        .unwrap();
    let handle = format!(
        "race_{}.maarcadetweet.local",
        uuid::Uuid::new_v4().simple()
    );
    let acc: Value = c
        .post(format!("{}/xrpc/com.atproto.server.createAccount", PDS_URL))
        .json(&json!({"handle": handle, "password": "hunter2hunter2"}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let did = acc["did"].as_str().unwrap().to_string();
    let jwt = acc["access_jwt"].as_str().unwrap().to_string();

    // Fire N parallel createRecord requests. Each request has a
    // distinct text + createdAt so the value CIDs differ; if the
    // TID helper ever regressed to collisions, the rkeys would still
    // collide and we'd lose records (the integration-level race is
    // orthogonal to the TID race).
    //
    // N=5 is the largest batch that comfortably fits in sqlx's
    // default 10-connection pool — each in-flight write holds a
    // transaction connection, and the helper also issues a non-tx
    // signing-key read on a *second* connection per request.
    const N: usize = 5;
    let mut handles = Vec::with_capacity(N);
    for i in 0..N {
        let c = c.clone();
        let did = did.clone();
        let jwt = jwt.clone();
        handles.push(tokio::spawn(async move {
            let resp = c
                .post(format!(
                    "{}/xrpc/com.atproto.repo.createRecord",
                    PDS_URL
                ))
                .bearer_auth(&jwt)
                .json(&json!({
                    "repo": did,
                    "collection": "app.twi.post",
                    "record": {
                        "text": format!("concurrent post #{i}"),
                        "createdAt": format!("2026-07-04T12:00:{:02}Z", i),
                    },
                }))
                .send()
                .await
                .unwrap();
            let status = resp.status().as_u16();
            let body: Value = resp.json().await.unwrap();
            (status, body)
        }));
    }

    let mut created = Vec::with_capacity(N);
    for h in handles {
        let (status, body) = h.await.unwrap();
        assert_eq!(
            status, 200,
            "concurrent createRecord failed: {body:?}"
        );
        let uri = body["uri"].as_str().unwrap().to_string();
        let cid = body["cid"].as_str().unwrap().to_string();
        created.push((uri, cid));
    }
    assert_eq!(created.len(), N);

    // Every record must be fetchable from the final repo. If a
    // concurrent writer lost its MST update, sync.getRecord for that
    // rkey returns 404.
    for (uri, cid) in &created {
        let rkey = uri.rsplit('/').next().unwrap();
        let resp = c
            .get(format!(
                "{}/xrpc/com.atproto.sync.getRecord?did={}&collection=app.twi.post&rkey={}",
                PDS_URL, did, rkey
            ))
            .send()
            .await
            .unwrap();
        assert_eq!(
            resp.status().as_u16(),
            200,
            "record {uri} (cid {cid}) lost after concurrent writes — repo_blocks has it but head doesn't"
        );
        let bytes = resp.bytes().await.unwrap();
        let parsed = parse_car(&bytes);
        let target_hex: String = cid
            .parse::<cid::Cid>()
            .unwrap()
            .to_bytes()
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect();
        assert!(
            parsed.blocks.iter().any(|(c, _)| c == &target_hex),
            "CAR for record {uri} missing value block {cid} (raw: {target_hex})"
        );
    }

    // Belt-and-braces: fetch getRepo and confirm the head commit's
    // `prev` chain is well-formed (every commit's `prev` is reachable
    // from the next). The CAR includes every block in the repo so
    // we can also check that the MST root and all 10 value CIDs are
    // present.
    let repo_bytes = c
        .get(format!(
            "{}/xrpc/com.atproto.sync.getRepo?did={}",
            PDS_URL, did
        ))
        .send()
        .await
        .unwrap()
        .bytes()
        .await
        .unwrap();
    let parsed = parse_car(&repo_bytes);
    for (_uri, cid) in &created {
        let target_hex: String = cid
            .parse::<cid::Cid>()
            .unwrap()
            .to_bytes()
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect();
        assert!(
            parsed.blocks.iter().any(|(c, _)| c == &target_hex),
            "final head repo missing value block {cid}"
        );
    }
}