give an account its handle in the first generation that names it
Signed-off-by: dawn <90008@klbr.net>
864 unmodified linesExpand all86586686786886986987087187287387487587687766 unmodified lines864 unmodified linesExpand all ), "{second_html}" );
assert!(!env.output.join("owner-handles.json").exists()); let tried: serde_json::Value = serde_json::from_slice(&std::fs::read(env.output.join("owner-handles.json")).unwrap()) .unwrap(); assert!(tried["entries"].get(OWNER).is_none(), "{tried}"); assert!(tried["unverified"].get(OWNER).is_some(), "{tried}");
std::fs::write( env.output.join("owner-handles.json"), serde_json::to_vec_pretty(&serde_json::json!({66 unmodified lines556 unmodified linesExpand all557558559560561562563564565169 unmodified linesExpand all556 unmodified linesExpand all .join() .map_err(|_| anyhow::anyhow!("atproto settle panicked"))? })? }; crate::index_page::verify_unseen(&config, &crate::xrpc::social::accounts(&atproto_repos)).await; crate::atproto::emit_with( &build_dir, &atproto_repos, &crate::xrpc::social::Social::of(&config),169 unmodified linesExpand all32 unmodified lines333435363738394041424372 unmodified lines1161171181191201211221191201211221231241251251261271281291291301311321337 unmodified lines14114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024136 unmodified lines2782792802811891901911921931941951961971981992002012822832842852862872882892902912922932942952052962972982993003019 unmodified lines311312313314223224225315316317318319320321322152 unmodified linesExpand all32 unmodified lines#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]pub(crate) struct StoredHandles { #[serde(default)] pub entries: BTreeMap<String, StoredHandle>, /// accounts whose first verification failed, and when. generate leaves them to post-activate #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub unverified: BTreeMap<String, u64>,}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]pub(crate) struct StoredHandle {72 unmodified lines Ok(found) => owners.extend(found.iter().map(|did| did.as_str().to_string())), Err(error) => { tracing::warn!(%error, "read collaborators for the handle cache"); authors.extend(stored.entries.keys().cloned()); authors.extend(stored.unverified.keys().cloned()); } } let limits = refresh_limits(config); let Ok(transport) = ReqwestHttp::new(limits) else { return; }; let Ok(pds) = ReqwestHttp::new(HttpLimits { max_response_bytes: AVATAR_MAX_BYTES, ..limits ..refresh_limits(config) }) else { return; }; let Ok(atproto) = config.atproto(transport) else { let Some(atproto) = refresh_atproto(config) else { return; }; let before = stored.clone(); let avatars = config.output_root.join(AVATARS_DIR);7 unmodified lines sweep_avatars(&avatars, &stored); }}
fn refresh_atproto( config: &ValidatedConfig,) -> Option<Atproto<ReqwestHttp, knot_runtime::SystemClock>> { config .atproto(ReqwestHttp::new(refresh_limits(config)).ok()?) .ok()}
/// a generation verifies the accounts it names for the first time itself, so their first/// answers already carry the handle. an account seen before costs nothing herepub(crate) async fn verify_unseen(config: &ValidatedConfig, accounts: &BTreeSet<String>) { let store = Datastore::at(&config.output_root); let record = store.record::<schema::OwnerHandles>(&()); let mut stored = record.read().ok().flatten().unwrap_or_default(); let unseen = unseen(&stored, accounts); if unseen.is_empty() { return; } let Some(atproto) = refresh_atproto(config) else { return; }; verify_unseen_with(&atproto, accounts, &mut stored, crate::publish::unix_now()).await; let mut latest = record.read().ok().flatten().unwrap_or_default(); for did in unseen { if let Some(entry) = stored.entries.get(did) && !latest.entries.contains_key(did) { latest.unverified.remove(did); latest.entries.insert(did.to_string(), entry.clone()); } else if let Some(&at) = stored.unverified.get(did) { latest.unverified.entry(did.to_string()).or_insert(at); } } if let Err(error) = record.put(&latest) { tracing::warn!(%error, "keep the handles of new accounts"); }}
fn unseen<'a>(stored: &StoredHandles, accounts: &'a BTreeSet<String>) -> Vec<&'a str> { accounts .iter() .map(String::as_str) .filter(|did| !stored.entries.contains_key(*did) && !stored.unverified.contains_key(*did)) .collect()}
/// a handle found here has no avatar yet and counts as never verified, so the next/// post-activate verifies it again and fetches the avatarpub(crate) async fn verify_unseen_with<H: HttpTransport, C: Clock>( atproto: &Atproto<H, C>, accounts: &BTreeSet<String>, stored: &mut StoredHandles, now_unix: u64,) { let unseen = unseen(stored, accounts); let mut found = verified(atproto, unseen.iter().copied()).await; for did in unseen { match found.remove(did) { Some((handle, _)) => { stored.entries.insert( did.to_string(), StoredHandle { handle, verified_at_unix: 0, avatar_cid: None, avatar_ext: None, }, ); } None => { stored.unverified.insert(did.to_string(), now_unix); } } }}
async fn verified<'a, H: HttpTransport, C: Clock>( atproto: &Atproto<H, C>, dids: impl IntoIterator<Item = &'a str>,) -> BTreeMap<&'a str, (String, Url)> { futures::stream::iter(dids.into_iter().map(|did| async move { let found = tokio::time::timeout(REFRESH_TIMEOUT, verify_owner_handle(atproto, did)) .await .ok() .flatten(); (did, found) })) .buffer_unordered(REFRESHES_IN_FLIGHT) .filter_map(|(did, found)| async move { Some((did, found?)) }) .collect() .await}
/// generations link their own copies, so removing one here is safe.fn sweep_avatars(avatars: &Path, stored: &StoredHandles) { let kept: BTreeSet<String> = stored .entries36 unmodified lines .iter() .copied() .chain(named.authors.iter().copied().filter(stale)) .collect(); let refreshed: Vec<_> = futures::stream::iter(due.into_iter().map(|raw_did| async move { let (handle, endpoint) = tokio::time::timeout(REFRESH_TIMEOUT, verify_owner_handle(atproto, raw_did)) .await .ok() .flatten()?; let avatar = tokio::time::timeout(AVATAR_TIMEOUT, avatar_of(pds, avatars, raw_did, &endpoint)) .await .ok() .flatten(); Some((raw_did, handle, avatar)) })) let found = verified(atproto, due).await; let refreshed: Vec<_> = futures::stream::iter(found.into_iter().map( |(raw_did, (handle, endpoint))| async move { let avatar = tokio::time::timeout(AVATAR_TIMEOUT, avatar_of(pds, avatars, raw_did, &endpoint)) .await .ok() .flatten(); (raw_did, handle, avatar) }, )) .buffer_unordered(REFRESHES_IN_FLIGHT) .collect() .await; for (raw_did, handle, avatar) in refreshed.into_iter().flatten() { for (raw_did, handle, avatar) in refreshed { stored.unverified.remove(raw_did); let previous = stored.entries.get(raw_did); let (avatar_cid, avatar_ext) = match avatar { Some(found) => found.unzip(), None => previous9 unmodified lines avatar_ext, }, ); } stored.entries.retain(|did, _| { named.owners.contains(did.as_str()) || named.authors.contains(did.as_str()) }); let is_named = |did: &String| named.owners.contains(did.as_str()) || named.authors.contains(did.as_str()); stored.entries.retain(|did, _| is_named(did)); stored.unverified.retain(|did, _| is_named(did));}
async fn verify_owner_handle<H: HttpTransport, C: Clock>( atproto: &Atproto<H, C>,152 unmodified linesExpand all27 unmodified lines28293031323233343536656 unmodified linesExpand all27 unmodified lines#[cfg(test)]mod tests;
pub use handles::sync_handles;pub(crate) use handles::{StoredHandles, cached_owner_labels};pub(crate) use handles::{StoredHandles, cached_owner_labels, verify_unseen};use handles::{read_activity, read_descriptions};
pub const KAOMOJI: &[&str] = &[ "(˶ᵔ ᵕ ᵔ˶)",656 unmodified linesExpand all1234567402 unmodified linesExpand all41041141241341441541641741824 unmodified lines44344444544644744844945045145245345445545645745815 unmodified lines474475476477478479480481482483484101 unmodified linesExpand all586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639358 unmodified linesExpand alluse super::handles::{ AUTHOR_REVERIFY_SECS, Named, avatar_of, published_description, refresh_handles_with, verify_unseen_with,};use super::*;use crate::config::IndexConfig;use crate::datastore::{Datastore, schema};402 unmodified linesExpand all } match name.as_str() { "_atproto.ptr.pet" => Ok(vec!["did=did:plc:verified".to_string()]), "_atproto.impostor.pet" => Ok(vec!["did=did:plc:other".to_string()]), "_atproto.fresh.test" => Ok(vec!["did=did:plc:fresh".to_string()]), _ => Ok(Vec::new()), } }) }24 unmodified lines
let offline = Arc::new(AtomicBool::new(false)); let net_down = Arc::clone(&offline); let author_lookups = Arc::new(AtomicUsize::new(0)); let requests = Arc::new(AtomicUsize::new(0)); let make_atproto = || { let net_down = Arc::clone(&net_down); let author_lookups = Arc::clone(&author_lookups); let requests = Arc::clone(&requests); let doc = doc.clone(); let http = FakeHttp::new(move |req| { requests.fetch_add(1, Ordering::SeqCst); if net_down.load(Ordering::SeqCst) { return Err(NetworkError::Request("plc down".into())); } let reply = |status: StatusCode, body: Bytes| HttpResponse {15 unmodified lines "https://plc.directory/did:plc:author" => { author_lookups.fetch_add(1, Ordering::SeqCst); Ok(reply(StatusCode::OK, doc("did:plc:author", "author.test"))) } "https://plc.directory/did:plc:fresh" => { Ok(reply(StatusCode::OK, doc("did:plc:fresh", "fresh.test"))) } "https://author.test/.well-known/atproto-did" => Ok(reply( StatusCode::OK, Bytes::from_static(b"did:plc:author\n"), )),101 unmodified linesExpand all ) .await; assert!(!stored.entries.contains_key("did:plc:author")); assert_eq!(stored.entries["did:plc:verified"].handle, "ptr.pet");
let seen = BTreeSet::from(["did:plc:verified", "did:plc:fresh", "did:plc:missing"].map(String::from)); let before = requests.load(Ordering::SeqCst); verify_unseen_with(&make_atproto(), &seen, &mut stored, later).await; assert!(requests.load(Ordering::SeqCst) > before); let fresh = &stored.entries["did:plc:fresh"]; assert_eq!(fresh.handle, "fresh.test"); assert_eq!( fresh.verified_at_unix, 0, "post-activate verifies it again with its avatar" ); assert_eq!(stored.unverified["did:plc:missing"], later); assert_eq!(stored.entries["did:plc:verified"].verified_at_unix, later);
let before = requests.load(Ordering::SeqCst); verify_unseen_with(&make_atproto(), &seen, &mut stored, later + 1).await; assert_eq!( requests.load(Ordering::SeqCst), before, "accounts seen before cost a generate nothing" );
let named = Named { owners: owners.clone(), authors: BTreeSet::from(["did:plc:fresh", "did:plc:missing"]), }; refresh_handles_with(&make_atproto(), &pds, &avatars, &named, &mut stored, later).await; let fresh = &stored.entries["did:plc:fresh"]; assert_eq!(fresh.verified_at_unix, later); assert_eq!(fresh.avatar_cid.as_deref(), Some(cid.as_str())); assert!(stored.unverified.contains_key("did:plc:missing")); let only_ptr = Named { owners: BTreeSet::from(["did:plc:verified"]), authors: BTreeSet::new(), }; refresh_handles_with( &make_atproto(), &pds, &avatars, &only_ptr, &mut stored, later, ) .await; assert!(stored.unverified.is_empty());}
const PNG: &[u8] = b"\x89PNG\r\n\x1a\n an avatar";
358 unmodified linesExpand all484 unmodified linesExpand all48548648748848948949049149249349449549649749849149950050150249550350450550650749750850951051151231 unmodified lines484 unmodified linesExpand all atomic_write(&path, maps.to_string().as_bytes()) .with_context(|| format!("write {}", path.display()))}
fn write_accounts(generation_dir: &Path, hosted: &[Hosted]) -> anyhow::Result<()> {pub(crate) fn accounts(repos: &[SettledRepo]) -> BTreeSet<String> { let hosted: Vec<Hosted> = repos.iter().map(Hosted::of).collect(); named_accounts(&hosted) .into_iter() .map(str::to_string) .collect()}
fn named_accounts<'a>(hosted: &[Hosted<'a>]) -> BTreeSet<&'a str> { let repos: BTreeSet<&str> = hosted.iter().map(|repo| repo.did).collect(); let accounts: BTreeSet<&str> = hosted hosted .iter() .flat_map(Hosted::accounts) .filter(|did| !repos.contains(did) && knot_types::AccountDid::new(*did).is_ok()) .collect(); .collect()}
fn write_accounts(generation_dir: &Path, hosted: &[Hosted]) -> anyhow::Result<()> { let path = generation_dir.join(ACCOUNTS_FILE); atomic_write(&path, &serde_json::to_vec(&accounts)?) atomic_write(&path, &serde_json::to_vec(&named_accounts(hosted))?) .with_context(|| format!("write {}", path.display()))}
fn link_avatars(generation_dir: &Path, social: &Social) -> Vec<(String, String)> {31 unmodified lines