tomdebone
c586fd39c9
AT Protocol PDS + AppView + Tauri Desktop Client, 160-char post limit. - PDS (Rust + axum + sqlx) - Auth: createAccount, createSession, refreshSession - Records: createRecord, deleteRecord (race-safe via SELECT FOR UPDATE) - Feed: feed.like.create, feed.repost.create - Sync: getRepo, getBlocks, getLatestCommit, getRecord (with MST proof), listRepos - Identity: resolveHandle - MST: spec-conformant (at-mst crate, 27 tests) - Repo: signed commits, TID counter (monotonic, 4096 wrap safe) - AppView (Rust + axum + sqlx) - Jetstream consumer (WebSocket, exponential backoff, 38k+ events indexed) - REST API: timeline/home (graph-aware), profile, search, post (with thread hydration) - Handle-sync worker (did:plc + did:web) - JSONB embed storage + thread columns (migration 0003) - Like/repost counter cache (migration 0004) - Tauri 2 + Svelte 5 Desktop Client - System tray (Show/Compose/Quit menu) - OS notifications (tauri-plugin-notification) - Auto-update (tauri-plugin-updater, placeholder endpoint) - Window-state (tauri-plugin-window-state) - 160-char compose with live counter - Image/Link embed rendering - LocalStorage-persisted like state - Timeline with poll (prepend new posts) - Custom TitleBar (transparent, no decorations) - Orange/IBM Plex Mono maarcade design Tests: 231 Rust + 9 vitest = 240 passed.
1715 lines
54 KiB
Rust
1715 lines
54 KiB
Rust
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}"
|
|
);
|
|
}
|
|
}
|