//! Shared test infrastructure: unique on-disk roots, REAL canonical content //! (manifests built with `makepad-asset-data`, hashed with real SHA-256), //! and REAL TCP fixture servers speaking the wire protocol over sockets. //! There are no mocks: the client under test dials actual endpoints and every //! digest it verifies is a genuine hash of genuine bytes. #![allow(dead_code)] use makepad_asset_client::json::{self, obj, s, Value}; use makepad_asset_client::wire; use makepad_asset_data::*; use std::collections::HashMap; use std::io::{Read, Write}; use std::net::{SocketAddr, TcpListener, TcpStream, UdpSocket}; use std::path::PathBuf; use std::str::FromStr; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; static DIR_COUNTER: AtomicU64 = AtomicU64::new(0); /// A fresh unique root per call: pid + counter + name, under the OS temp dir. pub fn test_root(name: &str) -> PathBuf { let n = DIR_COUNTER.fetch_add(1, Ordering::Relaxed); std::env::temp_dir().join(format!( "mp_asset_client_test_{}_{}_{}", std::process::id(), n, name )) } pub fn hex(bytes: &[u8]) -> String { let mut out = String::new(); for b in bytes { use std::fmt::Write; let _ = write!(out, "{b:02x}"); } out } // --------------------------------------------------------------------------- // real canonical content // --------------------------------------------------------------------------- /// Deterministic pseudo-random payload bytes. pub fn payload(seed: u64, len: usize) -> Vec { let mut out = Vec::with_capacity(len); let mut x = seed.wrapping_add(0x9e37_79b9_7f4a_7c15); while out.len() < len { x = x.wrapping_mul(0x2545_f491_4f6c_dd1d).wrapping_add(0x1234_5678); out.extend_from_slice(&x.to_le_bytes()); } out.truncate(len); out } /// Minimal valid mesh-bearing manifest: one render GLB plus the mandatory /// thumbnail, both referencing the digests of the supplied bytes. pub fn prop_manifest( asset_id: AssetId, glb: &[u8], thumb: &[u8], dependencies: Vec, ) -> AssetManifest { AssetManifest { asset_id, kind: AssetKind::Prop, files: vec![AssetFile { role: FileRole::RenderGlb, tier: DeviceTier::Any, lod: 0, media: MediaType::Glb, blob: BlobId::hash_of(glb), byte_len: glb.len() as u64, dims: None, }], dependencies, thumbnail: Some(ThumbnailMeta { blob: BlobId::hash_of(thumb), media: ThumbnailMedia::Png, width: 512, height: 512, byte_len: thumb.len() as u64, views: Vec::new(), }), metrics: Metrics { total_bytes: glb.len() as u64 + thumb.len() as u64, triangles: 12, vertices: 8, joints: 0, clips: 0, max_texture_dim: 0, media_millis: 0, }, coordinate_system: CoordinateSystem { units_per_meter: 1.0, up: Axis::YPos, forward: Axis::ZNeg, pivot: Pivot::Origin, }, bounds: Bounds { min: Vec3::new(-1.0, -1.0, -1.0), max: Vec3::new(1.0, 1.0, 1.0) }, anchors: vec![], capabilities: Capabilities { rigged: false, animated: false, collidable: false, loopable: false, spawnable: false, }, spawn_recipe: None, provenance: None, rights: Rights { license: "CC0-1.0".into(), license_revision: String::new(), terms_digest: None, terms_url: String::new(), credits: "fixture".into(), source: String::new(), source_archive: None, redistribution: Redistribution::Allowed, derivatives: DerivativePolicy::Allowed, }, } } /// One published asset inside the fixture store. pub struct FixtureAsset { pub asset_id: AssetId, pub namespace: String, pub alias: Option, pub title: String, pub revision: AssetRevisionId, pub manifest: AssetManifest, pub created_ms: u64, /// Search labels, mirroring the server's `search_labels` tag rows. pub tags: Vec, } /// In-memory content store served by [`FixtureServer`]. All identities are /// real digests of real canonical bytes. #[derive(Default)] pub struct FixtureStore { pub blobs: HashMap<[u8; 32], Vec>, /// arev digest → canonical manifest bytes. pub manifests: HashMap<[u8; 32], Vec>, /// grev digest → canonical game manifest bytes. pub game_manifests: HashMap<[u8; 32], Vec>, pub assets: Vec, /// alias string → (asset, head revision). pub aliases: HashMap, } impl FixtureStore { pub fn add_blob(&mut self, bytes: Vec) -> BlobId { let id = BlobId::hash_of(&bytes); self.blobs.insert(*id.as_bytes(), bytes); id } /// Publish a prop built from real bytes; returns the exact ref. pub fn add_prop( &mut self, id_byte: u8, namespace: &str, alias: Option<&str>, title: &str, glb: Vec, deps: Vec, ) -> AssetRevisionRef { let thumb = payload(9000 + id_byte as u64, 900); self.add_blob(glb.clone()); self.add_blob(thumb.clone()); let asset_id = AssetId::from_bytes([id_byte; 16]); let manifest = prop_manifest(asset_id, &glb, &thumb, deps); manifest.validate().expect("fixture manifest valid"); let bytes = manifest.to_canonical_bytes().expect("fixture canonical bytes"); let revision = manifest.revision().expect("fixture revision"); self.manifests.insert(*revision.as_bytes(), bytes); if let Some(a) = alias { self.aliases.insert(a.to_string(), (asset_id, revision)); } self.assets.push(FixtureAsset { asset_id, namespace: namespace.to_string(), alias: alias.map(str::to_string), title: title.to_string(), revision, manifest, created_ms: 1_700_000_000_000 + id_byte as u64, tags: Vec::new(), }); // Keyset pages compare display strings, so keep the canonical order // in the same key (base32 byte order differs from ASCII order). self.assets.sort_by_key(|a| a.asset_id.to_string()); AssetRevisionRef { asset_id, revision } } /// Attach search tags to an already-published fixture asset. pub fn tag_asset(&mut self, asset: &AssetRevisionRef, tags: &[&str]) { let row = self .assets .iter_mut() .find(|a| a.asset_id == asset.asset_id) .expect("fixture asset to tag"); row.tags = tags.iter().map(|t| (*t).to_string()).collect(); } /// Publish a game revision whose splash/manifest/lock blobs are real. pub fn add_game(&mut self, id_byte: u8, name: &str) -> GameRevisionId { let splash = payload(50_000 + id_byte as u64, 2_000); let card = payload(60_000 + id_byte as u64, 300); let lock = payload(70_000 + id_byte as u64, 400); let thumb = payload(80_000 + id_byte as u64, 800); let manifest = GameRevisionManifest { game_id: GameId::from_bytes([id_byte; 16]), name: name.into(), description: "fixture game".into(), author: "fixture".into(), splash_blob: self.add_blob(splash.clone()), manifest_blob: self.add_blob(card), lock_blob: self.add_blob(lock), thumbnail: ThumbnailMeta { blob: self.add_blob(thumb.clone()), media: ThumbnailMedia::Png, width: 512, height: 512, byte_len: thumb.len() as u64, views: Vec::new(), }, catalog_snapshot: None, search_algorithm_version: 1, engine_version: 1, protocol_version: 1, splash_byte_len: splash.len() as u64, }; manifest.validate().expect("fixture game valid"); let bytes = manifest.to_canonical_bytes().expect("fixture game bytes"); let revision = manifest.revision().expect("fixture game revision"); self.game_manifests.insert(*revision.as_bytes(), bytes); revision } } // --------------------------------------------------------------------------- // raw TCP server (hostile tests hand-craft the bytes) // --------------------------------------------------------------------------- /// A parsed incoming request, enough for routing and assertions. #[derive(Clone, Debug)] pub struct ParsedRequest { pub method: String, pub target: String, /// Lowercased names. pub headers: Vec<(String, String)>, pub body: Vec, } impl ParsedRequest { pub fn header(&self, name: &str) -> Option<&str> { self.headers.iter().find(|(k, _)| k == name).map(|(_, v)| v.as_str()) } /// Path segments of the target (no query). pub fn segs(&self) -> Vec { let path = self.target.split('?').next().unwrap_or(""); path.trim_start_matches('/').split('/').map(str::to_string).collect() } pub fn query_get(&self, key: &str) -> Option { let q = self.target.split_once('?')?.1; for pair in q.split('&') { let (k, v) = pair.split_once('=')?; if k == key { return Some(v.to_string()); } } None } } pub type RawHandler = dyn Fn(ParsedRequest, &mut TcpStream) + Send + Sync; /// Real listening TCP server; each connection serves one request through the /// handler and closes (matching the client's `Connection: close` model). pub struct RawServer { pub addr: SocketAddr, stop: Arc, join: Option>, } impl RawServer { pub fn start(handler: Arc) -> RawServer { let listener = TcpListener::bind("127.0.0.1:0").expect("bind fixture"); let addr = listener.local_addr().unwrap(); let stop = Arc::new(AtomicBool::new(false)); let stop_t = stop.clone(); let join = std::thread::spawn(move || { for conn in listener.incoming() { if stop_t.load(Ordering::Relaxed) { break; } let Ok(mut stream) = conn else { continue }; let handler = handler.clone(); std::thread::spawn(move || { if let Some(req) = read_request(&mut stream) { handler(req, &mut stream); } }); } }); RawServer { addr, stop, join: Some(join) } } pub fn stop(&mut self) { self.stop.store(true, Ordering::Relaxed); // Unblock accept. let _ = TcpStream::connect(self.addr); if let Some(join) = self.join.take() { let _ = join.join(); } } } impl Drop for RawServer { fn drop(&mut self) { self.stop(); } } /// Minimal exact request reader for the fixture side. pub fn read_request(stream: &mut TcpStream) -> Option { let mut buf = Vec::new(); let mut chunk = [0u8; 4096]; let head_end = loop { if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") { break i; } if buf.len() > 128 * 1024 { return None; } let n = stream.read(&mut chunk).ok()?; if n == 0 { return None; } buf.extend_from_slice(&chunk[..n]); }; let head = String::from_utf8(buf[..head_end].to_vec()).ok()?; let mut lines = head.split("\r\n"); let request_line = lines.next()?; let mut parts = request_line.split(' '); let method = parts.next()?.to_string(); let target = parts.next()?.to_string(); let mut headers = Vec::new(); for line in lines { let (k, v) = line.split_once(':')?; headers.push((k.trim().to_ascii_lowercase(), v.trim().to_string())); } let mut body = buf.split_off(head_end + 4); let content_length: usize = headers .iter() .find(|(k, _)| k == "content-length") .and_then(|(_, v)| v.parse().ok()) .unwrap_or(0); while body.len() < content_length { let n = stream.read(&mut chunk).ok()?; if n == 0 { return None; } body.extend_from_slice(&chunk[..n]); } body.truncate(content_length); Some(ParsedRequest { method, target, headers, body }) } // ---- response writers ------------------------------------------------------ pub fn write_raw(stream: &mut TcpStream, bytes: &[u8]) { let _ = stream.write_all(bytes); } pub fn response_head(status: u16, content_type: &str, len: u64, extra: &[(&str, &str)]) -> String { let mut out = format!( "HTTP/1.1 {status} X\r\nContent-Type: {content_type}\r\nContent-Length: {len}\r\n" ); for (k, v) in extra { out.push_str(&format!("{k}: {v}\r\n")); } out.push_str("Connection: close\r\n\r\n"); out } pub fn write_json_resp(stream: &mut TcpStream, status: u16, v: &Value) { let body = v.to_json().into_bytes(); let head = response_head(status, "application/json", body.len() as u64, &[]); let _ = stream.write_all(head.as_bytes()); let _ = stream.write_all(&body); } pub fn write_error(stream: &mut TcpStream, status: u16, msg: &str) { write_json_resp(stream, status, &obj(vec![("error", s(msg))])); } pub fn write_bytes_resp( stream: &mut TcpStream, status: u16, content_type: &str, body: &[u8], extra: &[(&str, &str)], ) { let head = response_head(status, content_type, body.len() as u64, extra); let _ = stream.write_all(head.as_bytes()); let _ = stream.write_all(body); } // --------------------------------------------------------------------------- // full protocol fixture server // --------------------------------------------------------------------------- pub struct FixtureOptions { pub server_id: [u8; 16], /// Value reported by /v1/health (tests can lie here). pub health_protocol_version: u64, /// Require `Authorization: Bearer ` on every route. pub auth_token: Option, } impl Default for FixtureOptions { fn default() -> Self { Self { server_id: [0x5a; 16], health_protocol_version: wire::PROTOCOL_VERSION as u64, auth_token: None, } } } #[derive(Default)] pub struct FixtureKnobs { /// Close the connection after this many blob body bytes — consumed once. pub kill_blob_after: Mutex>, /// Serve blob bytes with the first byte flipped (digest mismatch). pub corrupt_blobs: Mutex, /// Answer range requests with a full 200 (ignore Range). pub ignore_range: Mutex, /// Drip blob bodies as `(chunk_bytes, delay_ms)` — deterministic slow /// transfers for cancellation tests. pub drip_blob: Mutex>, } /// Mutable write-route state (uploads/registrations made by the client /// under test). Kept apart from the immutable seeded [`FixtureStore`]. #[derive(Default)] pub struct PublishedStore { pub blobs: HashMap<[u8; 32], Vec>, /// arev digest → canonical manifest bytes. pub manifests: HashMap<[u8; 32], Vec>, /// asset display id → (namespace, revisions[(rev display, published)]). pub assets: HashMap)>, /// alias → (asset display id, revision display id). pub aliases: HashMap, /// asset display id → annotation title (proves the annotation landed). pub annotations: HashMap, pub minted: u64, } #[derive(Default)] pub struct FixtureLog { pub requests: Mutex>, } impl FixtureLog { pub fn count(&self, method: &str, target_prefix: &str) -> usize { self.requests .lock() .unwrap() .iter() .filter(|r| r.method == method && r.target.starts_with(target_prefix)) .count() } pub fn last_matching(&self, method: &str, target_prefix: &str) -> Option { self.requests .lock() .unwrap() .iter() .rev() .find(|r| r.method == method && r.target.starts_with(target_prefix)) .cloned() } } /// The full read-API fixture: control plane and data plane on separate real /// TCP ports, sharing one store. JSON routes live ONLY on control, blobs /// ONLY on data — requests to the wrong plane 404, which proves the client /// routes correctly. pub struct FixtureServer { pub control: RawServer, pub data: RawServer, pub store: Arc, pub published: Arc>, pub knobs: Arc, pub log: Arc, pub options: Arc, } impl FixtureServer { pub fn start(store: FixtureStore, options: FixtureOptions) -> FixtureServer { let store = Arc::new(store); let published = Arc::new(Mutex::new(PublishedStore::default())); let knobs = Arc::new(FixtureKnobs::default()); let log = Arc::new(FixtureLog::default()); let options = Arc::new(options); let control = { let (store, published, log, options) = (store.clone(), published.clone(), log.clone(), options.clone()); RawServer::start(Arc::new(move |req, stream| { log.requests.lock().unwrap().push(req.clone()); if !auth_ok(&options, &req) { write_error(stream, 401, "unauthenticated"); return; } control_route(&store, &published, &options, &req, stream); })) }; let data = { let (store, published, knobs, log, options) = ( store.clone(), published.clone(), knobs.clone(), log.clone(), options.clone(), ); RawServer::start(Arc::new(move |req, stream| { log.requests.lock().unwrap().push(req.clone()); if !auth_ok(&options, &req) { write_error(stream, 401, "unauthenticated"); return; } data_route(&store, &published, &knobs, &req, stream); })) }; FixtureServer { control, data, store, published, knobs, log, options } } pub fn endpoints(&self) -> makepad_asset_client::ApiEndpoints { makepad_asset_client::ApiEndpoints { control: self.control.addr, data: self.data.addr } } /// Broadcast one discovery beacon for this fixture to a loopback port. pub fn send_beacon(&self, port: u16, auth_required: bool) { let beacon = makepad_asset_client::Beacon { protocol_version: wire::PROTOCOL_VERSION, server_id: self.options.server_id, control_port: self.control.addr.port(), data_port: self.data.addr.port(), auth_required, tls: false, capability_bits: wire::caps::ALL_V1, }; let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket.send_to(&beacon.encode(), ("127.0.0.1", port)).unwrap(); } } fn auth_ok(options: &FixtureOptions, req: &ParsedRequest) -> bool { match &options.auth_token { None => true, Some(token) => req.header("authorization") == Some(&format!("Bearer {token}")), } } // ---- control plane routes -------------------------------------------------- fn control_route( store: &FixtureStore, published: &Mutex, options: &FixtureOptions, req: &ParsedRequest, stream: &mut TcpStream, ) { let segs = req.segs(); let seg = |i: usize| segs.get(i).map(String::as_str).unwrap_or(""); // ---- write routes (publish flow) ---- match (req.method.as_str(), seg(0), seg(1)) { ("POST", "v1", "assets") if segs.len() == 2 => { let Ok(body) = json::parse(&req.body) else { write_error(stream, 400, "malformed json"); return; }; let ns = body.get("namespace").and_then(Value::as_str).unwrap_or("").to_string(); let mut p = published.lock().unwrap(); let asset_id = match body.get("asset_id").and_then(Value::as_str) { Some(id) => { if p.assets.contains_key(id) { write_error(stream, 409, "conflict"); return; } id.to_string() } None => { p.minted += 1; let mut bytes = [0xB0u8; 16]; bytes[8..].copy_from_slice(&p.minted.to_be_bytes()); AssetId::from_bytes(bytes).to_string() } }; p.assets.insert(asset_id.clone(), (ns.clone(), Vec::new())); write_json_resp( stream, 201, &obj(vec![("asset_id", s(asset_id)), ("namespace", s(ns))]), ); return; } ("POST", "v1", "assets") if segs.len() == 4 && seg(3) == "revisions" => { let asset = seg(2).to_string(); let mut p = published.lock().unwrap(); if !p.assets.contains_key(&asset) { write_error(stream, 404, "not found"); return; } // The revision IS the digest of the canonical bytes. let rev = AssetRevisionId::from_bytes(*BlobId::hash_of(&req.body).as_bytes()); p.manifests.insert(*rev.as_bytes(), req.body.clone()); p.assets.get_mut(&asset).unwrap().1.push((rev.to_string(), false)); write_json_resp( stream, 201, &obj(vec![ ("asset_id", s(asset)), ("revision", s(rev.to_string())), ("state", s("staged")), ]), ); return; } ("POST", "v1", "assets") if segs.len() == 6 && seg(3) == "revisions" && seg(5) == "publish" => { let (asset, rev) = (seg(2).to_string(), seg(4).to_string()); let mut p = published.lock().unwrap(); let Some((_, revs)) = p.assets.get_mut(&asset) else { write_error(stream, 404, "not found"); return; }; let Some(entry) = revs.iter_mut().find(|(r, _)| *r == rev) else { write_error(stream, 404, "not found"); return; }; entry.1 = true; write_json_resp( stream, 200, &obj(vec![ ("asset_id", s(asset)), ("revision", s(rev)), ("state", s("published")), ]), ); return; } ("PUT", "v1", "aliases") if segs.len() >= 4 => { let alias = segs[2..].join("/"); let Ok(body) = json::parse(&req.body) else { write_error(stream, 400, "malformed json"); return; }; let asset = body.get("asset_id").and_then(Value::as_str).unwrap_or("").to_string(); let rev = body.get("revision").and_then(Value::as_str).unwrap_or("").to_string(); published.lock().unwrap().aliases.insert(alias.clone(), (asset.clone(), rev.clone())); write_json_resp( stream, 200, &obj(vec![ ("alias", s(alias)), ("asset_id", s(asset)), ("head_revision", s(rev)), ]), ); return; } ("PUT", "v1", "assets") if segs.len() == 4 && seg(3) == "annotation" => { let asset = seg(2).to_string(); let title = json::parse(&req.body) .ok() .and_then(|b| b.get("title").and_then(Value::as_str).map(str::to_string)) .unwrap_or_default(); published.lock().unwrap().annotations.insert(asset, title); // The real server answers a bodyless 204 here. write_raw(stream, b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n"); return; } _ => {} } match (req.method.as_str(), seg(0), seg(1)) { ("GET", "v1", "health") => { write_json_resp( stream, 200, &obj(vec![ ("server_id", s(hex(&options.server_id))), ("protocol_version", Value::Int(options.health_protocol_version as i64)), ]), ); } ("POST", "v1", "catalog") => catalog_route(store, req, stream), ("GET", "v1", "assets") if segs.len() == 2 => assets_route(store, req, stream), ("GET", "v1", "assets") if segs.len() == 3 => { // Published-store assets first (write-route round-trips), then // the seeded store. let asset = seg(2).to_string(); let candidates = published.lock().unwrap().assets.get(&asset).map(|(ns, revs)| { let cands: Vec = revs .iter() .enumerate() .map(|(i, (rev, is_pub))| { obj(vec![ ("revision", s(rev.clone())), ("state", s(if *is_pub { "published" } else { "staged" })), ("staged_ms", Value::Int(1_000 + i as i64)), ( "published_ms", if *is_pub { Value::Int(2_000 + i as i64) } else { Value::Null }, ), ("quarantined_ms", Value::Null), ]) }) .collect(); (ns.clone(), cands) }); match candidates { Some((ns, cands)) => write_json_resp( stream, 200, &obj(vec![ ("asset_id", s(asset)), ("namespace", s(ns)), ("candidates", Value::Arr(cands)), ]), ), None => asset_detail_route(store, seg(2), stream), } } ("GET", "v1", "aliases") if segs.len() >= 4 => { let alias = segs[2..].join("/"); if let Some((asset_id, rev)) = store.aliases.get(&alias) { write_json_resp( stream, 200, &obj(vec![ ("alias", s(alias)), ("asset_id", s(asset_id.to_string())), ("head_revision", s(rev.to_string())), ]), ); return; } match published.lock().unwrap().aliases.get(&alias) { Some((asset_id, rev)) => write_json_resp( stream, 200, &obj(vec![ ("alias", s(alias)), ("asset_id", s(asset_id.clone())), ("head_revision", s(rev.clone())), ]), ), None => write_error(stream, 404, "not found"), } } ("GET", "v1", "revisions") if segs.len() == 3 => { let seeded = AssetRevisionId::from_str(seg(2)) .ok() .and_then(|rev| store.manifests.get(rev.as_bytes()).cloned()) .or_else(|| { AssetRevisionId::from_str(seg(2)).ok().and_then(|rev| { published.lock().unwrap().manifests.get(rev.as_bytes()).cloned() }) }); match seeded { Some(bytes) => { write_bytes_resp(stream, 200, "application/octet-stream", &bytes, &[]) } None => write_error(stream, 404, "not found"), } } ("GET", "v1", "game-revisions") if segs.len() == 3 => { match GameRevisionId::from_str(seg(2)) .ok() .and_then(|rev| store.game_manifests.get(rev.as_bytes())) { Some(bytes) => { write_bytes_resp(stream, 200, "application/octet-stream", bytes, &[]) } None => write_error(stream, 404, "not found"), } } _ => write_error(stream, 404, "not found"), } } fn catalog_route(store: &FixtureStore, req: &ParsedRequest, stream: &mut TcpStream) { let Ok(body) = json::parse(&req.body) else { write_error(stream, 400, "malformed json"); return; }; let q = body.get("q").and_then(Value::as_str).unwrap_or("").to_lowercase(); let ns = body.get("ns").and_then(Value::as_str); let tag = body.get("tag").and_then(Value::as_str); let exclude_tag = body.get("exclude_tag").and_then(Value::as_str); let limit = body.get("limit").and_then(Value::as_u64).unwrap_or(25) as usize; // Cursor binds to the query shape so a cursor from another query refuses. let fingerprint = format!( "f{}", q.len() + ns.map_or(0, str::len) * 100 + tag.map_or(0, str::len) * 10_000 + exclude_tag.map_or(0, str::len) * 1_000_000 ); let start = match body.get("cursor").and_then(Value::as_str) { None => 0usize, Some(c) => { let Some((idx, fp)) = c.split_once('.') else { write_error(stream, 400, "stale search cursor"); return; }; if fp != fingerprint { write_error(stream, 400, "stale search cursor"); return; } idx.trim_start_matches('i').parse().unwrap_or(0) } }; let matches: Vec<&FixtureAsset> = store .assets .iter() .filter(|a| q.is_empty() || a.title.to_lowercase().contains(&q)) .filter(|a| ns.is_none_or(|n| a.namespace == n)) .filter(|a| tag.is_none_or(|t| a.tags.iter().any(|x| x == t))) // Like the server: exclusion runs after the positive tag, so a row // carrying both drops. `total` is the length of this same list. .filter(|a| exclude_tag.is_none_or(|t| !a.tags.iter().any(|x| x == t))) .collect(); let page: Vec = matches .iter() .skip(start) .take(limit) .map(|a| { obj(vec![ ("asset_id", s(a.asset_id.to_string())), ("namespace", s(a.namespace.clone())), ("kind", s("prop")), ("title", s(a.title.clone())), ("snippet", s(format!("about {}", a.title))), ("score", Value::Int(100)), ("live", Value::Bool(a.alias.is_some())), ]) }) .collect(); let next = start + page.len(); let cursor = if next < matches.len() { s(format!("i{next}.{fingerprint}")) } else { Value::Null }; write_json_resp( stream, 200, &obj(vec![ ("hits", Value::Arr(page)), ("total", Value::Int(matches.len() as i64)), ("cursor", cursor), ]), ); } fn assets_route(store: &FixtureStore, req: &ParsedRequest, stream: &mut TcpStream) { let ns = req.query_get("ns"); let limit: usize = req.query_get("limit").and_then(|l| l.parse().ok()).unwrap_or(50); let after = req.query_get("cursor"); let rows: Vec<&FixtureAsset> = store .assets .iter() .filter(|a| ns.as_deref().is_none_or(|n| a.namespace == n)) .filter(|a| after.as_deref().is_none_or(|c| a.asset_id.to_string().as_str() > c)) .collect(); let page: Vec = rows .iter() .take(limit) .map(|a| { obj(vec![ ("asset_id", s(a.asset_id.to_string())), ("namespace", s(a.namespace.clone())), ("created_ms", Value::Int(a.created_ms as i64)), ]) }) .collect(); let cursor = if rows.len() > limit { s(rows[limit - 1].asset_id.to_string()) } else { Value::Null }; write_json_resp(stream, 200, &obj(vec![("assets", Value::Arr(page)), ("cursor", cursor)])); } fn asset_detail_route(store: &FixtureStore, id: &str, stream: &mut TcpStream) { let Ok(asset_id) = AssetId::from_str(id) else { write_error(stream, 400, "invalid input"); return; }; let Some(a) = store.assets.iter().find(|a| a.asset_id == asset_id) else { write_error(stream, 404, "not found"); return; }; write_json_resp( stream, 200, &obj(vec![ ("asset_id", s(a.asset_id.to_string())), ("namespace", s(a.namespace.clone())), ( "candidates", Value::Arr(vec![obj(vec![ ("revision", s(a.revision.to_string())), ("state", s("published")), ("staged_ms", Value::Int(a.created_ms as i64)), ("published_ms", Value::Int(a.created_ms as i64 + 5)), ("quarantined_ms", Value::Null), ])]), ), ]), ); } // ---- data plane routes ----------------------------------------------------- fn data_route( store: &FixtureStore, published: &Mutex, knobs: &FixtureKnobs, req: &ParsedRequest, stream: &mut TcpStream, ) { let segs = req.segs(); // Content-addressed upload (publish flow). if req.method == "POST" && segs.len() == 2 && segs[0] == "v1" && segs[1] == "blobs" { let blob = BlobId::hash_of(&req.body); let mut p = published.lock().unwrap(); let deduped = p.blobs.insert(*blob.as_bytes(), req.body.clone()).is_some(); write_json_resp( stream, 201, &obj(vec![ ("blob_id", s(blob.to_string())), ("size", Value::Int(req.body.len() as i64)), ("deduped", Value::Bool(deduped)), ]), ); return; } if !(segs.len() == 3 && segs[0] == "v1" && segs[1] == "blobs") { write_error(stream, 404, "not found"); return; } let Ok(blob) = BlobId::from_str(&segs[2]) else { write_error(stream, 400, "invalid input"); return; }; let seeded = store .blobs .get(blob.as_bytes()) .cloned() .or_else(|| published.lock().unwrap().blobs.get(blob.as_bytes()).cloned()); let Some(bytes) = seeded else { if req.method == "HEAD" { // A HEAD response must not carry a body; a framed error body // here would be an over-read on the client side. let head = response_head(404, "application/json", 0, &[]); write_raw(stream, head.as_bytes()); } else { write_error(stream, 404, "not found"); } return; }; let mut bytes = bytes.clone(); if *knobs.corrupt_blobs.lock().unwrap() { if let Some(b) = bytes.first_mut() { *b ^= 0xff; } } let etag = format!("\"{blob}\""); let size = bytes.len() as u64; if req.method == "HEAD" { let head = response_head( 200, "application/octet-stream", size, &[("ETag", etag.as_str()), ("Accept-Ranges", "bytes")], ); write_raw(stream, head.as_bytes()); return; } // Range handling: exactly the `bytes=-` shape the client sends. let range_start = if *knobs.ignore_range.lock().unwrap() { None } else { req.header("range") .and_then(|r| r.strip_prefix("bytes=")) .and_then(|r| r.strip_suffix('-')) .and_then(|r| r.parse::().ok()) // If-Range mismatch → serve the full body. .filter(|_| req.header("if-range").is_none_or(|ir| ir == etag)) }; let (status, body_slice, extra) = match range_start { Some(start) if start >= size => { let cr = format!("bytes */{size}"); let head = response_head(416, "application/json", 0, &[("Content-Range", &cr)]); write_raw(stream, head.as_bytes()); return; } Some(start) => { let cr = format!("bytes {start}-{}/{size}", size - 1); (206, &bytes[start as usize..], vec![("Content-Range".to_string(), cr)]) } None => (200, &bytes[..], vec![]), }; let mut extra_refs: Vec<(&str, &str)> = vec![("ETag", etag.as_str())]; for (k, v) in &extra { extra_refs.push((k.as_str(), v.as_str())); } let head = response_head( status, "application/octet-stream", body_slice.len() as u64, &extra_refs, ); write_raw(stream, head.as_bytes()); let kill_after = knobs.kill_blob_after.lock().unwrap().take(); match kill_after { Some(n) => { let n = (n as usize).min(body_slice.len()); write_raw(stream, &body_slice[..n]); // Dropping the stream closes the connection mid-body. } None => match *knobs.drip_blob.lock().unwrap() { Some((chunk, delay_ms)) => { for piece in body_slice.chunks(chunk.max(1)) { write_raw(stream, piece); let _ = stream.flush(); std::thread::sleep(std::time::Duration::from_millis(delay_ms)); } } None => write_raw(stream, body_slice), }, } }