Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177use std::time::{Duration, SystemTime, UNIX_EPOCH};
use axum::{Json, extract::State};use bobbin_types::sh_tangled::actor::get_trending::GetTrendingOutput;use futures::stream::{self, StreamExt};use jacquard_common::types::did::Did;use tokio::time::Instant;
use crate::{ AppState, FETCH_CONCURRENCY, XrpcError, view::{build_profile_detailed, build_repo_view_basic},};
const OUTPUT_LIMIT: usize = 4;// +1 are fallbacks in case a returned person is the viewer or a repo isnt resolvableconst PEOPLE_QUERY_LIMIT: usize = OUTPUT_LIMIT + 1;const REPO_QUERY_LIMIT: usize = OUTPUT_LIMIT + 1;const TRENDING_WINDOW: Duration = Duration::from_secs(7 * 24 * 60 * 60);const CACHE_TTL: Duration = Duration::from_secs(60 * 60);
#[derive(Clone, Default)]pub(crate) struct TrendingCandidates { people: Vec<Did>, repos: Vec<Did>,}
#[axum::debug_handler]pub(crate) async fn get_trending( State(state): State<AppState>,) -> Result<Json<GetTrendingOutput>, XrpcError> { let permit = state.heavy_permit()?;
let candidates = { let mut cached = state.trending_cache.lock().await; let now = Instant::now(); if cached .as_ref() .is_none_or(|(refreshed_at, _)| now - *refreshed_at >= CACHE_TTL) { let candidates = TrendingCandidates { people: state.edges.most_followed(PEOPLE_QUERY_LIMIT), repos: state .edges .top_starred_since(trending_cutoff_micros(), REPO_QUERY_LIMIT), }; if state.coverage.snapshot().is_ready() { *cached = Some((now, candidates.clone())); } candidates } else { cached .as_ref() .expect("fresh trending cache entry") .1 .clone() } };
let people = stream::iter(candidates.people) .map({ let state = &state; move |did| async move { build_profile_detailed(state, None, &did).await } }) .buffered(FETCH_CONCURRENCY) .collect() .await;
let hydrated = stream::iter(candidates.repos) .map({ let state = &state; move |did| async move { let result = build_repo_view_basic(state, None, &did).await; (did, result) } }) .buffered(FETCH_CONCURRENCY) .collect::<Vec<_>>() .await; let repos = hydrated .into_iter() .filter_map(|(did, result)| match result { Ok(repo) => Some(repo), Err(error) => { tracing::warn!(repo_did = %did, %error, "skipping trending repo, hydration failed"); None } }) .take(OUTPUT_LIMIT) .collect();
drop(permit); Ok(Json(GetTrendingOutput { people, repos, extra_data: None, }))}
fn trending_cutoff_micros() -> u64 { let elapsed = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() .saturating_sub(TRENDING_WINDOW); elapsed .as_secs() .saturating_mul(1_000_000) .saturating_add(elapsed.subsec_micros().into())}
#[cfg(test)]mod tests { use super::*; use bobbin_edge_index::EdgeStore; use bobbin_runtime::RuntimeHasher; use bobbin_types::{ edges::Edge, ids::{SubjectRef, nsid_static}, }; use jacquard_common::types::string::AtUri;
fn did(value: &str) -> Did { Did::new_owned(value).unwrap() }
#[test] fn trending_candidates_are_ranked_from_the_edge_index() { let edges = EdgeStore::new(RuntimeHasher::default()); let alpha = did("did:plc:alpha"); let beta = did("did:plc:beta"); let compiler = did("did:plc:compiler"); let editor = did("did:plc:editor"); let now = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap() .as_micros() as u64;
[ ("did:plc:a", "a1", alpha.clone()), ("did:plc:b", "a2", alpha.clone()), ("did:plc:c", "b1", beta.clone()), ] .into_iter() .for_each(|(author, rkey, subject)| { edges.add(Edge { kind: nsid_static("sh.tangled.graph.follow"), subject: SubjectRef::Did(subject), source: AtUri::new_owned(format!("at://{author}/sh.tangled.graph.follow/{rkey}")) .unwrap(), sort_micros: now, }); }); [ ("did:plc:a", "c1", compiler.clone(), now), ("did:plc:b", "c2", compiler.clone(), now), ("did:plc:c", "e1", editor.clone(), now), ] .into_iter() .for_each(|(author, rkey, subject, sort_micros)| { edges.add(Edge { kind: nsid_static("sh.tangled.feed.star"), subject: SubjectRef::Did(subject), source: AtUri::new_owned(format!("at://{author}/sh.tangled.feed.star/{rkey}")) .unwrap(), sort_micros, }); });
let candidates = TrendingCandidates { people: edges.most_followed(PEOPLE_QUERY_LIMIT), repos: edges.top_starred_since(trending_cutoff_micros(), REPO_QUERY_LIMIT), };
assert_eq!(candidates.people, vec![alpha, beta]); assert_eq!(candidates.repos, vec![compiler, editor]); }}