Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460use axum::extract::State;use axum::response::Response;use futures::future::OptionFuture;use futures::stream::{self, StreamExt, TryStreamExt};use lexicons::sh_tangled::feed::list_comments::ListComments;use serde::Deserialize;
use bobbin_edge_index::{ EdgeItem, EdgePage, EdgeStore, PageCursor, PageLimit, PageOffset, PageToken, SortDir,};use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi};use bobbin_types::sh_tangled::actor::{ProfileViewBasic, ProfileViewDetailed, ViewerState};use bobbin_types::sh_tangled::embed::commit::View as CommitEmbedView;use bobbin_types::sh_tangled::feed::CommentView;use bobbin_types::sh_tangled::feed::comment::{Comment, CommentRecord};use bobbin_types::sh_tangled::feed::get_timeline::{ FollowEvent, RepoEvent, StarEvent, TimelineItem, TimelineItemEvent,};use bobbin_types::sh_tangled::feed::star::{Star, StarRecord, StarSubject};use bobbin_types::sh_tangled::graph::follow::{Follow, FollowRecord};use bobbin_types::sh_tangled::repo::{self, Repo, RepoRecord, RepoViewBasic};use jacquard_axum::service_auth::ExtractOptionalServiceAuth;use jacquard_common::types::string::{AtUri, Datetime, Did, Handle, UriValue};use jacquard_common::xrpc::XrpcResp;use jacquard_common::{DefaultStr, IntoStatic as _};
use crate::{ AppState, HasSubject, HitProvenance, RecordView, SubjectQuery, XrpcError, XrpcQuery, fetch, hydrate_record_stream, json_stream, paged_tail, parse_cursor, parse_limit, parse_start, parse_subject,};
const REPO_NSID: &str = "sh.tangled.repo";const STAR_NSID: &str = "sh.tangled.feed.star";const FOLLOW_NSID: &str = "sh.tangled.graph.follow";const STAR_BY_NSID: &str = "sh.tangled.feed.star.by";const FOLLOW_BY_NSID: &str = "sh.tangled.graph.follow.by";
const HYDRATE_CONCURRENCY: usize = 8;
pub(crate) async fn list_comments( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<ListComments>,) -> Result<Response, XrpcError> { let subject = parse_subject(&SubjectQuery::Uri(q.subject), CommentRecord::SHAPE)?; let offset = q .offset .map(|offset| { u64::try_from(offset) .map(PageOffset::new) .map_err(|_| XrpcError::InvalidParams("offset must not be negative".into())) }) .transpose()?; let start = parse_start(q.cursor.as_deref(), offset)?; let limit = parse_limit( q.limit .map(|limit| { u32::try_from(limit).map_err(|_| { XrpcError::InvalidParams("limit must be between 1 and 1000".into()) }) }) .transpose()?, )?; let dir = match q.order.as_deref() { Some("asc") => SortDir::Asc, _ => SortDir::Desc, };
let nsid = nsid_static(CommentRecord::NSID); let page = state .edges .list(&EdgeKey::new(nsid.clone(), subject), start, limit, dir); let permit = state.heavy_permit()?; let owned = state.clone(); let items = hydrate_record_stream::<Comment<DefaultStr>>( &state, nsid, page.items, HitProvenance::Indexed, ) .map_ok(move |view| { let state = owned.clone(); async move { Ok(comment_view(&state, view).await) } }) .try_buffered(HYDRATE_CONCURRENCY) .try_filter_map(|view| async move { Ok(view) });
Ok(json_stream::<CommentView<DefaultStr>, _>( "items", items, paged_tail(page.next.map(PageToken::encode_token), page.total), permit, ))}
async fn comment_view( state: &AppState, view: RecordView<Comment<DefaultStr>>,) -> Option<CommentView<DefaultStr>> { let RecordView { uri, cid, value } = view; let embed = OptionFuture::from(value.embed.map(|commit| async { let change = state .change_ids .get( state.mirror_v2.as_deref(), &commit.repo, commit.commit.oid.as_ref(), ) .await; CommitEmbedView { change, commit: commit.commit.oid, repo: commit.repo, extra_data: None, } })) .await; Some(CommentView { body: value.body, cid: cid?, created_at: value.created_at, embed, reply_to: value.reply_to, subject: value.subject, uri, extra_data: None, })}
#[derive(Debug, Deserialize)]#[serde(rename_all = "camelCase")]pub(crate) struct GetTimelineQuery { #[serde(default)] following_only: bool, limit: Option<u32>, cursor: Option<String>,}
#[axum::debug_handler]pub(crate) async fn get_timeline( State(state): State<AppState>, ExtractOptionalServiceAuth(auth): ExtractOptionalServiceAuth, XrpcQuery(q): XrpcQuery<GetTimelineQuery>,) -> Result<Response, XrpcError> { let limit = parse_limit(q.limit)?; let cursor = parse_cursor(q.cursor.as_deref())?; let permit = state.heavy_permit()?; let viewer = auth.as_ref().map(|auth| auth.did().into_static());
// Prepare timeline skeleton let page = if q.following_only { // Following feed: the viewer's followed set, then a time-ordered k-way merge. let viewer = viewer.as_ref().ok_or_else(|| { XrpcError::AuthRequired("followingOnly requires an authenticated viewer".into()) })?; let followed = followed_dids(&state, &viewer).await; following_skeleton(&state, &followed, cursor, limit) } else { global_skeleton(&state, cursor, limit) };
// Hydrate skeleton let hydrated = stream::iter( page.items .iter() .map(|it| hydrate_item(&state, viewer.as_ref(), it)) .collect::<Vec<_>>(), ) .buffered(HYDRATE_CONCURRENCY) .collect::<Vec<_>>() .await;
let mut feed: Vec<TimelineItem<DefaultStr>> = Vec::with_capacity(hydrated.len()); for (item, r) in page.items.iter().zip(hydrated) { match r { Ok(item) => feed.push(item), Err(e) => { tracing::warn!(uri = %item.uri, error = %e, "skipping timeline item, hydration failed") } } }
let items = stream::iter(feed.into_iter().map(Ok::<_, XrpcError>)); Ok(json_stream::<TimelineItem<DefaultStr>, _>( "feed", items, paged_tail(page.next.map(PageToken::encode_token), page.total), permit, ))}
async fn followed_dids(state: &AppState, viewer: &Did<DefaultStr>) -> Vec<Did<DefaultStr>> { let key = EdgeKey::new(nsid_static(FOLLOW_BY_NSID), SubjectRef::Did(viewer.clone())); let uris = state.edges.sources_for(&key); // TODO: store follow actor information in edge stream::iter(uris.into_iter().map(|uri| async move { match fetch::<FollowRecord, Follow<DefaultStr>>(state, &uri).await { Ok((_, follow)) => Some(follow.subject), Err(e) => { tracing::warn!(uri = %uri, error = %e, "skipping followed did, follow record unresolved"); None } } })) .buffered(HYDRATE_CONCURRENCY) .collect::<Vec<_>>() .await .into_iter() .flatten() .collect()}
fn following_skeleton( state: &AppState, followed: &[Did<DefaultStr>], cursor: PageCursor, limit: PageLimit,) -> EdgePage { let keys: Vec<EdgeKey> = followed .iter() .flat_map(|u| { [REPO_NSID, STAR_BY_NSID, FOLLOW_BY_NSID] .map(|kind| EdgeKey::new(nsid_static(kind), SubjectRef::Did(u.clone()))) }) .collect(); state.edges.list_multi(&keys, cursor, limit, SortDir::Desc)}
fn global_skeleton(state: &AppState, cursor: PageCursor, limit: PageLimit) -> EdgePage { let keys = [REPO_NSID, STAR_NSID, FOLLOW_NSID] .map(|kind| EdgeKey::new(nsid_static(kind), SubjectRef::Global)); state.edges.list_multi(&keys, cursor, limit, SortDir::Desc)}
async fn hydrate_item( state: &AppState, viewer: Option<&Did<DefaultStr>>, item: &EdgeItem,) -> Result<TimelineItem<DefaultStr>, XrpcError> { let collection = item .uri .collection() .ok_or_else(|| XrpcError::InvalidParams("timeline uri missing collection".into()))?; let (event, event_at) = match collection.as_ref() { REPO_NSID => hydrate_repo_event(state, viewer, &item.uri).await?, STAR_NSID => hydrate_star_event(state, viewer, &item.uri).await?, FOLLOW_NSID => hydrate_follow_event(state, viewer, &item.uri).await?, other => { return Err(XrpcError::InvalidRecord(format!( "unexpected timeline collection: {other}" ))); } }; Ok(TimelineItem::new().event(event).event_at(event_at).build())}
async fn hydrate_repo_event( state: &AppState, viewer: Option<&Did<DefaultStr>>, uri: &AtUri<DefaultStr>,) -> Result<(TimelineItemEvent<DefaultStr>, Datetime), XrpcError> { let (body, repo) = fetch::<RepoRecord, Repo<DefaultStr>>(state, uri).await?; let repo_did = repo.repo_did.clone().ok_or_else(|| { XrpcError::InvalidRecord("repo has no repo_did; cannot build repoViewBasic".into()) })?; let view = build_repo_view_basic(state, viewer, uri, &repo, repo_did).await?; let source = match &repo.source { Some(UriValue::Did(did)) => build_repo_view_from_did(state, viewer, did).await.ok(), Some(UriValue::At(_aturi)) => None, // we drop the legacy support _ => None, }; let event_at = repo.created_at.clone(); let ev = RepoEvent::new() .uri(body.uri.clone()) .cid(body.cid.clone()) .repo(view) .source(source) .build(); Ok((TimelineItemEvent::RepoEvent(Box::new(ev)), event_at))}
async fn hydrate_star_event( state: &AppState, viewer: Option<&Did<DefaultStr>>, uri: &AtUri<DefaultStr>,) -> Result<(TimelineItemEvent<DefaultStr>, Datetime), XrpcError> { let (body, star) = fetch::<StarRecord, Star<DefaultStr>>(state, uri).await?; let starrer = owner_did_from_aturi(uri) .ok_or_else(|| XrpcError::InvalidParams("star uri authority must be a did".into()))?; let actor = build_profile_basic(state, viewer, &starrer).await; let repo_did = match &star.subject { StarSubject::Repo(r) => r.did.clone(), StarSubject::String(_) => { return Err(XrpcError::InvalidRecord( "timeline star subject is not a repo".into(), )); } }; let repo = build_repo_view_from_did(state, viewer, &repo_did).await?; let event_at = star.created_at.clone(); let ev = StarEvent::new() .uri(body.uri.clone()) .cid(body.cid.clone()) .actor(actor) .repo(repo) .build(); Ok((TimelineItemEvent::StarEvent(Box::new(ev)), event_at))}
async fn hydrate_follow_event( state: &AppState, viewer: Option<&Did<DefaultStr>>, uri: &AtUri<DefaultStr>,) -> Result<(TimelineItemEvent<DefaultStr>, Datetime), XrpcError> { let (body, follow) = fetch::<FollowRecord, Follow<DefaultStr>>(state, uri).await?; let follower = owner_did_from_aturi(uri) .ok_or_else(|| XrpcError::InvalidParams("follow uri authority must be a did".into()))?; let actor = build_profile_basic(state, viewer, &follower).await; let subject = build_profile_detailed(state, viewer, &follow.subject).await; let event_at = follow.created_at.clone(); let ev = FollowEvent::new() .uri(body.uri.clone()) .cid(body.cid.clone()) .actor(actor) .subject(subject) .build(); Ok((TimelineItemEvent::FollowEvent(Box::new(ev)), event_at))}
async fn build_repo_view_from_did( state: &AppState, viewer: Option<&Did<DefaultStr>>, repo_did: &Did<DefaultStr>,) -> Result<RepoViewBasic<DefaultStr>, XrpcError> { let ident = state .resolver .lookup_by_repo_did(repo_did) .await .ok_or(XrpcError::NotFound)?; let uri = AtUri::<DefaultStr>::from_parts_owned( ident.owner.as_str(), RepoRecord::NSID, ident.rkey.as_str(), ) .map_err(|e| XrpcError::InvalidRecord(format!("repo uri assembly: {e}")))?; let (_, repo) = fetch::<RepoRecord, Repo<DefaultStr>>(state, &uri).await?; build_repo_view_basic(state, viewer, &uri, &repo, repo_did.clone()).await}
async fn build_repo_view_basic( state: &AppState, viewer: Option<&Did<DefaultStr>>, repo_uri: &AtUri<DefaultStr>, repo_record: &Repo<DefaultStr>, repo_did: Did<DefaultStr>,) -> Result<RepoViewBasic<DefaultStr>, XrpcError> { let owner_did = owner_did_from_aturi(repo_uri) .ok_or_else(|| XrpcError::InvalidParams("repo uri authority must be a did".into()))?; let owner = build_profile_basic(state, viewer, &owner_did).await; let slug = repo_record.name.clone().unwrap_or_else(|| { repo_uri .rkey() .map(|r| DefaultStr::from(r.as_ref())) .unwrap_or_default() }); let star_key = EdgeKey::new(nsid_static(STAR_NSID), SubjectRef::Did(repo_did.clone())); let star_count = state.edges.count(&star_key) as i64; let viewer = viewer.map(|v| { let mut viewer = repo::ViewerState::default(); viewer.star = state.edges.viewer_source(&star_key, v.as_str()); viewer }); Ok(RepoViewBasic::new() .did(repo_did) .owner(owner) .slug(slug) .created_at(repo_record.created_at.clone()) .description(repo_record.description.clone()) .star_count(star_count) .viewer(viewer) .build())}
async fn build_profile_basic( state: &AppState, viewer: Option<&Did<DefaultStr>>, did: &Did<DefaultStr>,) -> ProfileViewBasic<DefaultStr> { let handle = resolve_handle(state, did); let avatar = None; // state.avatar.as_ref().and_then(|s| s.url(did)); let viewer_state = build_actor_viewer_state(&state.edges, viewer, did); ProfileViewBasic::new() .did(did.clone()) .handle(handle) .maybe_avatar(avatar) .maybe_viewer(viewer_state) .build()}
async fn build_profile_detailed( state: &AppState, viewer: Option<&Did<DefaultStr>>, did: &Did<DefaultStr>,) -> ProfileViewDetailed<DefaultStr> { let handle = resolve_handle(state, did); let avatar = None; // state.avatar.as_ref().and_then(|s| s.url(did)); let viewer_state = build_actor_viewer_state(&state.edges, viewer, did); let followers = state.edges.count(&EdgeKey::new( nsid_static(FOLLOW_NSID), SubjectRef::Did(did.clone()), )) as i64; let follows = state.edges.count(&EdgeKey::new( nsid_static(FOLLOW_BY_NSID), SubjectRef::Did(did.clone()), )) as i64; ProfileViewDetailed::new() .did(did.clone()) .handle(handle) .followers_count(followers) .follows_count(follows) .maybe_avatar(avatar) .maybe_viewer(viewer_state) .build()}
fn build_actor_viewer_state( edges: &EdgeStore, viewer: Option<&Did<DefaultStr>>, subject: &Did<DefaultStr>,) -> Option<ViewerState<DefaultStr>> { let viewer = viewer?; if viewer == subject { return None; } let following = edges.viewer_source( &EdgeKey::new(nsid_static(FOLLOW_NSID), SubjectRef::Did(subject.clone())), viewer.as_str(), ); let followed_by = edges.viewer_source( &EdgeKey::new(nsid_static(FOLLOW_NSID), SubjectRef::Did(viewer.clone())), subject.as_str(), ); (following.is_some() || followed_by.is_some()).then(|| ViewerState { following, followed_by, ..Default::default() })}
fn resolve_handle(state: &AppState, did: &Did<DefaultStr>) -> Handle<DefaultStr> { state .identity .get_by_did(did) .ok() .map(|doc| doc.handle) .unwrap_or_else(|| { Handle::new_static("handle.invalid").expect("handle.invalid is a valid handle") })}