use knot_runtime::Clock;
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use anyhow::Context as _;
use futures::StreamExt as _;
use jacquard_repo::storage::BlockStore as _;
use knot_atproto::Atproto;
use knot_git::Repo;
use knot_record::blocks::GitBlocks;
use knot_record::chain;
use knot_record::manifest::{MANIFEST_COLLECTION, MANIFEST_RKEY, RepoDescription};
use knot_runtime::{HttpLimits, HttpRequest, HttpTransport, NetworkError, ReqwestHttp};
use knot_types::{AccountDid, BlobCid, RepoDid, Sha256Digest};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use url::Url;
use crate::config::ValidatedConfig;
use crate::datastore::{Datastore, schema};
use crate::manifest::{GenerationManifest, RepoEntry};
use crate::plan::repo_key;
use crate::xrpc::social::{ACCOUNTS_FILE, AVATAR_ORIGINAL_MAX, AVATARS_DIR, Avatar};
use super::avatar_sizes::has_sized;
const REFRESH_TIMEOUT: Duration = Duration::from_secs(3);
const AVATAR_TIMEOUT: Duration = Duration::from_secs(10);
/// a bigger avatar is only kept at its sizes, see `AVATAR_ORIGINAL_MAX`
const AVATAR_FETCH_MAX: u64 = 5_000_000;
/// tangled's appview writes the profile under its older sh.tangled name
const TANGLED_PROFILES: [&str; 2] = ["org.tangled.actor.profile", "sh.tangled.actor.profile"];
const PROFILE_COLLECTIONS: [&str; 3] = [
TANGLED_PROFILES[0],
TANGLED_PROFILES[1],
"app.bsky.actor.profile",
];
/// a pds may send getBlob on to the host that keeps its blobs
const AVATAR_REDIRECTS: usize = 3;
/// owners are verified on every sync, everyone else the social records name this often
pub(crate) const AUTHOR_REVERIFY_SECS: u64 = 6 * 60 * 60;
const REFRESHES_IN_FLIGHT: usize = 16;
#[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 {
pub handle: String,
pub verified_at_unix: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub avatar_cid: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub avatar_ext: Option<String>,
/// the account's org.tangled.actor.profile record, sanitized
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile: Option<Map<String, Value>>,
}
impl StoredHandle {
pub(crate) fn avatar_file(&self) -> Option<String> {
Some(format!(
"{}.{}",
self.avatar_cid.as_deref()?,
self.avatar_ext.as_deref()?
))
}
/// the original, when it was small enough to keep, and the sizes made of it
fn cached_avatar(&self) -> impl Iterator<Item = String> {
let sized = self.avatar_cid.iter().flat_map(|cid| {
crate::xrpc::social::AVATAR_SIZES
.map(|size| crate::xrpc::social::sized_avatar(cid, size))
});
self.avatar_file().into_iter().chain(sized)
}
}
/// what a generation answers from the cache, including which avatar files it finds on disk.
/// verification times and failures don't count
pub(crate) fn served_digest(stored: &StoredHandles, avatars: &Path) -> String {
type Answered<'a> = (
&'a str,
Option<(String, bool)>,
&'a Option<Map<String, Value>>,
);
let served: BTreeMap<&str, Answered> = stored
.entries
.iter()
.map(|(did, entry)| {
let avatar = entry
.avatar_cid
.as_deref()
.zip(entry.avatar_ext.as_deref())
.and_then(|(cid, ext)| Avatar::of(avatars, cid, ext))
.map(|avatar| (avatar.file(), avatar.sized));
(
did.as_str(),
(entry.handle.as_str(), avatar, &entry.profile),
)
})
.collect();
crate::cas::Digest::of_bytes(&serde_json::to_vec(&served).expect("handles serialize")).hex()
}
pub(crate) fn cached_handles(output_root: &Path) -> StoredHandles {
Datastore::at(output_root)
.record::<schema::OwnerHandles>(&())
.read()
.ok()
.flatten()
.unwrap_or_default()
}
pub(crate) fn cached_owner_labels(
output_root: &Path,
owners: &BTreeSet<&str>,
) -> BTreeMap<String, String> {
let stored = cached_handles(output_root);
owners
.iter()
.map(|&did| {
let label = stored
.entries
.get(did)
.map(|entry| entry.handle.clone())
.unwrap_or_else(|| did.to_string());
(did.to_string(), label)
})
.collect()
}
fn refresh_limits(config: &ValidatedConfig) -> HttpLimits {
HttpLimits {
connect_timeout: Duration::from_secs(2),
read_timeout: Duration::from_secs(2),
request_timeout: REFRESH_TIMEOUT,
..config.http_limits()
}
}
/// a change the active generation doesn't answer yet gets a generation of its own
pub async fn sync_handles(config: &Arc<ValidatedConfig>) {
let Some(current) = crate::publish::current_generation(&config.output_root) else {
return;
};
let gen_dir = config.output_root.join("generations").join(current);
let Ok(manifest) = GenerationManifest::load(&gen_dir) else {
return;
};
let mut owners: BTreeSet<String> = manifest
.repos
.iter()
.map(|entry| entry.owner.clone())
.collect();
owners.insert(config.owner.as_str().to_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 mut authors: BTreeSet<String> = std::fs::read(gen_dir.join(ACCOUNTS_FILE))
.ok()
.and_then(|bytes| serde_json::from_slice(&bytes).ok())
.unwrap_or_default();
match crate::admin::keys::accounts(config) {
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 Ok(pds) = ReqwestHttp::new(HttpLimits {
max_response_bytes: AVATAR_FETCH_MAX,
..refresh_limits(config)
}) else {
return;
};
let Some(atproto) = refresh_atproto(config) else {
return;
};
let before = stored.clone();
let avatars = config.output_root.join(AVATARS_DIR);
let now = crate::publish::unix_now();
let named = Named {
owners: owners.iter().map(String::as_str).collect(),
authors: authors.iter().map(String::as_str).collect(),
};
refresh_handles_with(&atproto, &pds, &avatars, &named, &mut stored, now).await;
if stored != before && record.put(&stored).is_ok() {
sweep_avatars(&avatars, &stored);
}
let digest = served_digest(&stored, &avatars);
if manifest.handles_digest.as_deref() == Some(digest.as_str()) {
return;
}
let attempt = FailedPublish {
digest,
generation: manifest.generation_id.clone(),
at_unix: now,
};
let failed_path = config.output_root.join(FAILED_PUBLISH_FILE);
let failed = std::fs::read(&failed_path)
.ok()
.and_then(|bytes| serde_json::from_slice::<FailedPublish>(&bytes).ok());
if failed.is_some_and(|failed| failed.holds_back(&attempt)) {
return;
}
match crate::writes::publish_scoped(config, None).await {
Ok(_) => {
let _ = std::fs::remove_file(&failed_path);
}
Err(error) => {
tracing::warn!("publish the refreshed handles: {error:#}");
if let Ok(bytes) = serde_json::to_vec(&attempt) {
let _ = crate::cas::atomic_write(&failed_path, &bytes);
}
}
}
}
const FAILED_PUBLISH_FILE: &str = "handles-publish.failed.json";
/// a publish that failed for these handles over this generation isn't tried again sooner
const PUBLISH_RETRY_SECS: u64 = 3600;
/// the last handles publish that failed, so the drain timer doesn't rebuild every run
#[derive(Debug, PartialEq, Serialize, Deserialize)]
pub(crate) struct FailedPublish {
pub(crate) digest: String,
pub(crate) generation: String,
pub(crate) at_unix: u64,
}
impl FailedPublish {
pub(crate) fn holds_back(&self, attempt: &FailedPublish) -> bool {
self.digest == attempt.digest
&& self.generation == attempt.generation
&& attempt.at_unix.saturating_sub(self.at_unix) < PUBLISH_RETRY_SECS
}
}
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 here
pub(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 avatar
pub(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,
profile: 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.
pub(super) fn sweep_avatars(avatars: &Path, stored: &StoredHandles) {
let kept: BTreeSet<String> = stored
.entries
.values()
.flat_map(StoredHandle::cached_avatar)
.collect();
let Ok(entries) = std::fs::read_dir(avatars) else {
return;
};
for entry in entries.flatten() {
let name = entry.file_name();
if name.to_str().is_some_and(|name| !kept.contains(name)) && entry.path().is_file() {
let _ = std::fs::remove_file(entry.path());
}
}
}
/// the accounts a generation names. a cached handle for any other is dropped
pub(crate) struct Named<'a> {
pub owners: BTreeSet<&'a str>,
/// everyone else the social records name
pub authors: BTreeSet<&'a str>,
}
pub(crate) async fn refresh_handles_with<H: HttpTransport, C: Clock, P: HttpTransport>(
atproto: &Atproto<H, C>,
pds: &P,
avatars: &Path,
named: &Named<'_>,
stored: &mut StoredHandles,
now_unix: u64,
) {
let stale = |did: &&str| {
stored.entries.get(*did).is_none_or(|entry| {
now_unix.saturating_sub(entry.verified_at_unix) >= AUTHOR_REVERIFY_SECS
})
};
let due: BTreeSet<&str> = named
.owners
.iter()
.copied()
.chain(named.authors.iter().copied().filter(stale))
.collect();
let found = verified(atproto, due).await;
let refreshed: Vec<_> = futures::stream::iter(found.into_iter().map(
|(raw_did, (handle, endpoint))| async move {
let found =
tokio::time::timeout(AVATAR_TIMEOUT, profile_of(pds, avatars, raw_did, &endpoint))
.await
.ok()
.flatten();
(raw_did, handle, found)
},
))
.buffer_unordered(REFRESHES_IN_FLIGHT)
.collect()
.await;
for (raw_did, handle, found) in refreshed {
stored.unverified.remove(raw_did);
let previous = stored.entries.get(raw_did);
let ((avatar_cid, avatar_ext), profile) = match found {
Some(found) => (found.avatar.unzip(), found.profile),
None => previous
.map(|entry| {
let avatar = (entry.avatar_cid.clone(), entry.avatar_ext.clone());
(avatar, entry.profile.clone())
})
.unwrap_or_default(),
};
stored.entries.insert(
raw_did.to_string(),
StoredHandle {
handle,
verified_at_unix: now_unix,
avatar_cid,
avatar_ext,
profile,
},
);
}
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>,
raw_did: &str,
) -> Option<(String, Url)> {
let did = AccountDid::new(raw_did).ok()?;
let identity = atproto.resolve_identity(&did).await.ok()?;
let handle = identity.primary_handle()?;
let resolved = atproto.resolve_handle_to_did(handle).await.ok()?;
(resolved == did).then(|| (handle.as_str().to_string(), identity.pds.url().clone()))
}
#[derive(Debug, PartialEq, Eq)]
pub(super) struct Fetched {
/// the cid and extension of the avatar this knot keeps, `None` when it can't serve one
pub avatar: Option<(String, String)>,
pub profile: Option<Map<String, Value>>,
}
/// `None` means the pds failed.
pub(super) async fn profile_of<P: HttpTransport>(
pds: &P,
avatars: &Path,
did: &str,
endpoint: &Url,
) -> Option<Fetched> {
let mut link = None;
let mut profile = None;
for collection in PROFILE_COLLECTIONS {
let query = [("repo", did), ("collection", collection), ("rkey", "self")];
let response = pds
.execute(xrpc(endpoint, "com.atproto.repo.getRecord", &query)?)
.await
.ok()?;
if response.status.is_server_error() {
return None;
}
let record: Value = serde_json::from_slice(&response.body).unwrap_or_default();
if profile.is_none()
&& TANGLED_PROFILES.contains(&collection)
&& response.status.is_success()
{
profile = record["value"]
.is_object()
.then(|| super::actor_profile::sanitized(&record["value"]));
}
if let Some(found) = record["value"]["avatar"]["ref"]["$link"].as_str() {
link = Some(found.to_string());
break;
}
}
let avatar = match link {
Some(link) => avatar_of(pds, avatars, did, endpoint, &link).await?,
None => None,
};
Some(Fetched { avatar, profile })
}
fn xrpc(endpoint: &Url, method: &str, query: &[(&str, &str)]) -> Option<HttpRequest> {
let mut url = endpoint.join(&format!("xrpc/{method}")).ok()?;
url.query_pairs_mut().extend_pairs(query);
Some(HttpRequest::get(url))
}
/// `None` means the pds failed, `Some(None)` means no avatar this knot can serve.
async fn avatar_of<P: HttpTransport>(
pds: &P,
avatars: &Path,
did: &str,
endpoint: &Url,
link: &str,
) -> Option<Option<(String, String)>> {
let Some(cid) = link.parse().ok().and_then(BlobCid::new) else {
return Some(None);
};
let name = cid.to_string();
let served = crate::xrpc::social::avatar_types();
if let Some((_, ext)) = served
.clone()
.find(|(_, ext)| avatars.join(format!("{name}.{ext}")).is_file())
{
if !has_sized(avatars, &name)
&& let Ok(bytes) = std::fs::read(avatars.join(format!("{name}.{ext}")))
{
sized(avatars, &name, &bytes);
}
return Some(Some((name, ext.to_string())));
}
if has_sized(avatars, &name) {
return Some(Some((name, "webp".to_string())));
}
let query = [("did", did), ("cid", name.as_str())];
let mut url = xrpc(endpoint, "com.atproto.sync.getBlob", &query)?.url;
let mut hops = 0;
let response = loop {
let response = match pds.execute(HttpRequest::get(url.clone())).await {
Ok(response) => response,
Err(NetworkError::TooLarge { .. }) => return Some(None),
Err(_) => return None,
};
if !response.status.is_redirection() {
break response;
}
let next = response
.headers
.get(http::header::LOCATION)
.and_then(|location| location.to_str().ok())
.and_then(|location| url.join(location).ok())
.filter(|next| next.scheme() == "https");
hops += 1;
match next {
Some(next) if hops <= AVATAR_REDIRECTS => url = next,
_ => return Some(None),
}
};
if response.status.is_server_error() {
return None;
}
let bytes = &response.body;
let fits = response.status.is_success() && bytes.len() as u64 <= AVATAR_FETCH_MAX;
if !fits || BlobCid::from_digest(Sha256Digest::hash(bytes)) != cid {
return Some(None);
}
let sniffed = knot_record::attachment::sniffed(bytes);
let Some((_, ext)) = served.into_iter().find(|(mime, _)| Some(*mime) == sniffed) else {
return Some(None);
};
let sized = sized(avatars, &name, bytes);
if bytes.len() as u64 <= AVATAR_ORIGINAL_MAX {
let path = avatars.join(format!("{name}.{ext}"));
std::fs::create_dir_all(avatars).ok()?;
crate::cas::atomic_write(&path, bytes).ok()?;
} else if !sized {
return Some(None);
}
Some(Some((name, ext.to_string())))
}
/// an image that won't decode is still served whole, just not at its sizes
fn sized(avatars: &Path, name: &str, bytes: &[u8]) -> bool {
super::avatar_sizes::write_sized(avatars, name, bytes)
.inspect_err(|error| tracing::warn!(cid = name, "no sized avatars: {error:#}"))
.is_ok()
}
pub(super) async fn read_descriptions(
chain_root: &Path,
entries: &[&RepoEntry],
) -> anyhow::Result<BTreeMap<String, String>> {
let mut descriptions = BTreeMap::new();
for entry in entries {
let did = RepoDid::new(entry.repo_did.as_str())?;
if let Some(desc) = published_description(chain_root, &did).await? {
descriptions.insert(entry.repo_did.clone(), desc);
}
}
Ok(descriptions)
}
pub(crate) async fn published_description(
chain_root: &Path,
did: &RepoDid,
) -> anyhow::Result<Option<String>> {
#[derive(serde::Deserialize)]
struct Manifest {
#[serde(default)]
description: Option<RepoDescription>,
}
let path = chain_root.join(repo_key(did.as_str())).join("chain.git");
if !path.exists() {
return Ok(None);
}
let chain_repo =
Repo::open(&path).with_context(|| format!("open atproto chain {}", path.display()))?;
let Some(tip) = chain::tip(&chain_repo)? else {
return Ok(None);
};
let key = format!("{MANIFEST_COLLECTION}/{MANIFEST_RKEY}");
let tree = chain::load_tree(&chain_repo, &tip);
let Some(cid) = tree.get(&key).await? else {
return Ok(None);
};
let Some(bytes) = GitBlocks::at(&chain_repo, tip.blocks).get(&cid).await? else {
return Ok(None);
};
Ok(serde_ipld_dagcbor::from_slice::<Manifest>(&bytes)?
.description
.map(|d| d.as_str().to_string()))
}
pub(super) fn read_activity(
config: &ValidatedConfig,
entries: &[&RepoEntry],
) -> BTreeMap<String, i64> {
let Ok(layout) = config.layout() else {
return BTreeMap::new();
};
let mut activity = BTreeMap::new();
for entry in entries {
let Ok(did) = RepoDid::new(entry.repo_did.as_str()) else {
continue;
};
let Ok(repo) = layout.open(&did) else {
continue;
};
let gix_repo = repo.git();
let newest = entry
.public_refs
.iter()
.filter_map(|(_, hex)| {
let oid = gix::ObjectId::from_hex(hex.as_bytes()).ok()?;
let mut obj = gix_repo.find_object(oid).ok()?;
if obj.kind == gix::object::Kind::Tag {
obj = obj.peel_to_kind(gix::object::Kind::Commit).ok()?;
}
let commit = obj.try_into_commit().ok()?;
Some(commit.committer().ok()?.time().ok()?.seconds)
})
.max();
if let Some(ts) = newest {
activity.insert(entry.repo_did.clone(), ts);
}
}
activity
}