Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671367236733674367536763677367836793680368136823683368436853686368736883689369036913692369336943695369636973698369937003701use std::future::Future;use std::sync::Arc;
use bobbin_resolver::{NormalizeRepoRefs, decode_record, normalize_record_fields};
use axum::{ Router, body::{Body, Bytes}, extract::{FromRequestParts, Query, RawQuery, State, rejection::QueryRejection}, http::{ HeaderMap, HeaderName, HeaderValue, Method, StatusCode, header::{ ACCEPT_RANGES, AUTHORIZATION, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_SECURITY_POLICY, CONTENT_TYPE, ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, IF_RANGE, LAST_MODIFIED, RANGE, X_CONTENT_TYPE_OPTIONS, }, request::Parts, }, response::{IntoResponse, Json, Response}, routing::get,};pub use bobbin_codesearch::{CodeSearch, CodeSearchError};use bobbin_edge_index::{ ClaimIndex, Coverage, CoverageWatch, CursorParseError, EdgeItem, EdgePage, EdgeStore, InviteIndex, InviteScope, Offer, PageCursor, PageLimit, PageOffset, PageStart, PageToken, ParsedCid, Rejection, Settlement, Settlements, SortDir, StateKind, StreamHealth, StreamHealthSnapshot, TicketStateKind, TotalCount, Watch,};use bobbin_knot_proxy::{KnotProxyError, MirrorProxy, ProxyResponse};use bobbin_record_lru::RecordStore;use bobbin_resolver::{IdentityResolveError, IdentityResolver, RepoIdResolver};use bobbin_runtime::{ReqwestHttp, UnixMicros};use bobbin_search::{ ActorIndex, ActorIndexError, SearchCursor, SearchError, SearchFilters, SearchHit, SearchOffset, SearchReader,};use bobbin_slingshot_client::{SlingshotClient, SlingshotError};use bobbin_types::edges::REPO_SOURCE_EDGE_KIND;use bobbin_types::edges::{TICKET_ISSUE_KIND, TICKET_PULL_KIND};use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi};use bobbin_types::org_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord};use bobbin_types::org_tangled::feed::comment::{ Comment as FeedComment, CommentRecord as FeedCommentRecord,};use bobbin_types::org_tangled::feed::reaction::{Reaction, ReactionRecord};use bobbin_types::org_tangled::feed::star::{Star, StarRecord};use bobbin_types::org_tangled::feed::subscription::SubscriptionRecord;use bobbin_types::org_tangled::git::refs::{Refs, RefsRecord};use bobbin_types::org_tangled::graph::collaborator::{Collaborator, CollaboratorRecord};use bobbin_types::org_tangled::graph::follow::{Follow, FollowRecord};use bobbin_types::org_tangled::graph::vouch::{Vouch, VouchRecord};use bobbin_types::org_tangled::key::ssh_key::{SshKey, SshKeyGetRecordOutput, SshKeyRecord};use bobbin_types::org_tangled::label::definition::{ Definition as LabelDefinition, DefinitionRecord as LabelDefinitionRecord,};use bobbin_types::org_tangled::repo::manifest::{ Manifest, ManifestGetRecordOutput, ManifestRecord,};use bobbin_types::org_tangled::review::patch::{Patch, PatchRecord};use bobbin_types::org_tangled::snippet::string::{ SnippetString, SnippetStringGetRecordOutput, SnippetStringRecord,};use bobbin_types::org_tangled::track::ticket::{Ticket, TicketGetRecordOutput, TicketRecord};use bobbin_types::record::RecordBody;use bobbin_types::search::SearchableRecord;use bobbin_types::sh_tangled::repo::artifact::{Artifact, ArtifactRecord};
use futures::Stream;use futures::stream::{self, StreamExt, TryStreamExt};use jacquard_axum::service_auth::{self, ExtractOptionalServiceAuth, ServiceAuthConfig};use jacquard_common::types::datetime::Datetime;use jacquard_common::types::did::Did;use jacquard_common::types::ident::AtIdentifier;use jacquard_common::types::nsid::Nsid;use jacquard_common::types::string::{AtUri, Cid};use jacquard_common::xrpc::XrpcResp;use jacquard_common::{DefaultStr, IntoStatic};use jacquard_identity::JacquardResolver;use serde::{Deserialize, Deserializer, Serialize};use std::convert::Infallible;use std::time::Duration;use thiserror::Error;use tokio::time::Instant;use url::form_urlencoded;
use tower_http::classify::ServerErrorsFailureClass;use tower_http::cors::{AllowOrigin, CorsLayer};use tower_http::trace::{DefaultMakeSpan, OnFailure, OnResponse, TraceLayer};use tracing::{Level, Span};
mod actor;mod backpressure;mod change_ids;mod client_address;mod codesearch;mod commit_stats;mod enrich;mod feed;mod filter;mod labels;mod pocket;mod profile_activity;mod recordpath;mod repo;mod view;
pub use backpressure::{ AwaitLimiter, AwaitPermit, HeavyLimiter, HeavyPermit, MaxAwaiting, MaxInFlight, PerRequestAnonBytes, PressureVerdict, ReservedFloor,};use client_address::X_FORWARDED_FOR;pub use client_address::{ClientAddress, SocketPeer};use filter::{ CountFilter, IssueFilter, ListFilter, NoFilter, PullFilter, StateViewQuery, VouchNetworkFilter,};pub use labels::Moderation;pub use pocket::PocketProxy;use trusted_proxies::TrustedProxies;
const DEFAULT_LIMIT: u32 = 50;const ACTOR_TYPEAHEAD_DEFAULT_LIMIT: u32 = 10;const ACTOR_TYPEAHEAD_MAX_LIMIT: u32 = 100;const ACTOR_TYPEAHEAD_MAX_QUERY_BYTES: usize = 253;const FETCH_CONCURRENCY: usize = 8;const TANGLED_NSID_PREFIX: &str = "sh.tangled.";const ORG_TANGLED_NSID_PREFIX: &str = "org.tangled.";const MAX_OUTSTANDING_OFFERS: usize = 128;
pub type Directory = JacquardResolver<ReqwestHttp>;
#[derive(Clone, Debug, Default)]pub enum Cors { #[default] Off, Any, Origins(Vec<HeaderValue>),}
#[derive(Debug, Error)]#[error("{0}")]pub struct CorsOriginError(String);
impl Cors { pub fn parse_env(raw: &str) -> Result<Self, CorsOriginError> { Self::from_origins( raw.split(',') .map(str::trim) .filter(|entry| !entry.is_empty()), ) }
fn from_origins<'a>( origins: impl IntoIterator<Item = &'a str>, ) -> Result<Self, CorsOriginError> { let origins: Vec<&str> = origins.into_iter().collect(); match origins.as_slice() { [] => Ok(Self::Off), ["*"] => Ok(Self::Any), _ if origins.contains(&"*") => Err(CorsOriginError( "CORS wildcard `*` cannot be combined with exact origins".into(), )), _ => origins .into_iter() .map(|origin| { let url = url::Url::parse(origin).map_err(|e| { CorsOriginError(format!("invalid CORS origin `{origin}`: {e}")) })?; if !matches!(url.scheme(), "http" | "https") || url.origin().ascii_serialization() != origin { return Err(CorsOriginError(format!( "CORS origin must be an exact http(s) origin: `{origin}`" ))); } origin .parse() .map_err(|e| CorsOriginError(format!("invalid CORS header: {e}"))) }) .collect::<Result<Vec<_>, _>>() .map(Self::Origins), } }}
impl<'de> Deserialize<'de> for Cors { fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> { let origins = Vec::<String>::deserialize(deserializer)?; Self::from_origins(origins.iter().map(String::as_str)).map_err(serde::de::Error::custom) }}
pub fn default_directory() -> Directory { JacquardResolver::new(ReqwestHttp::new(reqwest::Client::new()), Default::default())}
#[derive(Clone)]pub struct AppState { pub records: Arc<dyn RecordStore>, pub slingshot: SlingshotClient, pub edges: Arc<EdgeStore>, pub coverage: Arc<CoverageWatch>, pub mirror: Option<Arc<MirrorProxy>>, pub search: Arc<dyn SearchReader>, /// Zoekt-backed code search. `None` when unconfigured, and the endpoint answers 501. pub codesearch: Option<Arc<CodeSearch>>, /// Pocket, holding per-account preferences. `None` when unconfigured, and the /// two `com.bad-example.pocket.*` endpoints answer 501. pub pocket: Option<Arc<PocketProxy>>, pub actors: Arc<ActorIndex>, pub resolver: Arc<RepoIdResolver>, pub identity: Arc<IdentityResolver>, pub directory: Arc<Directory>, pub limiter: Option<Arc<HeavyLimiter>>, pub awaiting: Arc<AwaitLimiter>, pub client_address: Arc<ClientAddress>, pub settlements: Arc<Settlements>, pub invites: Arc<InviteIndex>, pub claims: Arc<ClaimIndex>, /// the ingest stream's liveness, absent when nothing publishes it pub stream_health: Option<Arc<StreamHealth>>, pub moderation: Option<Moderation>, cors: Cors, service_auth_config: service_auth::ServiceAuthConfig<Arc<Directory>>, enrich_router: Arc<std::sync::OnceLock<Router>>, trending_cache: Arc<tokio::sync::Mutex<Option<(Instant, actor::TrendingCandidates)>>>, commit_stats_cache: Arc<commit_stats::CommitStatsCache>, change_ids: Arc<change_ids::ChangeIdCache>,}
impl AppState { #[allow(clippy::too_many_arguments)] pub fn new( records: Arc<dyn RecordStore>, slingshot: SlingshotClient, edges: Arc<EdgeStore>, coverage: Arc<CoverageWatch>, search: Arc<dyn SearchReader>, resolver: Arc<RepoIdResolver>, directory: Arc<Directory>, ) -> Self { let identity = Arc::new(IdentityResolver::with_slingshot( slingshot.clone(), Arc::new(bobbin_runtime::SystemClock::new()), bobbin_runtime::RuntimeHasher::default(), )); Self { records, slingshot, edges, coverage, mirror: None, search, codesearch: None, pocket: None, actors: Arc::new(ActorIndex::new()), resolver, identity, directory: directory.clone(), limiter: None, awaiting: Arc::new(AwaitLimiter::new(MaxAwaiting::default())), client_address: Arc::new(ClientAddress::default()), settlements: Arc::new(Settlements::new()), invites: Arc::new(InviteIndex::new()), claims: Arc::new(ClaimIndex::new()), stream_health: None, moderation: None, cors: Cors::Off, service_auth_config: ServiceAuthConfig::new( Did::new_static("did:web:localhost").unwrap(), directory, ), enrich_router: Arc::new(std::sync::OnceLock::new()), trending_cache: Arc::new(tokio::sync::Mutex::new(None)), commit_stats_cache: Arc::new(commit_stats::CommitStatsCache::default()), change_ids: Arc::new(change_ids::ChangeIdCache::default()), } }
pub fn with_actors(mut self, actors: Arc<ActorIndex>) -> Self { self.actors = actors; self } pub fn with_limiter(mut self, limiter: Option<Arc<HeavyLimiter>>) -> Self { self.limiter = limiter; self }
pub fn with_max_awaiting(mut self, max: MaxAwaiting) -> Self { self.awaiting = Arc::new(AwaitLimiter::new(max)); self }
pub fn with_mirror(mut self, mirror: Option<Arc<MirrorProxy>>) -> Self { self.mirror = mirror; self }
pub fn with_codesearch(mut self, codesearch: Option<Arc<CodeSearch>>) -> Self { self.codesearch = codesearch; self }
pub fn with_pocket(mut self, pocket: Option<Arc<PocketProxy>>) -> Self { self.pocket = pocket; self }
pub fn with_identity(mut self, identity: Arc<IdentityResolver>) -> Self { self.identity = identity; self }
pub fn with_proxies(mut self, proxies: TrustedProxies) -> Self { self.client_address = Arc::new(ClientAddress::new(proxies)); self }
pub fn with_settlements(mut self, settlements: Arc<Settlements>) -> Self { self.settlements = settlements; self }
pub fn with_invites(mut self, invites: Arc<InviteIndex>) -> Self { self.invites = invites; self }
pub fn with_claims(mut self, claims: Arc<ClaimIndex>) -> Self { self.claims = claims; self }
pub fn with_moderation(mut self, moderation: Option<Moderation>) -> Self { self.moderation = moderation; self }
pub fn with_stream_health(mut self, stream_health: Arc<StreamHealth>) -> Self { self.stream_health = Some(stream_health); self }
pub fn with_cors(mut self, cors: Cors) -> Self { self.cors = cors; self }
pub fn with_service_did(mut self, did: Did) -> Self { self.service_auth_config = ServiceAuthConfig::new(did, self.directory.clone()); self }
/// for internal xrpc dispatch [`enrich`] pub fn self_router(&self) -> Router { self.enrich_router .get_or_init(|| router(self.clone())) .clone() }
pub(crate) fn heavy_permit(&self) -> Result<Option<HeavyPermit>, XrpcError> { self.limiter.as_ref().map(|l| l.try_enter()).transpose() }}
impl service_auth::ServiceAuth for AppState { type Resolver = Directory;
fn service_did(&self) -> Did<&str> { self.service_auth_config.service_did() } fn resolver(&self) -> &Self::Resolver { self.service_auth_config.resolver() } fn require_lxm(&self) -> bool { service_auth::ServiceAuth::require_lxm(&self.service_auth_config) } fn allowed_services(&self) -> &[jacquard_common::SmolStr] { self.service_auth_config.allowed_services() } fn replay_protection_enabled(&self) -> bool { self.service_auth_config.replay_protection_enabled() } fn replay_store(&self) -> &dyn jacquard_axum::service_auth::ReplayStore { self.service_auth_config.replay_store() }}
pub fn router(state: AppState) -> Router { let cors = match &state.cors { Cors::Off => None, Cors::Any => Some(AllowOrigin::any()), Cors::Origins(origins) => Some(AllowOrigin::list(origins.clone())), } .map(|origins| { CorsLayer::new() .allow_origin(origins) .allow_methods([Method::GET, Method::POST, Method::OPTIONS]) .allow_headers([ AUTHORIZATION, CONTENT_TYPE, HeaderName::from_static("atproto-proxy"), HeaderName::from_static("atproto-accept-labelers"), HeaderName::from_static("range"), HeaderName::from_static("if-range"), HeaderName::from_static("if-none-match"), HeaderName::from_static("if-modified-since"), ]) .expose_headers([ CONTENT_TYPE, HeaderName::from_static("ratelimit-limit"), HeaderName::from_static("ratelimit-remaining"), HeaderName::from_static("ratelimit-reset"), ]) .max_age(Duration::from_secs(86400)) }); let router = Router::new() .route("/xrpc/sh.tangled.repo.getRepo", get(get_repo)) .route("/xrpc/sh.tangled.repo.getRepos", get(get_repos)) .route( "/xrpc/sh.tangled.repo.getRepoByRepoDid", get(get_repo_by_repo_did), ) .route( "/xrpc/sh.tangled.repo.getReposByRepoDids", get(get_repos_by_repo_dids), ) .route("/xrpc/sh.tangled.repo.getRepoByName", get(get_repo_by_name)) .route("/xrpc/sh.tangled.actor.getProfile", get(get_profile)) .route("/xrpc/sh.tangled.actor.getProfiles", get(get_profiles)) .route( "/xrpc/sh.tangled.actor.getProfileActivity", get(profile_activity::get_profile_activity), ) .route( "/xrpc/sh.tangled.actor.getTrending", get(actor::get_trending), ) .route("/xrpc/org.tangled.track.getTicket", get(get_ticket)) .route("/xrpc/org.tangled.track.getTickets", get(get_tickets)) .route( "/xrpc/org.tangled.track.getTicketView", get(repo::get_ticket_view), ) .route( "/xrpc/org.tangled.temp.review.listPatches", get(list_patches), ) .route("/xrpc/sh.tangled.feed.listStars", get(list_stars)) .route("/xrpc/sh.tangled.feed.countStars", get(count_stars)) .route("/xrpc/sh.tangled.feed.getStar", get(get_star)) .route("/xrpc/sh.tangled.graph.listFollows", get(list_follows)) .route("/xrpc/sh.tangled.graph.countFollows", get(count_follows)) .route("/xrpc/sh.tangled.graph.getFollow", get(get_follow)) .route("/xrpc/sh.tangled.repo.listIssues", get(list_issues)) .route("/xrpc/sh.tangled.repo.countIssues", get(count_issues)) .route("/xrpc/sh.tangled.repo.listPulls", get(list_pulls)) .route("/xrpc/sh.tangled.repo.countPulls", get(count_pulls)) .route( "/xrpc/sh.tangled.feed.listComments", get(feed::list_comments), ) .route( "/xrpc/sh.tangled.feed.countComments", get(count_feed_comments), ) .route("/xrpc/sh.tangled.feed.listReactions", get(list_reactions)) .route("/xrpc/sh.tangled.feed.countReactions", get(count_reactions)) .route( "/xrpc/sh.tangled.repo.listCollaborators", get(list_collaborators), ) .route( "/xrpc/sh.tangled.repo.countCollaborators", get(count_collaborators), ) .route("/xrpc/sh.tangled.repo.listRepos", get(list_repos)) .route("/xrpc/sh.tangled.repo.countRepos", get(count_repos)) .route("/xrpc/sh.tangled.repo.countForks", get(count_forks)) .route( "/xrpc/sh.tangled.publicKey.getPublicKey", get(get_public_key), ) .route("/xrpc/sh.tangled.publicKey.listKeys", get(list_public_keys)) .route( "/xrpc/sh.tangled.publicKey.countKeys", get(count_public_keys), ) .route("/xrpc/sh.tangled.feed.getTimeline", get(feed::get_timeline)) .route( "/xrpc/sh.tangled.graph.listNetworkVouches", get(list_network_vouches), ) .route("/xrpc/sh.tangled.feed.listStarsBy", get(list_stars_by)) .route("/xrpc/sh.tangled.feed.countStarsBy", get(count_stars_by)) .route( "/xrpc/sh.tangled.feed.listReactionsBy", get(list_reactions_by), ) .route( "/xrpc/sh.tangled.feed.countReactionsBy", get(count_reactions_by), ) .route("/xrpc/sh.tangled.graph.listFollowsBy", get(list_follows_by)) .route( "/xrpc/sh.tangled.graph.countFollowsBy", get(count_follows_by), ) .route( "/xrpc/sh.tangled.repo.listCollaboratorInvitesBy", get(list_collaborator_invites_by), ) .route( "/xrpc/sh.tangled.repo.listArtifactsBy", get(list_artifacts_by), ) .route( "/xrpc/sh.tangled.repo.countArtifactsBy", get(count_artifacts_by), ) .route( "/xrpc/sh.tangled.repo.listCollaboratorsBy", get(list_collaborators_by), ) .route( "/xrpc/sh.tangled.repo.countCollaboratorsBy", get(count_collaborators_by), ) .route( "/xrpc/sh.tangled.feed.listCommentsBy", get(list_feed_comments_by), ) .route( "/xrpc/sh.tangled.feed.countCommentsBy", get(count_feed_comments_by), ) .route( "/xrpc/sh.tangled.label.listDefinitions", get(list_label_definitions), ) .route( "/xrpc/sh.tangled.label.countDefinitions", get(count_label_definitions), ) .route("/xrpc/sh.tangled.repo.listArtifacts", get(list_artifacts)) .route("/xrpc/sh.tangled.repo.countArtifacts", get(count_artifacts)) .route("/xrpc/sh.tangled.string.getString", get(get_string)) .route("/xrpc/sh.tangled.string.listStrings", get(list_strings)) .route("/xrpc/sh.tangled.string.countStrings", get(count_strings)) .route("/xrpc/sh.tangled.search.query", get(search_query)) .route( "/xrpc/org.tangled.temp.search.searchCode", get(codesearch::search_code), ) .route( "/xrpc/sh.tangled.query.enrichResponse", axum::routing::post(enrich::enrich), ) .route( "/xrpc/sh.tangled.actor.searchActorsTypeahead", get(search_actors_typeahead), ) .route("/_ready", get(get_ready)) .route("/xrpc/sh.tangled.bobbin.getCoverage", get(get_coverage)) .route("/xrpc/sh.tangled.bobbin.awaitRecord", get(await_record)) .route( "/xrpc/org.tangled.temp.notification.listRecipients", get(list_recipients), ) .route( "/xrpc/org.tangled.feed.subscription.getForActor", get(get_subscription_for_actor), ) .route( "/xrpc/blue.microcosm.identity.resolveMiniDoc", get(resolve_mini_doc), ) .route( &format!("/xrpc/{}", pocket::GET_PREFERENCES), get(pocket::get_preferences), ) .route( &format!("/xrpc/{}", pocket::PUT_PREFERENCES), axum::routing::post(pocket::put_preferences), ) .merge(mirror_routes()) .layer( TraceLayer::new_for_http() .make_span_with(DefaultMakeSpan::new().level(Level::INFO)) .on_request(()) .on_response(LatencyFreeTrace) .on_failure(LatencyFreeTrace), ); let router = match cors { Some(cors) => router.layer(cors), None => router, }; router.with_state(state)}
#[derive(Clone, Copy, Debug)]struct LatencyFreeTrace;
impl<B> OnResponse<B> for LatencyFreeTrace { fn on_response(self, response: &Response<B>, _latency: Duration, _span: &Span) { tracing::event!( target: "tower_http::trace::on_response", Level::INFO, status = response.status().as_u16(), "request completed", ); }}
impl OnFailure<ServerErrorsFailureClass> for LatencyFreeTrace { fn on_failure(&mut self, error: ServerErrorsFailureClass, _latency: Duration, _span: &Span) { tracing::event!( target: "tower_http::trace::on_failure", Level::WARN, error = %error, "request failed", ); }}
const MIRROR_PROXIED_NSIDS: &[&str] = &[ "sh.tangled.git.temp2.getBlame", "sh.tangled.git.temp2.getBlob", "sh.tangled.git.temp2.getDiff", "sh.tangled.git.temp2.getInterdiff", "sh.tangled.git.temp2.listCommits", "sh.tangled.git.temp2.mergeCheck", "org.tangled.temp.git.getArchive", "org.tangled.temp.git.getBlob", "org.tangled.temp.git.getBranch", "org.tangled.temp.git.getEntry", "org.tangled.temp.git.getMergeBase", "org.tangled.temp.git.getRef", "org.tangled.temp.git.getTag", "org.tangled.temp.git.getTree", "org.tangled.temp.git.listBranches", "org.tangled.temp.git.listTags", // TODO: remove legacy endpoint support "sh.tangled.git.temp.formatPatch", "sh.tangled.git.temp.listLanguages",];
const MIRROR_WRITE_NSIDS: &[&str] = &[ "sh.tangled.git.mergeCommit", "org.tangled.temp.git.mergeCommit", "org.tangled.temp.git.deleteBranch",];
const PASSTHROUGH_HEADERS: &[&HeaderName] = &[ &CONTENT_TYPE, &CONTENT_LENGTH, &CONTENT_ENCODING, &ETAG, &CACHE_CONTROL, &LAST_MODIFIED, &CONTENT_DISPOSITION, &ACCEPT_RANGES, &CONTENT_RANGE, &X_CONTENT_TYPE_OPTIONS, &CONTENT_SECURITY_POLICY,];
const RANGE_OR_CONDITIONAL_HEADERS: &[&HeaderName] = &[&RANGE, &IF_RANGE, &IF_NONE_MATCH, &IF_MODIFIED_SINCE];
type ProxyParams = Vec<(String, String)>;
fn mirror_routes() -> Router<AppState> { let reads = register_proxied(Router::new(), MIRROR_PROXIED_NSIDS, proxy_mirror_handler); register_mirror_writes(reads)}
fn register_mirror_writes(router: Router<AppState>) -> Router<AppState> { MIRROR_WRITE_NSIDS.iter().fold(router, |router, &nsid_lit| { let nsid = nsid_static(nsid_lit); router.route( &format!("/xrpc/{nsid_lit}"), axum::routing::post( move |State(state): State<AppState>, headers: HeaderMap, body: Bytes| { proxy_mirror_write_handler(state, headers, body, nsid.clone()) }, ), ) })}
fn register_proxied<H, Fut>( router: Router<AppState>, nsids: &[&'static str], handler: H,) -> Router<AppState>where H: Fn(AppState, HeaderMap, SocketPeer, ProxyParams, Nsid<DefaultStr>) -> Fut + Clone + Send + Sync + 'static, Fut: Future<Output = Result<Response, XrpcError>> + Send + 'static,{ nsids.iter().fold(router, |router, &nsid_lit| { let handler = handler.clone(); let nsid = nsid_static(nsid_lit); router.route( &format!("/xrpc/{nsid_lit}"), get( move |State(state): State<AppState>, headers: HeaderMap, socket: SocketPeer, Query(params): Query<ProxyParams>| { handler(state, headers, socket, params, nsid.clone()) }, ), ) })}
#[derive(Clone, Debug)]pub enum SubjectQuery { Did(Did<DefaultStr>), Uri(AtUri<DefaultStr>),}
impl<'de> Deserialize<'de> for SubjectQuery { fn deserialize<D>(deserializer: D) -> Result<Self, D::Error> where D: serde::Deserializer<'de>, { let raw = String::deserialize(deserializer)?; if let Ok(did) = Did::<DefaultStr>::new_owned(&raw) { return Ok(Self::Did(did)); } AtUri::<DefaultStr>::new_owned(&raw) .map(Self::Uri) .map_err(serde::de::Error::custom) }}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct ExpectedNsid { canon: Nsid<DefaultStr>,}
impl ExpectedNsid { pub fn new(canon: Nsid<DefaultStr>) -> Self { Self { canon } }
pub fn from_static(s: &'static str) -> Self { Self::new(nsid_static(s)) }
pub fn as_nsid(&self) -> &Nsid<DefaultStr> { &self.canon }
pub fn as_str(&self) -> &str { self.canon.as_ref() }
fn accepts(&self, other: &str) -> bool { other == self.canon.as_ref() }}
#[derive(Debug, Deserialize)]struct GetRepoQuery { repo: AtUri<DefaultStr>,}
#[derive(Debug, Deserialize)]struct GetRepoByRepoDidQuery { #[serde(rename = "repoDid")] repo_did: Did<DefaultStr>,}
#[derive(Debug, Deserialize)]struct GetRepoByNameQuery { owner: Did<DefaultStr>, name: String,}
#[derive(Debug, Deserialize)]struct GetProfileQuery { actor: AtUri<DefaultStr>,}
#[derive(Debug, Deserialize)]struct GetTicketQuery { uri: AtUri<DefaultStr>,}
#[derive(Debug, Deserialize)]struct GetStringQuery { uri: AtUri<DefaultStr>,}
#[derive(Debug, Deserialize)]struct GetPublicKeyQuery { #[serde(rename = "publicKey")] public_key: AtUri<DefaultStr>,}
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)]#[serde(rename_all = "lowercase")]enum Order { Asc, #[default] Desc,}
impl From<Order> for SortDir { fn from(o: Order) -> Self { match o { Order::Asc => SortDir::Asc, Order::Desc => SortDir::Desc, } }}
#[derive(Debug, Deserialize)]struct TypedListQuery<F> { subject: SubjectQuery, cursor: Option<String>, offset: Option<PageOffset>, limit: Option<u32>, #[serde(default)] order: Order, #[serde(flatten)] filter: F,}
impl<F> TypedListQuery<F> { fn dir(&self) -> SortDir { self.order.into() }}
#[derive(Debug, Deserialize)]struct CountQuery { subject: SubjectQuery,}
#[derive(Debug, Deserialize)]struct TypedCountQuery<F> { subject: SubjectQuery, #[serde(flatten)] filter: F,}
#[derive(Debug, Deserialize)]struct GetEdgeQuery { actor: Did<DefaultStr>, subject: SubjectQuery,}
#[derive(Debug, Deserialize)]struct SearchQueryParams { q: String, nsid: Option<Nsid<DefaultStr>>, author: Option<Did<DefaultStr>>, repo: Option<Did<DefaultStr>>, since: Option<String>, until: Option<String>, cursor: Option<String>, offset: Option<PageOffset>, limit: Option<u32>,}
#[derive(Debug, Deserialize)]struct ActorTypeaheadParams { q: String, limit: Option<u32>,}
pub struct XrpcQuery<T>(pub T);
impl<S, T> FromRequestParts<S> for XrpcQuery<T>where S: Send + Sync, T: serde::de::DeserializeOwned + Send + 'static,{ type Rejection = XrpcError;
async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> { Query::<T>::from_request_parts(parts, state) .await .map(|Query(t)| Self(t)) .map_err(|rej: QueryRejection| XrpcError::InvalidParams(rej.body_text())) }}
#[derive(Debug, Error)]pub enum XrpcError { #[error("invalid request: {0}")] InvalidParams(String), #[error("authentication required: {0}")] AuthRequired(String), // /// A service-auth token failed verification. Typed so the jwt-level checks speak jacquard's // /// vocabulary; adopting `jacquard-axum` later replaces the producer, not this variant. // #[error(transparent)] // Auth(#[from] jacquard_common::service_auth::ServiceAuthError), #[error("record not found")] NotFound, #[error("upstream unavailable: {0}")] UpstreamUnavailable(String), #[error("upstream gone: {0}")] UpstreamGone(String), #[error("invalid record: {0}")] InvalidRecord(String), #[error("internal: {0}")] Internal(String), #[error("overloaded, shedding under memory pressure")] Overloaded, #[error("invite index is still warming, so it can't list offers yet")] Warming, #[error("record has not been ingested yet")] NotSettled, #[error("not implemented: {0}")] NotImplemented(&'static str),}
impl XrpcError { pub fn overloaded() -> Self { Self::Overloaded }}
#[derive(Serialize)]struct ErrorBody { error: &'static str, message: String,}
impl IntoResponse for XrpcError { fn into_response(self) -> Response { let (status, error) = match &self { Self::InvalidParams(_) => (StatusCode::BAD_REQUEST, "InvalidRequest"), Self::AuthRequired(_) => (StatusCode::UNAUTHORIZED, "AuthenticationRequired"), Self::NotFound => (StatusCode::NOT_FOUND, "RecordNotFound"), Self::UpstreamUnavailable(_) => (StatusCode::BAD_GATEWAY, "UpstreamFailed"), Self::UpstreamGone(_) => (StatusCode::BAD_GATEWAY, "UpstreamGone"), Self::InvalidRecord(_) => (StatusCode::BAD_GATEWAY, "InvalidRecord"), Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "InternalError"), Self::Overloaded => (StatusCode::SERVICE_UNAVAILABLE, "Overloaded"), Self::Warming => (StatusCode::SERVICE_UNAVAILABLE, "Warming"), Self::NotSettled => (StatusCode::GATEWAY_TIMEOUT, "NotSettled"), Self::NotImplemented(_) => (StatusCode::NOT_IMPLEMENTED, "MethodNotImplemented"), }; let body = ErrorBody { error, message: self.to_string(), }; (status, Json(body)).into_response() }}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct CoverageEnvelope { ready: bool, events_processed: u64, last_cursor: u64, #[serde(skip_serializing_if = "Option::is_none")] stream: Option<StreamEnvelope>,}
impl From<Coverage> for CoverageEnvelope { fn from(c: Coverage) -> Self { Self { ready: c.is_ready(), events_processed: c.events_processed(), last_cursor: c.last_cursor().raw(), stream: None, } }}
/// a wedged ingest shows up as a state that never moved on#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct StreamEnvelope { state: &'static str, state_since: String, last_error: Option<Box<str>>,}
impl From<StreamHealthSnapshot> for StreamEnvelope { fn from(snapshot: StreamHealthSnapshot) -> Self { Self { state: snapshot.state.as_str(), state_since: rfc3339_millis(snapshot.state_since_unix_micros), last_error: snapshot.last_error, } }}
fn rfc3339_millis(stamp: UnixMicros) -> String { i64::try_from(stamp.raw()) .ok() .and_then(chrono::DateTime::<chrono::Utc>::from_timestamp_micros) .map(|at| at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)) .unwrap_or_default()}
struct Deduped<T>(T);
impl<T: Serialize> Serialize for Deduped<T> { fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> where S: serde::Serializer, { serde_json::to_value(&self.0) .map_err(serde::ser::Error::custom)? .serialize(serializer) }}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct RecordView<V> { uri: AtUri<DefaultStr>, #[serde(skip_serializing_if = "Option::is_none")] cid: Option<Cid<DefaultStr>>, value: V,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct NetworkVouchItem<V> { actor: Did<DefaultStr>, #[serde(flatten)] view: RecordView<V>,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct CollaboratorListItem { subject: Did<DefaultStr>, added_by: Did<DefaultStr>, created_at: Datetime, effective_since: Datetime, #[serde(skip_serializing_if = "Option::is_none")] uri: Option<AtUri<DefaultStr>>, #[serde(skip_serializing_if = "Option::is_none")] cid: Option<Cid<DefaultStr>>,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct StatefulItem<V> { #[serde(flatten)] view: RecordView<V>, state: &'static str, comment_count: u64,}
pub(crate) fn source_authority_did(source: &AtUri<DefaultStr>) -> Option<Did<DefaultStr>> { match source.authority() { AtIdentifier::Did(d) => Some(d.clone().into_static()), AtIdentifier::Handle(_) => None, }}
fn enrich_ticket_view( state: &AppState, view: RecordView<Ticket<DefaultStr>>,) -> StatefulItem<Ticket<DefaultStr>> { let comment_count = state.edges.count(&EdgeKey::new( nsid_static("org.tangled.feed.comment"), SubjectRef::Uri(view.uri.clone()), )); let kind = TicketStateKind::from_ticket(view.value.state.as_ref()).unwrap_or_default(); StatefulItem { view, state: kind.wire(), comment_count, }}
#[derive(Clone, Copy, Debug, Serialize)]#[serde(rename_all = "camelCase")]struct CountResponse { count: u64, distinct_authors: u64,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct EdgeUriResponse { uri: AtUri<DefaultStr>,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct SearchHitView { uri: AtUri<DefaultStr>, cid: Option<Cid<DefaultStr>>, nsid: Nsid<DefaultStr>, score: f32, value: SearchableRecord,}
#[derive(Serialize)]struct ActorTypeaheadResponse { actors: Vec<ActorTypeaheadView>,}
#[derive(Serialize)]struct ActorTypeaheadView { uri: AtUri<DefaultStr>, did: Did<DefaultStr>, handle: String,}
fn map_slingshot(err: SlingshotError) -> XrpcError { use SlingshotError as E; match err { E::NotFound => XrpcError::NotFound, e @ (E::Decode(_) | E::MissingField(_) | E::InvalidAtUri(_) | E::InvalidCid(_) | E::UriMismatch { .. }) => XrpcError::InvalidRecord(e.to_string()), e @ (E::Network(_) | E::Build(_) | E::Upstream(_) | E::BodyTooLarge { .. } | E::BadScheme(_)) => XrpcError::UpstreamUnavailable(e.to_string()), }}
fn parse_uri(raw: &str) -> Result<AtUri<DefaultStr>, XrpcError> { AtUri::<DefaultStr>::new_owned(raw).map_err(|e| XrpcError::InvalidParams(format!("uri: {e}")))}
pub enum SubjectShape { BareDid, Collection(&'static str), BareDidOrOneOfCollections(&'static [&'static str]), OneOfCollections(&'static [&'static str]), AnyAtUri, BareDidOrAnyAtUri,}
pub trait HasSubject { const SHAPE: SubjectShape;}
pub trait MirrorOf { type Record: XrpcResp; const EDGE_KIND: &'static str; const SHAPE: SubjectShape;}
macro_rules! edge_kinds { ($($collection:literal => $record:ty, $shape:expr $(, mirror $by:ident)? ;)*) => { $( impl HasSubject for $record { const SHAPE: SubjectShape = $shape; } )* $($( pub struct $by; impl MirrorOf for $by { type Record = $record; const EDGE_KIND: &'static str = concat!($collection, ".by"); const SHAPE: SubjectShape = SubjectShape::BareDid; } )?)*
pub(crate) fn subject_shape(collection: &str) -> Option<(&'static str, SubjectShape)> { Some(match collection { $($collection => ($collection, <$record as HasSubject>::SHAPE),)* TICKET_ISSUE_KIND => (TICKET_ISSUE_KIND, <TicketRecord as HasSubject>::SHAPE), TICKET_PULL_KIND => (TICKET_PULL_KIND, <TicketRecord as HasSubject>::SHAPE), _ => return None, }) }
pub(crate) fn mirror_kind(collection: &str) -> Option<&'static str> { Some(match collection { $($($collection => <$by as MirrorOf>::EDGE_KIND,)?)* _ => return None, }) } };}
edge_kinds! { "org.tangled.feed.star" => StarRecord, SubjectShape::BareDidOrOneOfCollections(&["org.tangled.snippet.string"]), mirror StarBy; "org.tangled.feed.comment" => FeedCommentRecord, SubjectShape::OneOfCollections(&["org.tangled.track.ticket", "org.tangled.review.patch", "org.tangled.snippet.string"]), mirror FeedCommentBy; "org.tangled.feed.reaction" => ReactionRecord, SubjectShape::AnyAtUri, mirror ReactionBy; "org.tangled.graph.follow" => FollowRecord, SubjectShape::BareDid, mirror FollowBy; "org.tangled.graph.vouch" => VouchRecord, SubjectShape::BareDid, mirror VouchBy; "org.tangled.label.definition" => LabelDefinitionRecord, SubjectShape::BareDid; "org.tangled.key.sshKey" => SshKeyRecord, SubjectShape::BareDid; "org.tangled.repo.manifest" => ManifestRecord, SubjectShape::BareDid; "sh.tangled.repo.artifact" => ArtifactRecord, SubjectShape::BareDid, mirror ArtifactBy; "org.tangled.graph.collaborator" => CollaboratorRecord, SubjectShape::BareDid, mirror CollaboratorBy; "org.tangled.snippet.string" => SnippetStringRecord, SubjectShape::BareDid; "org.tangled.feed.subscription" => SubscriptionRecord, SubjectShape::BareDidOrAnyAtUri; "org.tangled.review.patch" => PatchRecord, SubjectShape::Collection("org.tangled.track.ticket");}
// tickets index under two kinds split by `targetBranch`, so they sit outside `edge_kinds!`impl HasSubject for TicketRecord { const SHAPE: SubjectShape = SubjectShape::BareDid;}
const FORK_SUBJECT_SHAPE: SubjectShape = SubjectShape::BareDid;
fn parse_subject(raw: &SubjectQuery, shape: SubjectShape) -> Result<SubjectRef, XrpcError> { let uri = match raw { SubjectQuery::Did(did) => { return match shape { SubjectShape::BareDid | SubjectShape::BareDidOrOneOfCollections(_) | SubjectShape::BareDidOrAnyAtUri => Ok(SubjectRef::Did(did.clone())), SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!( "subject must be at://<did>/{expected}/<rkey>, got bare did" ))), SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!( "subject must be at://<did>/<nsid>/<rkey> with nsid in [{}], got bare did", allowed.join(", "), ))), SubjectShape::AnyAtUri => Err(XrpcError::InvalidParams( "subject must be at-uri form, got bare did".into(), )), }; } SubjectQuery::Uri(uri) => uri, }; if matches!(uri.authority(), AtIdentifier::Handle(_)) { return Err(XrpcError::InvalidParams( "subject authority must be a did, not a handle".into(), )); } let Some(collection) = uri.collection() else { return Err(XrpcError::InvalidParams( "subject must be a bare did or full at://<did>/<nsid>/<rkey>".into(), )); }; let c = collection.as_ref(); match shape { SubjectShape::BareDid => Err(XrpcError::InvalidParams(format!( "subject must be a bare did, got at-uri with collection {c}" ))), SubjectShape::Collection(expected) if c == expected => { require_rkey(uri, expected)?; Ok(SubjectRef::Uri(uri.clone())) } SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!( "subject must be at://<did>/{expected}/<rkey>, got collection {c}" ))), SubjectShape::OneOfCollections(allowed) if allowed.contains(&c) => { require_rkey(uri, c)?; Ok(SubjectRef::Uri(uri.clone())) } SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!( "subject must be at://<did>/<nsid>/<rkey> with nsid in [{}], got collection {c}", allowed.join(", "), ))), SubjectShape::BareDidOrOneOfCollections(allowed) if allowed.contains(&c) => { require_rkey(uri, c)?; Ok(SubjectRef::Uri(uri.clone())) } SubjectShape::BareDidOrOneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!( "subject must be a bare did or at://<did>/<nsid>/<rkey> with nsid in [{}], got collection {c}", allowed.join(", "), ))), SubjectShape::AnyAtUri | SubjectShape::BareDidOrAnyAtUri => { Ok(SubjectRef::Uri(uri.clone())) } }}
fn require_rkey(uri: &AtUri<DefaultStr>, expected: &str) -> Result<(), XrpcError> { uri.rkey().map(|_| ()).ok_or_else(|| { XrpcError::InvalidParams(format!( "subject must be at://<did>/{expected}/<rkey>; missing rkey" )) })}
pub(crate) fn parse_cursor(raw: Option<&str>) -> Result<PageCursor, XrpcError> { PageCursor::from_token(raw) .map_err(|e: CursorParseError| XrpcError::InvalidParams(format!("cursor: {e}")))}
// cursor and offset are mutually exclusive, offset pages return a cursor for follow-uppub(crate) fn parse_start( cursor: Option<&str>, offset: Option<PageOffset>,) -> Result<PageStart, XrpcError> { match (cursor, offset) { (Some(_), Some(_)) => Err(XrpcError::InvalidParams( "cursor and offset are mutually exclusive".into(), )), (Some(c), None) => Ok(PageStart::Cursor(parse_cursor(Some(c))?)), (None, Some(o)) => Ok(PageStart::Offset(o)), (None, None) => Ok(PageStart::Cursor(PageCursor::Start)), }}
pub(crate) fn parse_limit(raw: Option<u32>) -> Result<PageLimit, XrpcError> { PageLimit::new(raw.unwrap_or(DEFAULT_LIMIT)) .map_err(|e| XrpcError::InvalidParams(format!("limit: {e}")))}
async fn resolve_for_view( state: &AppState, expected_nsid: &Nsid<DefaultStr>, uri: AtUri<DefaultStr>,) -> Result<Arc<RecordBody>, XrpcError> { let raw = uri.as_ref().to_owned(); resolve(state, ExpectedNsid::new(expected_nsid.clone()), uri) .await .map(|(body, _did)| body) .map_err(|e| match e { XrpcError::NotFound => XrpcError::UpstreamGone(raw), other => other, })}
async fn resolve( state: &AppState, expected: ExpectedNsid, uri: AtUri<DefaultStr>,) -> Result<(Arc<RecordBody>, Did<DefaultStr>), XrpcError> { let collection = uri .collection() .ok_or_else(|| XrpcError::InvalidParams("uri missing collection".into()))?; if !expected.accepts(collection.as_ref()) { return Err(XrpcError::InvalidParams(format!( "collection mismatch: expected {}, got {}", expected.as_str(), collection.as_ref() ))); } let rkey = uri .rkey() .ok_or_else(|| XrpcError::InvalidParams("uri missing rkey".into()))?; let did_ref = match uri.authority() { AtIdentifier::Did(d) => d, AtIdentifier::Handle(_) => { return Err(XrpcError::InvalidParams( "uri authority must be a did, not a handle".into(), )); } }; let did: Did<DefaultStr> = did_ref.clone().into_static();
// unclaimed repos don't exist as far as xrpc is concerned if expected.as_str() == ManifestRecord::NSID && !state.claims.is_claimed(&uri) { return Err(XrpcError::NotFound); } if let Some(hit) = state.records.get(&uri) { return Ok((hit, did)); } let body = state .slingshot .get_record(&did_ref, &collection, &rkey) .await .map_err(map_slingshot)?; verify_type_tag(&body, &expected)?; state.records.put(uri, body.clone()); Ok((body, did))}
#[derive(Deserialize)]struct TypeTag<'a> { #[serde(rename = "$type", borrow)] ty: &'a str,}
fn verify_type_tag(body: &RecordBody, expected: &ExpectedNsid) -> Result<(), XrpcError> { let bytes = body.value.as_ref(); let ty: std::borrow::Cow<'_, str> = match serde_json::from_slice::<TypeTag>(bytes) { Ok(t) => std::borrow::Cow::Borrowed(t.ty), Err(_) => { let value: serde_json::Value = serde_json::from_slice(bytes) .map_err(|e| XrpcError::InvalidRecord(format!("$type peek: {e}")))?; value .as_object() .and_then(|m| m.get("$type")) .and_then(|v| v.as_str()) .map(|s| std::borrow::Cow::Owned(s.to_owned())) .ok_or_else(|| XrpcError::InvalidRecord("$type peek: missing $type field".into()))? } }; if !expected.accepts(ty.as_ref()) { return Err(XrpcError::InvalidRecord(format!( "$type mismatch: expected {}, got {}", expected.as_str(), ty ))); } Ok(())}
fn deserialize_record<V>(bytes: &[u8]) -> Result<V, XrpcError>where V: serde::de::DeserializeOwned,{ serde_json::from_slice::<V>(bytes).or_else(|canon_err| { normalize_record_fields(bytes) .and_then(|normalized| serde_json::from_slice::<V>(&normalized).ok()) .ok_or_else(|| XrpcError::InvalidRecord(canon_err.to_string())) })}
async fn fetch_from_uri<R, V>( state: &AppState, uri: AtUri<DefaultStr>,) -> Result<(Arc<RecordBody>, V), XrpcError>where R: XrpcResp, V: serde::de::DeserializeOwned + NormalizeRepoRefs,{ let raw = uri.as_str().to_owned(); let nsid = nsid_static(R::NSID); let (body, _did) = resolve(state, ExpectedNsid::new(nsid.clone()), uri).await?; let value: V = deserialize_record(&body.value)?; let value = value .normalize(&state.resolver) .await .ok_or(XrpcError::UpstreamGone(raw))?; Ok((body, value))}
pub(crate) async fn fetch<R, V>( state: &AppState, uri: &AtUri<DefaultStr>,) -> Result<(Arc<RecordBody>, V), XrpcError>where R: XrpcResp, V: serde::de::DeserializeOwned + NormalizeRepoRefs,{ fetch_from_uri::<R, V>(state, uri.clone()).await}
async fn get_repo( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetRepoQuery>,) -> Result<Json<Deduped<RepoOutput>>, XrpcError> { get_manifest(&state, q.repo).await}
async fn get_repo_by_repo_did( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetRepoByRepoDidQuery>,) -> Result<Json<Deduped<RepoOutput>>, XrpcError> { get_manifest(&state, view::manifest_uri(&q.repo_did)).await}
async fn get_repo_by_name( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetRepoByNameQuery>,) -> Result<Json<Deduped<RepoOutput>>, XrpcError> { let repo_did = state .claims .repo_by_name(q.owner.as_str(), &q.name) .and_then(|r| Did::<DefaultStr>::new_owned(r).ok()) .ok_or(XrpcError::NotFound)?; get_manifest(&state, view::manifest_uri(&repo_did)).await}
#[derive(Serialize)]struct RepoOutput { #[serde(flatten)] record: ManifestGetRecordOutput<DefaultStr>, #[serde(skip_serializing_if = "Option::is_none")] knot: Option<String>, #[serde(rename = "defaultBranch", skip_serializing_if = "Option::is_none")] default_branch: Option<String>,}
async fn get_manifest( state: &AppState, uri: AtUri<DefaultStr>,) -> Result<Json<Deduped<RepoOutput>>, XrpcError> { let repo_did = owner_did_from_aturi(&uri); let (fetched, knot, default_branch) = tokio::join!( fetch_from_uri::<ManifestRecord, Manifest<DefaultStr>>(state, uri), async { match &repo_did { Some(did) => state.identity.repo_knot(did).await, None => None, } }, async { let did = repo_did.as_ref()?; let refs = view::refs_uri(did); let (_, value) = fetch_from_uri::<RefsRecord, Refs<DefaultStr>>(state, refs) .await .ok()?; Some(value.default_branch.to_string()) } ); let (body, value) = fetched?; Ok(Json(Deduped(RepoOutput { record: ManifestGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, }, knot, default_branch, })))}
async fn get_profile( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetProfileQuery>,) -> Result<Json<Deduped<ProfileGetRecordOutput<DefaultStr>>>, XrpcError> { let (body, value) = fetch::<ProfileRecord, Profile<DefaultStr>>(&state, &q.actor).await?; Ok(Json(Deduped(ProfileGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, })))}
async fn get_ticket( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetTicketQuery>,) -> Result<Json<Deduped<TicketGetRecordOutput<DefaultStr>>>, XrpcError> { let (body, value) = fetch::<TicketRecord, Ticket<DefaultStr>>(&state, &q.uri).await?; Ok(Json(Deduped(TicketGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, })))}
async fn get_string( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetStringQuery>,) -> Result<Json<Deduped<SnippetStringGetRecordOutput<DefaultStr>>>, XrpcError> { let (body, value) = fetch::<SnippetStringRecord, SnippetString<DefaultStr>>(&state, &q.uri).await?; Ok(Json(Deduped(SnippetStringGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, })))}
async fn get_public_key( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetPublicKeyQuery>,) -> Result<Json<Deduped<SshKeyGetRecordOutput<DefaultStr>>>, XrpcError> { let (body, value) = fetch::<SshKeyRecord, SshKey<DefaultStr>>(&state, &q.public_key).await?; Ok(Json(Deduped(SshKeyGetRecordOutput { cid: Some(body.cid.clone()), uri: body.uri.clone(), value, })))}
async fn get_repos( State(state): State<AppState>, RawQuery(query): RawQuery,) -> Result<Response, XrpcError> { let uris = collect_repeated(query.as_deref(), BULK_REPOS_KEY); bulk_fetch::<ManifestRecord, Manifest<DefaultStr>>(&state, uris).await}
async fn get_repos_by_repo_dids( State(state): State<AppState>, RawQuery(query): RawQuery,) -> Result<Response, XrpcError> { let dids = collect_repeated(query.as_deref(), BULK_REPO_DIDS_KEY); if dids.is_empty() { return Err(XrpcError::InvalidParams("at least one did required".into())); } if dids.len() > BULK_LIMIT { return Err(XrpcError::InvalidParams(format!( "at most {BULK_LIMIT} dids per request" ))); } let dids = dids .iter() .map(|s| { Did::<DefaultStr>::new_owned(s) .map_err(|_| XrpcError::InvalidParams(format!("invalid did: {s}"))) }) .collect::<Result<Vec<_>, _>>()?; let uris = dids.iter().map(view::manifest_uri).collect(); bulk_stream::<ManifestRecord, Manifest<DefaultStr>>(&state, uris)}
async fn get_profiles( State(state): State<AppState>, RawQuery(query): RawQuery,) -> Result<Response, XrpcError> { let uris = collect_repeated(query.as_deref(), BULK_PROFILES_KEY); bulk_fetch::<ProfileRecord, Profile<DefaultStr>>(&state, uris).await}
async fn get_tickets( State(state): State<AppState>, RawQuery(query): RawQuery,) -> Result<Response, XrpcError> { let uris = collect_repeated(query.as_deref(), BULK_TICKETS_KEY); bulk_fetch::<TicketRecord, Ticket<DefaultStr>>(&state, uris).await}
const BULK_REPOS_KEY: &str = "repos";const BULK_REPO_DIDS_KEY: &str = "dids";const BULK_PROFILES_KEY: &str = "actors";const BULK_TICKETS_KEY: &str = "uris";const BULK_LIMIT: usize = 50;
fn collect_repeated(query: Option<&str>, key: &str) -> Vec<String> { let Some(q) = query else { return Vec::new(); }; form_urlencoded::parse(q.as_bytes()) .filter_map(|(k, v)| (k == key).then(|| v.into_owned())) .collect()}
async fn hydrate_record_view<V>( state: &AppState, nsid: &Nsid<DefaultStr>, uri: AtUri<DefaultStr>,) -> Result<Option<RecordView<V>>, XrpcError>where V: serde::de::DeserializeOwned + NormalizeRepoRefs,{ let body = resolve_for_view(state, nsid, uri).await?; let value: V = deserialize_record::<V>(&body.value)?; let Some(value) = value.normalize(&state.resolver).await else { return Ok(None); }; Ok(Some(RecordView { uri: body.uri.clone(), cid: Some(body.cid.clone()), value, }))}
fn datetime_of(at: UnixMicros) -> Option<Datetime> { let micros = i64::try_from(at.raw()).ok()?; chrono::DateTime::from_timestamp_micros(micros).map(|dt| Datetime::new(dt.fixed_offset()))}
#[derive(Clone, Copy)]enum HitProvenance { ClientSupplied, Indexed,}
fn is_index_evictable(err: &XrpcError) -> bool { matches!( err, XrpcError::NotFound | XrpcError::UpstreamGone(_) | XrpcError::InvalidRecord(_) | XrpcError::InvalidParams(_) )}
fn drop_unhydratable<V>( provenance: HitProvenance, nsid: &Nsid<DefaultStr>, uri: &AtUri<DefaultStr>, result: Result<Option<V>, XrpcError>,) -> Result<Option<V>, XrpcError> { match result { Ok(view) => Ok(view), Err(err @ (XrpcError::NotFound | XrpcError::UpstreamGone(_))) => { tracing::debug!( uri = %uri, nsid = %nsid.as_ref(), error = %err, "dropping gone hit during hydration", ); Ok(None) } Err(err @ XrpcError::UpstreamUnavailable(_)) => { tracing::warn!( uri = %uri, nsid = %nsid.as_ref(), error = %err, "dropping hit, upstream unavailable during hydration", ); Ok(None) } Err(err @ XrpcError::InvalidRecord(_)) => { tracing::warn!( uri = %uri, nsid = %nsid.as_ref(), error = %err, "dropping invalid hit during hydration", ); Ok(None) } Err(err @ XrpcError::InvalidParams(_)) => match provenance { HitProvenance::ClientSupplied => Err(err), HitProvenance::Indexed => { tracing::warn!( uri = %uri, nsid = %nsid.as_ref(), error = %err, "dropping malformed indexed hit during hydration", ); Ok(None) } }, Err( err @ (XrpcError::Internal(_) | XrpcError::Overloaded | XrpcError::Warming | XrpcError::NotSettled | XrpcError::AuthRequired(_) | XrpcError::NotImplemented(_)), ) => Err(err), }}
fn hydrate_stream<T, Fut, V>( items: impl IntoIterator<Item = T>, produce: impl FnMut(T) -> Fut,) -> impl Stream<Item = Result<V, XrpcError>>where Fut: Future<Output = Result<Option<V>, XrpcError>>,{ stream::iter(items) .map(produce) .buffered(FETCH_CONCURRENCY) .try_filter_map(|view| async move { Ok(view) })}
fn hydrate_keyed_record_stream<K, V>( state: &AppState, nsid: Nsid<DefaultStr>, items: Vec<(K, EdgeItem)>, provenance: HitProvenance,) -> impl Stream<Item = Result<(K, RecordView<V>), XrpcError>> + Send + 'staticwhere K: Send + 'static, V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,{ let owned = state.clone(); hydrate_stream(items, move |(key, item)| { let owned = owned.clone(); let nsid = nsid.clone(); async move { let EdgeItem { uri, .. } = item; let result = hydrate_record_view::<V>(&owned, &nsid, uri.clone()).await; if matches!(provenance, HitProvenance::Indexed) && let Err(err) = &result && is_index_evictable(err) { owned.edges.remove_source(&uri); } drop_unhydratable(provenance, &nsid, &uri, result) .map(|view| view.map(|view| (key, view))) } })}
fn hydrate_record_stream<V>( state: &AppState, nsid: Nsid<DefaultStr>, items: Vec<EdgeItem>, provenance: HitProvenance,) -> impl Stream<Item = Result<RecordView<V>, XrpcError>> + Send + 'staticwhere V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,{ hydrate_keyed_record_stream( state, nsid, items.into_iter().map(|item| ((), item)).collect(), provenance, ) .map_ok(|(_, view)| view)}
enum PagePhase { Head, Body { first: bool }, Done,}
struct PageState<S> { items: std::pin::Pin<Box<S>>, phase: PagePhase, array_key: &'static str, tail: Vec<u8>, permit: Option<HeavyPermit>,}
pub(crate) fn paged_tail(cursor: Option<String>, total: Option<TotalCount>) -> Vec<u8> { let encoded = cursor.map(|value| serde_json::to_string(&value).unwrap_or_else(|_| "\"\"".to_owned())); match (encoded, total) { (Some(cursor), Some(total)) => { format!("],\"cursor\":{cursor},\"total\":{}}}", total.get()).into_bytes() } (Some(cursor), None) => format!("],\"cursor\":{cursor}}}").into_bytes(), (None, Some(total)) => format!("],\"total\":{}}}", total.get()).into_bytes(), (None, None) => unpaged_tail(), }}
fn unpaged_tail() -> Vec<u8> { b"]}".to_vec()}
pub(crate) fn json_stream<V, S>( array_key: &'static str, items: S, tail: Vec<u8>, permit: Option<HeavyPermit>,) -> Responsewhere V: Serialize + Send + 'static, S: Stream<Item = Result<V, XrpcError>> + Send + 'static,{ let init = PageState { items: Box::pin(items), phase: PagePhase::Head, array_key, tail, permit, }; let chunks = stream::unfold(init, |mut st| async move { match st.phase { PagePhase::Head => { let head = format!("{{\"{}\":[", st.array_key).into_bytes(); st.phase = PagePhase::Body { first: true }; Some((Ok::<Vec<u8>, Infallible>(head), st)) } PagePhase::Body { first } => match st.items.next().await { Some(Ok(view)) => match serde_json::to_vec(&Deduped(&view)) { Ok(encoded) => { let mut chunk = Vec::with_capacity(encoded.len() + 1); if !first { chunk.push(b','); } chunk.extend_from_slice(&encoded); st.phase = PagePhase::Body { first: false }; Some((Ok(chunk), st)) } Err(e) => { tracing::warn!(error = %e, "skipping hit, serialize failed mid-stream"); Some((Ok(Vec::new()), st)) } }, Some(Err(e)) => { tracing::warn!(error = %e, "ending page early, hydration failed mid-stream"); let tail = std::mem::take(&mut st.tail); st.phase = PagePhase::Done; Some((Ok(tail), st)) } None => { let tail = std::mem::take(&mut st.tail); st.phase = PagePhase::Done; Some((Ok(tail), st)) } }, PagePhase::Done => { drop(st.permit.take()); None } } }); ( [(CONTENT_TYPE, "application/json")], Body::from_stream(chunks), ) .into_response()}
async fn bulk_fetch<R, V>(state: &AppState, uris: Vec<String>) -> Result<Response, XrpcError>where R: XrpcResp, V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,{ if uris.is_empty() { return Err(XrpcError::InvalidParams("at least one uri required".into())); } if uris.len() > BULK_LIMIT { return Err(XrpcError::InvalidParams(format!( "at most {BULK_LIMIT} uris per request" ))); } let parsed: Vec<AtUri<DefaultStr>> = uris .iter() .map(|s| parse_uri(s)) .collect::<Result<_, _>>()?; let nsid = nsid_static(R::NSID); if let Some(bad) = parsed .iter() .find(|uri| uri.collection().is_none_or(|c| c.as_ref() != nsid.as_ref())) { return Err(XrpcError::InvalidParams(format!( "uri collection must be {}, got {}", nsid.as_ref(), bad.as_ref() ))); } bulk_stream::<R, V>(state, parsed)}
fn bulk_stream<R, V>( state: &AppState, parsed: Vec<AtUri<DefaultStr>>,) -> Result<Response, XrpcError>where R: XrpcResp, V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,{ let nsid = nsid_static(R::NSID); let permit = state.heavy_permit()?; let items = parsed .into_iter() .map(|uri| EdgeItem { uri, sort_micros: 0, sort_source: 0, }) .collect(); let views = hydrate_record_stream::<V>(state, nsid, items, HitProvenance::ClientSupplied); Ok(json_stream::<RecordView<V>, _>( "items", views, unpaged_tail(), permit, ))}
fn record_edge_page<R, F>( state: &AppState, q: &TypedListQuery<F>,) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>where R: XrpcResp + HasSubject, F: ListFilter,{ record_edge_page_of::<R, F>(state, q, R::NSID)}
/// `kind` is the edge kind to page, hydrated as `R`fn record_edge_page_of<R, F>( state: &AppState, q: &TypedListQuery<F>, kind: &'static str,) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>where R: XrpcResp + HasSubject, F: ListFilter,{ let subject = parse_subject(&q.subject, R::SHAPE)?; let start = parse_start(q.cursor.as_deref(), q.offset)?; let limit = parse_limit(q.limit)?; let dir = q.dir(); let key = EdgeKey::new(nsid_static(kind), subject); let nsid = nsid_static(R::NSID); let page = match (q.filter.is_identity(), q.filter.state_view()) { (true, _) => state.edges.list(&key, start, limit, dir), (_, Some(StateViewQuery { kind })) => { state.edges.page_ticket_state(&key, kind, start, limit, dir) } (_, None) => { return Err(XrpcError::Internal( "non-identity filter did not resolve to a state view".into(), )); } }; Ok((page, nsid))}
fn record_list_response<V>( state: &AppState, page: EdgePage, nsid: Nsid<DefaultStr>,) -> Result<Response, XrpcError>where V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,{ let permit = state.heavy_permit()?; let views = hydrate_record_stream::<V>(state, nsid, page.items, HitProvenance::Indexed); Ok(json_stream::<RecordView<V>, _>( "items", views, paged_tail(page.next.map(PageToken::encode_token), page.total), permit, ))}
fn collaborator_list_item( view: RecordView<Collaborator<DefaultStr>>,) -> Result<Option<CollaboratorListItem>, XrpcError> { let RecordView { uri, cid, value } = view; // the repo owns the record and keys it on the collaborator's did let Some(added_by) = source_authority_did(&uri) else { return Ok(None); }; let Some(subject) = uri .rkey() .and_then(|rkey| Did::new_owned(rkey.as_ref()).ok()) else { return Ok(None); }; Ok(Some(CollaboratorListItem { subject, added_by, created_at: value.created_at.clone(), effective_since: value.created_at, uri: Some(uri), cid, }))}
fn collaborator_list_response( state: &AppState, page: EdgePage, nsid: Nsid<DefaultStr>,) -> Result<Response, XrpcError> { let permit = state.heavy_permit()?; let views = hydrate_record_stream::<Collaborator<DefaultStr>>( state, nsid, page.items, HitProvenance::Indexed, ) .try_filter_map(|view| async move { collaborator_list_item(view) }); Ok(json_stream::<CollaboratorListItem, _>( "items", views, paged_tail(page.next.map(PageToken::encode_token), page.total), permit, ))}
async fn list_records<R, V, F>( state: &AppState, q: TypedListQuery<F>,) -> Result<Response, XrpcError>where R: XrpcResp + HasSubject, V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static, F: ListFilter,{ let (page, nsid) = record_edge_page::<R, F>(state, &q)?; record_list_response::<V>(state, page, nsid)}
async fn list_records_by_distinct_author<R, V>( state: &AppState, q: TypedListQuery<NoFilter>,) -> Result<Response, XrpcError>where R: XrpcResp + HasSubject, V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,{ let subject = parse_subject(&q.subject, R::SHAPE)?; let start = parse_start(q.cursor.as_deref(), q.offset)?; let limit = parse_limit(q.limit)?; let nsid = nsid_static(R::NSID); let key = EdgeKey::new(nsid.clone(), subject); let page = state .edges .list_distinct_authors(&key, start, limit, q.dir()); record_list_response::<V>(state, page, nsid)}
fn count_for<R: XrpcResp + HasSubject>( state: &AppState, q: CountQuery,) -> Result<CountResponse, XrpcError> { let subject = parse_subject(&q.subject, R::SHAPE)?; let key = EdgeKey::new(nsid_static(R::NSID), subject); Ok(CountResponse { count: state.edges.count(&key), distinct_authors: state.edges.count_distinct_authors(&key), })}
fn count_typed_of<R, F>( state: &AppState, q: TypedCountQuery<F>, kind: &'static str,) -> Result<CountResponse, XrpcError>where R: XrpcResp + HasSubject, F: CountFilter,{ let subject = parse_subject(&q.subject, R::SHAPE)?; let key = EdgeKey::new(nsid_static(kind), subject); let counted = q.filter.count(state, &key); Ok(CountResponse { count: counted.count.get(), distinct_authors: counted.distinct_authors.get(), })}
// does `actor` have an edge of this kind pointing at `subject`, returns its own urifn get_for<R: XrpcResp + HasSubject>( state: &AppState, q: GetEdgeQuery,) -> Result<EdgeUriResponse, XrpcError> { let subject = parse_subject(&q.subject, R::SHAPE)?; let key = EdgeKey::new(nsid_static(R::NSID), subject); let uri = state .edges .viewer_source(&key, q.actor.as_str()) .ok_or(XrpcError::NotFound)?; Ok(EdgeUriResponse { uri })}
fn mirror_edge_page<M, F>( state: &AppState, q: &TypedListQuery<F>,) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>where M: MirrorOf, F: ListFilter,{ let subject = parse_subject(&q.subject, M::SHAPE)?; let start = parse_start(q.cursor.as_deref(), q.offset)?; let limit = parse_limit(q.limit)?; let dir = q.dir(); let edge_nsid = nsid_static(M::EDGE_KIND); let page = match (q.filter.is_identity(), q.filter.state_view()) { (true, _) => { let key = EdgeKey::new(edge_nsid, subject); state.edges.list(&key, start, limit, dir) } (_, Some(StateViewQuery { kind })) => { let key = EdgeKey::new(edge_nsid, subject); state.edges.page_ticket_state(&key, kind, start, limit, dir) } (_, None) => { return Err(XrpcError::Internal( "non-identity filter did not resolve to a state view".into(), )); } }; let record_nsid = nsid_static(<M::Record as XrpcResp>::NSID); Ok((page, record_nsid))}
async fn list_mirror<M, V, F>(state: &AppState, q: TypedListQuery<F>) -> Result<Response, XrpcError>where M: MirrorOf, V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static, F: ListFilter,{ let (page, record_nsid) = mirror_edge_page::<M, F>(state, &q)?; let permit = state.heavy_permit()?; let views = hydrate_record_stream::<V>(state, record_nsid, page.items, HitProvenance::Indexed); Ok(json_stream::<RecordView<V>, _>( "items", views, paged_tail(page.next.map(PageToken::encode_token), page.total), permit, ))}
fn count_mirror<M: MirrorOf>(state: &AppState, q: CountQuery) -> Result<CountResponse, XrpcError> { let subject = parse_subject(&q.subject, M::SHAPE)?; let key = EdgeKey::new(nsid_static(M::EDGE_KIND), subject); Ok(CountResponse { count: state.edges.count(&key), distinct_authors: state.edges.count_distinct_authors(&key), })}
async fn list_stars( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records_by_distinct_author::<StarRecord, Star<DefaultStr>>(&state, q).await}
async fn count_stars( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<StarRecord>(&state, q).map(Json)}
async fn get_star( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetEdgeQuery>,) -> Result<Json<EdgeUriResponse>, XrpcError> { get_for::<StarRecord>(&state, q).map(Json)}async fn get_subscription_for_actor( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetEdgeQuery>,) -> Result<Json<EdgeUriResponse>, XrpcError> { get_for::<SubscriptionRecord>(&state, q).map(Json)}
async fn list_follows( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records_by_distinct_author::<FollowRecord, Follow<DefaultStr>>(&state, q).await}
async fn count_follows( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<FollowRecord>(&state, q).map(Json)}
async fn get_follow( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<GetEdgeQuery>,) -> Result<Json<EdgeUriResponse>, XrpcError> { get_for::<FollowRecord>(&state, q).map(Json)}
async fn list_issues( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<IssueFilter>>,) -> Result<Response, XrpcError> { list_tickets(&state, &q, TICKET_ISSUE_KIND)}
async fn count_issues( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedCountQuery<IssueFilter>>,) -> Result<Json<CountResponse>, XrpcError> { count_typed_of::<TicketRecord, _>(&state, q, TICKET_ISSUE_KIND).map(Json)}
async fn list_pulls( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<PullFilter>>,) -> Result<Response, XrpcError> { list_tickets(&state, &q, TICKET_PULL_KIND)}
async fn count_pulls( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedCountQuery<PullFilter>>,) -> Result<Json<CountResponse>, XrpcError> { count_typed_of::<TicketRecord, _>(&state, q, TICKET_PULL_KIND).map(Json)}
fn list_tickets<F: ListFilter>( state: &AppState, q: &TypedListQuery<F>, kind: &'static str,) -> Result<Response, XrpcError> { let (page, nsid) = record_edge_page_of::<TicketRecord, _>(state, q, kind)?; let permit = state.heavy_permit()?; let owned = state.clone(); let items = hydrate_record_stream::<Ticket<DefaultStr>>( state, nsid, page.items, HitProvenance::Indexed, ) .map(move |view| view.map(|v| enrich_ticket_view(&owned, v))); Ok(json_stream::<StatefulItem<Ticket<DefaultStr>>, _>( "items", items, paged_tail(page.next.map(PageToken::encode_token), page.total), permit, ))}
#[allow(dead_code)]async fn list_feed_comments( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records::<FeedCommentRecord, FeedComment<DefaultStr>, _>(&state, q).await}
// todo: distinctAuthors arent really distinct because right now authors are just the repoasync fn count_feed_comments( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<FeedCommentRecord>(&state, q).map(Json)}
async fn list_reactions( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records::<ReactionRecord, Reaction<DefaultStr>, _>(&state, q).await}
async fn count_reactions( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<ReactionRecord>(&state, q).map(Json)}
#[derive(Debug, Deserialize)]struct PatchListQuery { ticket: AtUri<DefaultStr>, cursor: Option<String>, offset: Option<PageOffset>, limit: Option<u32>, #[serde(default)] order: Order,}
async fn list_patches( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<PatchListQuery>,) -> Result<Response, XrpcError> { let q = TypedListQuery { subject: SubjectQuery::Uri(q.ticket), cursor: q.cursor, offset: q.offset, limit: q.limit, order: q.order, filter: NoFilter, }; list_records::<PatchRecord, Patch<DefaultStr>, _>(&state, q).await}
async fn list_collaborators( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { let (page, nsid) = record_edge_page::<CollaboratorRecord, NoFilter>(&state, &q)?; collaborator_list_response(&state, page, nsid)}
async fn count_collaborators( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<CollaboratorRecord>(&state, q).map(Json)}
async fn list_repos( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records::<ManifestRecord, Manifest<DefaultStr>, _>(&state, q).await}
async fn count_repos( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<ManifestRecord>(&state, q).map(Json)}
// forks are repo records pointing back at a repo, so they get their own edge// kind instead of a collectionasync fn count_forks( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { let subject = parse_subject(&q.subject, FORK_SUBJECT_SHAPE)?; let key = EdgeKey::new(nsid_static(REPO_SOURCE_EDGE_KIND), subject); Ok(Json(CountResponse { count: state.edges.count(&key), distinct_authors: state.edges.count_distinct_authors(&key), }))}
async fn list_public_keys( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records::<SshKeyRecord, SshKey<DefaultStr>, _>(&state, q).await}
async fn count_public_keys( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<SshKeyRecord>(&state, q).map(Json)}
struct NetworkVouchQuery { actors: Vec<Did<DefaultStr>>, start: PageCursor, limit: PageLimit,}
fn one_query_value(query: Option<&str>, key: &str) -> Result<Option<String>, XrpcError> { let values = collect_repeated(query, key); match values.as_slice() { [] => Ok(None), [value] => Ok(Some(value.clone())), _ => Err(XrpcError::InvalidParams(format!( "{key} may only be specified once" ))), }}
fn parse_network_vouch_query(query: Option<&str>) -> Result<NetworkVouchQuery, XrpcError> { let actors = collect_repeated(query, "actors"); if actors.is_empty() { return Err(XrpcError::InvalidParams( "at least one actor required".into(), )); } if actors.len() > BULK_LIMIT { return Err(XrpcError::InvalidParams(format!( "at most {BULK_LIMIT} actors per request" ))); } let actors = actors .iter() .map(|actor| { Did::<DefaultStr>::new_owned(actor) .map_err(|_| XrpcError::InvalidParams(format!("invalid did: {actor}"))) }) .collect::<Result<Vec<_>, _>>()?; let mut seen = std::collections::HashSet::new(); let actors = actors .into_iter() .filter(|actor| seen.insert(actor.as_ref().to_owned())) .collect::<Vec<_>>(); let cursor = one_query_value(query, "cursor")?; if actors.len() > 1 && cursor.is_some() { return Err(XrpcError::InvalidParams( "cursor requires exactly one actor".into(), )); } let limit = one_query_value(query, "limit")? .map(|limit| { limit .parse::<u32>() .map_err(|_| XrpcError::InvalidParams(format!("invalid limit: {limit}"))) }) .transpose()?; Ok(NetworkVouchQuery { actors, start: parse_vouch_cursor(cursor.as_deref())?, limit: parse_limit(limit)?, })}
const VOUCH_KIND_VOUCH: &str = "vouch";
fn network_of(actor: &Did<DefaultStr>, vouched: Vec<String>) -> std::collections::HashSet<String> { vouched .into_iter() .chain(std::iter::once(actor.as_ref().to_owned())) .collect()}
async fn vouch_network( state: &AppState, actor: &Did<DefaultStr>,) -> std::collections::HashSet<String> { let nsid = nsid_static("org.tangled.graph.vouch"); let sources = state.edges.sources_for(&EdgeKey::new( nsid_static("org.tangled.graph.vouch.by"), SubjectRef::Did(actor.clone()), )); let vouched = stream::iter(sources) .map(|uri| { let nsid = nsid.clone(); async move { let subject = uri.rkey().map(|rkey| rkey.as_ref().to_owned())?; let view = hydrate_record_view::<Vouch<DefaultStr>>(state, &nsid, uri) .await .ok()??; (AsRef::<str>::as_ref(&view.value.kind) == VOUCH_KIND_VOUCH).then_some(subject) } }) .buffer_unordered(FETCH_CONCURRENCY) .filter_map(|subject| async move { subject }) .collect() .await; network_of(actor, vouched)}
#[derive(Clone, Copy)]enum VouchList { Received, Sent,}
fn vouch_page(edges: &EdgeStore, key: &EdgeKey, list: VouchList, cursor: PageCursor) -> EdgePage { let limit = PageLimit::new(PageLimit::MAX).unwrap(); match list { VouchList::Received => edges.list_distinct_authors(key, cursor, limit, SortDir::Desc), VouchList::Sent => edges.list(key, cursor, limit, SortDir::Desc), }}
fn vouch_items( edges: &EdgeStore, key: EdgeKey, list: VouchList, start: PageCursor,) -> impl Iterator<Item = EdgeItem> + '_ { vouch_items_from(start, move |cursor| vouch_page(edges, &key, list, cursor))}
fn vouch_items_from( start: PageCursor, mut fetch: impl FnMut(PageCursor) -> EdgePage,) -> impl Iterator<Item = EdgeItem> { let mut next = Some(start); let mut items = Vec::<EdgeItem>::new().into_iter(); std::iter::from_fn(move || { loop { if let Some(item) = items.next() { return Some(item); } let cursor = next.take()?; let page = fetch(cursor); next = page.next.map(PageCursor::After); items = page.items.into_iter(); } })}
fn merge_vouches( received: impl Iterator<Item = EdgeItem>, sent: impl Iterator<Item = EdgeItem>,) -> impl Iterator<Item = EdgeItem> { let mut received = received.peekable(); let mut sent = sent.peekable(); std::iter::from_fn(move || match (received.peek(), sent.peek()) { (Some(left), Some(right)) => match left.sort_token().cmp(&right.sort_token()) { std::cmp::Ordering::Greater => received.next(), std::cmp::Ordering::Less => sent.next(), std::cmp::Ordering::Equal => { sent.next(); received.next() } }, (Some(_), None) => received.next(), (None, Some(_)) => sent.next(), (None, None) => None, })}
fn parse_vouch_cursor(raw: Option<&str>) -> Result<PageCursor, XrpcError> { let Some(raw) = raw else { return Ok(PageCursor::Start); }; PageToken::decode_token(raw) .or_else(|_| { raw.parse::<u64>() .map(|micros| PageToken::new(micros, 0)) .map_err(|_| CursorParseError::Malformed) }) .map(PageCursor::After) .map_err(|error| XrpcError::InvalidParams(format!("cursor: {error}")))}
fn paginate_vouches( candidates: impl Iterator<Item = EdgeItem>, limit: PageLimit,) -> (Vec<EdgeItem>, Option<String>) { let limit = limit.get() as usize; let mut seen = std::collections::HashSet::new(); let mut items = candidates .filter(|item| seen.insert(item.uri.clone())) .take(limit + 1) .collect::<Vec<_>>(); let has_more = items.len() > limit; items.truncate(limit); let next = has_more.then(|| items.last().unwrap().sort_token().encode_token()); (items, next)}
fn network_vouch_page( edges: &EdgeStore, actor: &Did<DefaultStr>, network: &VouchNetworkFilter, start: PageCursor, limit: PageLimit,) -> (Vec<EdgeItem>, Option<String>) { let received = vouch_items( edges, EdgeKey::new( nsid_static("org.tangled.graph.vouch"), SubjectRef::Did(actor.clone()), ), VouchList::Received, start, ) .filter(|item| network.matches_author(item)); let sent = vouch_items( edges, EdgeKey::new( nsid_static("org.tangled.graph.vouch.by"), SubjectRef::Did(actor.clone()), ), VouchList::Sent, start, ) .filter(|item| network.matches_subject(item)); paginate_vouches(merge_vouches(received, sent), limit)}
fn network_vouch_pages( edges: &EdgeStore, actors: &[Did<DefaultStr>], network: &VouchNetworkFilter, start: PageCursor, limit: PageLimit,) -> (Vec<(Did<DefaultStr>, EdgeItem)>, Option<String>) { let single = actors.len() == 1; let mut next_cursor = None; let mut tagged = Vec::new(); for actor in actors { let (items, cursor) = network_vouch_page(edges, actor, network, start, limit); tagged.extend(items.into_iter().map(|item| (actor.clone(), item))); if single { next_cursor = cursor; } } (tagged, next_cursor)}
#[cfg(test)]mod vouch_pagination_tests { use std::collections::HashSet;
use super::*; use bobbin_runtime::RuntimeHasher; use bobbin_types::edges::Edge;
fn did(value: &str) -> Did<DefaultStr> { Did::new_owned(value).unwrap() }
fn uri(value: String) -> AtUri<DefaultStr> { AtUri::new_owned(value).unwrap() }
fn edge(kind: &'static str, subject: &Did<DefaultStr>, author: usize, micros: u64) -> Edge { Edge { kind: nsid_static(kind), subject: SubjectRef::Did(subject.clone()), source: uri(format!( "at://did:plc:author{author}/org.tangled.graph.vouch/r{author}" )), sort_micros: micros, } }
#[test] fn received_filter_reaches_beyond_the_first_candidate_page() { let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4)); let subject = did("did:plc:subject"); (0..=PageLimit::MAX as usize).for_each(|author| { store.add(edge( "org.tangled.graph.vouch", &subject, author, author as u64, )); }); let key = EdgeKey::new( nsid_static("org.tangled.graph.vouch"), SubjectRef::Did(subject), ); let candidates = vouch_items(&store, key, VouchList::Received, PageCursor::Start) .filter(|item| item.uri.as_ref().starts_with("at://did:plc:author0/")); let (items, cursor) = paginate_vouches(candidates, PageLimit::new(1).unwrap());
assert_eq!(items.len(), 1); assert!(items[0].uri.as_ref().starts_with("at://did:plc:author0/")); assert!(cursor.is_none()); }
#[test] fn cursor_walk_keeps_every_timestamp_tie() { let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4)); let subject = did("did:plc:subject"); store.add(edge("org.tangled.graph.vouch", &subject, 1, 42)); store.add(edge("org.tangled.graph.vouch.by", &subject, 2, 42)); store.add(edge("org.tangled.graph.vouch", &subject, 3, 42)); let received = EdgeKey::new( nsid_static("org.tangled.graph.vouch"), SubjectRef::Did(subject.clone()), ); let sent = EdgeKey::new( nsid_static("org.tangled.graph.vouch.by"), SubjectRef::Did(subject), ); let mut start = PageCursor::Start; let mut uris = Vec::new();
loop { let candidates = merge_vouches( vouch_items(&store, received.clone(), VouchList::Received, start), vouch_items(&store, sent.clone(), VouchList::Sent, start), ); let (items, cursor) = paginate_vouches(candidates, PageLimit::new(1).unwrap()); uris.extend(items.into_iter().map(|item| item.uri)); let Some(cursor) = cursor else { break; }; assert_eq!(cursor.len(), 24); start = parse_vouch_cursor(Some(&cursor)).unwrap(); }
assert_eq!(uris.len(), 3); assert_eq!(uris.iter().collect::<HashSet<_>>().len(), 3); }
#[test] fn page_fetch_is_deferred_until_items_are_needed() { let calls = std::cell::Cell::new(0usize); let item = |source: u32| EdgeItem { uri: uri(format!( "at://did:plc:author/org.tangled.graph.vouch/r{source}" )), sort_micros: 42, sort_source: source, }; let mut items = vouch_items_from(PageCursor::Start, |cursor| { calls.set(calls.get() + 1); match cursor { PageCursor::Start => EdgePage { items: vec![item(3), item(2)], next: Some(PageToken::new(42, 2)), total: None, }, PageCursor::After(token) => { assert_eq!(token, PageToken::new(42, 2)); EdgePage { items: vec![item(1)], next: None, total: None, } } } }); assert_eq!(calls.get(), 0); assert_eq!(items.next().unwrap().sort_source, 3); assert_eq!(calls.get(), 1); assert_eq!(items.next().unwrap().sort_source, 2); assert_eq!(calls.get(), 1); assert_eq!(items.next().unwrap().sort_source, 1); assert_eq!(calls.get(), 2); assert!(items.next().is_none()); assert!(items.next().is_none()); assert_eq!(calls.get(), 2); }
#[test] fn deferred_fetch_continues_through_an_empty_page_with_a_cursor() { let calls = std::cell::Cell::new(0usize); let mut items = vouch_items_from(PageCursor::Start, |cursor| { calls.set(calls.get() + 1); match cursor { PageCursor::Start => EdgePage { items: Vec::new(), next: Some(PageToken::new(42, 2)), total: None, }, PageCursor::After(token) => { assert_eq!(token, PageToken::new(42, 2)); EdgePage { items: vec![EdgeItem { uri: uri("at://did:plc:author/org.tangled.graph.vouch/r1".to_owned()), sort_micros: 42, sort_source: 1, }], next: None, total: None, } } } }); assert_eq!(items.next().unwrap().sort_source, 1); assert_eq!(calls.get(), 2); assert!(items.next().is_none()); assert_eq!(calls.get(), 2); }
#[test] fn decimal_vouch_cursor_keeps_legacy_resume_semantics() { assert_eq!( parse_vouch_cursor(Some("42")).unwrap(), PageCursor::After(PageToken::new(42, 0)), ); }
fn vouch_edge( kind: &'static str, index_subject: &Did<DefaultStr>, author: &Did<DefaultStr>, record_subject: &Did<DefaultStr>, micros: u64, ) -> Edge { Edge { kind: nsid_static(kind), subject: SubjectRef::Did(index_subject.clone()), source: uri(format!( "at://{author}/org.tangled.graph.vouch/{record_subject}" )), sort_micros: micros, } }
#[test] fn viewer_authored_direct_vouch_is_returned() { let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4)); let viewer = did("did:plc:viewer"); let actor = did("did:plc:actor"); store.add(vouch_edge( "org.tangled.graph.vouch", &actor, &viewer, &actor, 42, )); let network = VouchNetworkFilter::new(network_of(&viewer, Vec::new())); let (items, _) = network_vouch_page( &store, &actor, &network, PageCursor::Start, PageLimit::new(10).unwrap(), );
assert_eq!(items.len(), 1); assert_eq!(source_authority_did(&items[0].uri), Some(viewer)); }
#[test] fn sent_vouches_match_their_record_subject_against_the_network() { let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4)); let actor = did("did:plc:actor"); let member = did("did:plc:member"); let outsider = did("did:plc:outsider"); store.add(vouch_edge( "org.tangled.graph.vouch.by", &actor, &actor, &member, 42, )); store.add(vouch_edge( "org.tangled.graph.vouch.by", &actor, &actor, &outsider, 41, )); let network = VouchNetworkFilter::new(HashSet::from([member.as_ref().to_owned()])); let (items, _) = network_vouch_page( &store, &actor, &network, PageCursor::Start, PageLimit::new(10).unwrap(), );
assert_eq!(items.len(), 1); assert_eq!(items[0].uri.rkey().unwrap().as_ref(), member.as_ref()); }
#[test] fn batch_tags_each_actor_and_applies_the_limit_per_actor() { let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4)); let first = did("did:plc:first"); let second = did("did:plc:second"); let authors = [ did("did:plc:author1"), did("did:plc:author2"), did("did:plc:author3"), did("did:plc:author4"), ]; for (index, actor) in [first.clone(), second.clone()].iter().enumerate() { for offset in 0..2 { let author = &authors[index * 2 + offset]; store.add(vouch_edge( "org.tangled.graph.vouch", actor, author, actor, 10 - offset as u64, )); } } let query = parse_network_vouch_query(Some( "actors=did%3Aplc%3Afirst&actors=did%3Aplc%3Asecond&limit=1", )) .unwrap(); let network = VouchNetworkFilter::new( authors .iter() .map(|author| author.as_ref().to_owned()) .collect(), ); let (items, cursor) = network_vouch_pages(&store, &query.actors, &network, query.start, query.limit);
assert_eq!( items .iter() .map(|(actor, _)| actor.clone()) .collect::<Vec<_>>(), vec![first, second] ); assert!(cursor.is_none()); }
#[test] fn repeated_actors_collapse_into_one_group() { let query = parse_network_vouch_query(Some( "actors=did%3Aplc%3Afirst&actors=did%3Aplc%3Afirst&actors=did%3Aplc%3Asecond", )) .unwrap();
assert_eq!( query.actors, vec![did("did:plc:first"), did("did:plc:second")] ); }
fn vouch_state(edges: Arc<EdgeStore>) -> AppState { AppState::new( Arc::new(bobbin_record_lru::LruRecordStore::new( bobbin_record_lru::CacheCapacity::from_bytes(64 * 1024), )), SlingshotClient::with_default_http(url::Url::parse("http://127.0.0.1:1").unwrap()) .unwrap(), edges, Arc::new(CoverageWatch::new()), Arc::new( bobbin_search::SearchIndex::new( bobbin_search::DEFAULT_WRITER_HEAP_BYTES, Arc::new(bobbin_runtime::SystemClock::new()), ) .unwrap(), ) as Arc<dyn SearchReader>, Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), Arc::new(crate::default_directory()), ) }
fn cache_vouch( state: &AppState, author: &Did<DefaultStr>, subject: &Did<DefaultStr>, kind: &str, ) { let at = uri(format!("at://{author}/org.tangled.graph.vouch/{subject}")); let value = serde_json::json!({ "$type": "org.tangled.graph.vouch", "kind": kind, "createdAt": "2026-05-01T00:00:00Z" }); state.records.put( at.clone(), Arc::new(RecordBody { uri: at, cid: Cid::new(b"bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i") .unwrap(), value: serde_json::to_vec(&value).unwrap().into(), }), ); }
#[tokio::test] async fn a_denounced_did_stays_out_of_the_viewer_network() { let edges = Arc::new(EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4))); let viewer = did("did:plc:viewer"); let friend = did("did:plc:friend"); let enemy = did("did:plc:enemy"); edges.add(vouch_edge( "org.tangled.graph.vouch.by", &viewer, &viewer, &friend, 42, )); edges.add(vouch_edge( "org.tangled.graph.vouch.by", &viewer, &viewer, &enemy, 41, )); let state = vouch_state(edges); cache_vouch(&state, &viewer, &friend, "vouch"); cache_vouch(&state, &viewer, &enemy, "denounce");
let network = vouch_network(&state, &viewer).await;
assert!(network.contains(friend.as_ref())); assert!(!network.contains(enemy.as_ref())); assert!(network.contains(viewer.as_ref())); }
#[test] fn cursor_is_rejected_for_multiple_actors() { assert!(matches!( parse_network_vouch_query(Some( "actors=did%3Aplc%3Afirst&actors=did%3Aplc%3Asecond&cursor=42" )), Err(XrpcError::InvalidParams(_)) )); }}
async fn list_network_vouches( State(state): State<AppState>, ExtractOptionalServiceAuth(auth): ExtractOptionalServiceAuth, RawQuery(query): RawQuery,) -> Result<Response, XrpcError> { let q = parse_network_vouch_query(query.as_deref())?; let permit = state.heavy_permit()?; let vouch_nsid = nsid_static("org.tangled.graph.vouch"); let (page_items, next_cursor) = match auth.as_ref() { Some(auth) => { let viewer = auth.did().into_static(); let network = VouchNetworkFilter::new(vouch_network(&state, &viewer).await); network_vouch_pages(&state.edges, &q.actors, &network, q.start, q.limit) } None => (Vec::new(), None), }; let views = hydrate_keyed_record_stream::<_, Vouch<DefaultStr>>( &state, vouch_nsid, page_items, HitProvenance::Indexed, ) .map_ok(|(actor, view)| NetworkVouchItem { actor, view }); Ok(json_stream::<NetworkVouchItem<Vouch<DefaultStr>>, _>( "items", views, paged_tail(next_cursor, None), permit, ))}
async fn list_stars_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_mirror::<StarBy, Star<DefaultStr>, _>(&state, q).await}async fn count_stars_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_mirror::<StarBy>(&state, q).map(Json)}
async fn list_reactions_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_mirror::<ReactionBy, Reaction<DefaultStr>, _>(&state, q).await}async fn count_reactions_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_mirror::<ReactionBy>(&state, q).map(Json)}
async fn list_follows_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_mirror::<FollowBy, Follow<DefaultStr>, _>(&state, q).await}async fn count_follows_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_mirror::<FollowBy>(&state, q).map(Json)}
#[derive(serde::Deserialize)]struct InvitesByQuery { subject: Did<DefaultStr>,}
#[derive(serde::Serialize)]#[serde(rename_all = "camelCase")]struct InviteItem { uri: AtUri<DefaultStr>, knot: Did<DefaultStr>, repo: Did<DefaultStr>, added_by: Did<DefaultStr>, created_at: Datetime,}
#[derive(serde::Serialize)]struct InvitesByOutput { items: Vec<InviteItem>, pending: Vec<Did<DefaultStr>>, truncated: bool,}
const COLLABORATOR_INVITE_COLLECTION: &str = "sh.tangled.repo.collaboratorInvite";
fn list_invites_by( state: &AppState, q: InvitesByQuery,) -> Result<Json<InvitesByOutput>, XrpcError> { let subject = q.subject; if !state.invites.discovered() { return Err(XrpcError::Warming); }
let offered: Vec<(InviteScope, Offer)> = state.invites.outstanding(&subject); let truncated = offered.len() > MAX_OUTSTANDING_OFFERS;
let items = offered .into_iter() .take(MAX_OUTSTANDING_OFFERS) .map(|(scope, offer)| { let created_at = datetime_of(offer.offered_at).ok_or_else(|| { XrpcError::Internal(format!( "offer to {} has timestamp we can't render, {}", subject.as_ref(), offer.offered_at.raw() )) })?; let uri = AtUri::<DefaultStr>::new_owned(format!( "at://{}/{}/{}", scope.repo.as_ref(), COLLABORATOR_INVITE_COLLECTION, subject.as_ref() )) .map_err(|e| { XrpcError::Internal(format!( "offer to {} won't form record uri, {e}", subject.as_ref() )) })?; Ok(InviteItem { uri, knot: scope.knot, repo: scope.repo, added_by: offer.invited_by, created_at, }) }) .collect::<Result<Vec<InviteItem>, XrpcError>>()?;
Ok(Json(InvitesByOutput { items, pending: state.invites.pending(), truncated, }))}
async fn list_collaborator_invites_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<InvitesByQuery>,) -> Result<Json<InvitesByOutput>, XrpcError> { list_invites_by(&state, q)}
async fn list_artifacts_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_mirror::<ArtifactBy, Artifact<DefaultStr>, _>(&state, q).await}async fn count_artifacts_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_mirror::<ArtifactBy>(&state, q).map(Json)}
async fn list_collaborators_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { let (page, nsid) = mirror_edge_page::<CollaboratorBy, NoFilter>(&state, &q)?; collaborator_list_response(&state, page, nsid)}async fn count_collaborators_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_mirror::<CollaboratorBy>(&state, q).map(Json)}
async fn list_feed_comments_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_mirror::<FeedCommentBy, FeedComment<DefaultStr>, _>(&state, q).await}async fn count_feed_comments_by( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_mirror::<FeedCommentBy>(&state, q).map(Json)}
async fn list_label_definitions( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records::<LabelDefinitionRecord, LabelDefinition<DefaultStr>, _>(&state, q).await}
async fn count_label_definitions( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<LabelDefinitionRecord>(&state, q).map(Json)}
async fn list_artifacts( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records::<ArtifactRecord, Artifact<DefaultStr>, _>(&state, q).await}
async fn count_artifacts( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<ArtifactRecord>(&state, q).map(Json)}
async fn list_strings( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,) -> Result<Response, XrpcError> { list_records::<SnippetStringRecord, SnippetString<DefaultStr>, _>(&state, q).await}
async fn count_strings( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<CountQuery>,) -> Result<Json<CountResponse>, XrpcError> { count_for::<SnippetStringRecord>(&state, q).map(Json)}
#[derive(Deserialize)]struct ResolveMiniDocParams { identifier: AtIdentifier<DefaultStr>,}
async fn resolve_mini_doc( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<ResolveMiniDocParams>,) -> Result<Response, XrpcError> { let doc = state .identity .resolve_minidoc(&q.identifier) .await .map_err(|error| match error { IdentityResolveError::NotFound => XrpcError::NotFound, IdentityResolveError::Upstream(message) => XrpcError::UpstreamUnavailable(message), IdentityResolveError::Decode(message) => XrpcError::InvalidRecord(message), })?; Ok(Json(doc).into_response())}
async fn get_ready(State(state): State<AppState>) -> (StatusCode, Json<CoverageEnvelope>) { let mut envelope = CoverageEnvelope::from(state.coverage.snapshot()); envelope.stream = state .stream_health .as_deref() .map(|health| health.snapshot().into()); let status = envelope .ready .then_some(StatusCode::OK) .unwrap_or(StatusCode::SERVICE_UNAVAILABLE); (status, Json(envelope))}
async fn get_coverage(State(state): State<AppState>) -> Json<CoverageEnvelope> { let mut envelope = CoverageEnvelope::from(state.coverage.snapshot()); envelope.stream = state .stream_health .as_deref() .map(|health| health.snapshot().into()); Json(envelope)}// enough for a write to travel pds -> relay -> indexer -> bobbin, plus slack// for a slow hopconst AWAIT_RECORD_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug, Deserialize)]struct AwaitRecordQuery { uri: AtUri<DefaultStr>, cid: Option<ParsedCid>,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]struct AwaitRecordResponse { status: &'static str, #[serde(skip_serializing_if = "Option::is_none")] cid: Option<ParsedCid>, #[serde(skip_serializing_if = "Option::is_none")] reason: Option<&'static str>, #[serde(skip_serializing_if = "Option::is_none")] detail: Option<Box<str>>,}
impl From<Settlement> for AwaitRecordResponse { fn from(s: Settlement) -> Self { Self { status: s.outcome.as_str(), cid: s.cid, reason: s.outcome.rejection().map(Rejection::as_str), detail: s.outcome.detail().map(Box::from), } }}
async fn await_record( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<AwaitRecordQuery>,) -> Result<Json<AwaitRecordResponse>, XrpcError> { let Some((collection, _)) = q.uri.collection().zip(q.uri.rkey()) else { return Err(XrpcError::InvalidParams( "uri must be at://<did>/<collection>/<rkey>".into(), )); }; if !collection.as_str().starts_with(TANGLED_NSID_PREFIX) && !collection.as_str().starts_with(ORG_TANGLED_NSID_PREFIX) { return Err(XrpcError::InvalidParams(format!( "bobbin never sees {collection} records" ))); } // records are only ever indexed under their did, a handle would wait forever if source_authority_did(&q.uri).is_none() { return Err(XrpcError::InvalidParams( "uri authority must be a did, not a handle".into(), )); } // Drop releases the await slot when the handler returns let _awaiting = state.awaiting.try_enter()?; let settled = match state.settlements.watch(&q.uri, q.cid) { Watch::Settled(s) => s, // Drop removes the waiter from the list when the handler returns, // so a timeout won't leave a channel no one is waiting on Watch::Waiting(_registration, rx) => { match tokio::time::timeout(AWAIT_RECORD_TIMEOUT, rx).await { Ok(Ok(s)) => s, Ok(Err(_)) | Err(_) => return Err(XrpcError::NotSettled), } } }; Ok(Json(settled.into()))}#[derive(Deserialize)]#[allow(dead_code)]struct ListRecipientsQuery { subject: String, collection: Option<String>,}
#[derive(Serialize)]struct ListRecipientsResponse { dids: Vec<String>,}
async fn list_recipients( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<ListRecipientsQuery>,) -> Result<Json<ListRecipientsResponse>, XrpcError> { use std::collections::BTreeSet;
let subject_ref = if let Ok(uri) = AtUri::<DefaultStr>::new_owned(&q.subject) { SubjectRef::Uri(uri) } else if let Ok(did) = Did::<DefaultStr>::new_owned(&q.subject) { SubjectRef::Did(did) } else { return Err(XrpcError::InvalidParams( "subject must be an at-uri or did".into(), )); };
let key = EdgeKey::new(nsid_static("org.tangled.feed.subscription"), subject_ref); let sources = state.edges.sources_for(&key);
// NOTE: collection filtering intentionally disabled. // All subscribers are returned regardless of their stored collection filter. let filtered = sources; let mut seen = BTreeSet::new(); let mut dids = Vec::new(); for src in &filtered { if let Some(did) = owner_did_from_aturi(src) && seen.insert(did.to_string()) { dids.push(did.to_string()); } }
Ok(Json(ListRecipientsResponse { dids }))}
async fn search_actors_typeahead( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<ActorTypeaheadParams>,) -> Result<Json<ActorTypeaheadResponse>, XrpcError> { let query = q.q.trim().trim_start_matches('@'); if query.is_empty() { return Err(XrpcError::InvalidParams("q must not be empty".into())); } if query.len() > ACTOR_TYPEAHEAD_MAX_QUERY_BYTES { return Err(XrpcError::InvalidParams(format!( "q must be at most {ACTOR_TYPEAHEAD_MAX_QUERY_BYTES} bytes", ))); } let limit = q.limit.unwrap_or(ACTOR_TYPEAHEAD_DEFAULT_LIMIT); if !(1..=ACTOR_TYPEAHEAD_MAX_LIMIT).contains(&limit) { return Err(XrpcError::InvalidParams(format!( "limit must be between 1 and {ACTOR_TYPEAHEAD_MAX_LIMIT}", ))); } let _permit = state.heavy_permit()?; let suggestions = state .actors .suggest(query, limit) .await .map_err(map_actor_search_err)?; let actors = suggestions .into_iter() .map(|suggestion| { let uri = AtUri::new_owned(format!( "at://{}/org.tangled.actor.profile/self", suggestion.did )) .map_err(|error| { XrpcError::Internal(format!("could not build profile URI: {error}")) })?; Ok(ActorTypeaheadView { uri, did: suggestion.did, handle: suggestion.handle, }) }) .collect::<Result<Vec<_>, XrpcError>>()?; Ok(Json(ActorTypeaheadResponse { actors }))}async fn search_query( State(state): State<AppState>, XrpcQuery(q): XrpcQuery<SearchQueryParams>,) -> Result<Response, XrpcError> { if q.q.trim().is_empty() { return Err(XrpcError::InvalidParams("q must not be empty".into())); } let cursor = match (q.cursor.as_deref(), q.offset) { (Some(_), Some(_)) => { return Err(XrpcError::InvalidParams( "cursor and offset are mutually exclusive".into(), )); } (Some(c), None) => SearchCursor::from_token(Some(c)) .map_err(|e| XrpcError::InvalidParams(format!("cursor: {e}")))?, (None, Some(o)) => SearchCursor::At(SearchOffset::new(o.get() as u32)), (None, None) => SearchCursor::Start, }; let limit = parse_limit(q.limit)?; let filters = build_search_filters(&q)?; let permit = state.heavy_permit()?; let page = state .search .search(&q.q, filters, cursor, limit.get()) .await .map_err(map_search_err)?; let next = page.next.map(SearchOffset::encode_token); let owned = state.clone(); let hits = hydrate_stream(page.hits, move |hit| { let owned = owned.clone(); let uri = hit.uri.clone(); let nsid = hit.nsid.clone(); async move { let result = hydrate_search_hit(&owned, hit).await; drop_unhydratable(HitProvenance::Indexed, &nsid, &uri, result) } }); Ok(json_stream::<SearchHitView, _>( "hits", hits, paged_tail(next, None), permit, ))}
fn build_search_filters(q: &SearchQueryParams) -> Result<SearchFilters, XrpcError> { let since = q .since .as_deref() .map(parse_rfc3339_seconds) .transpose() .map_err(|e| XrpcError::InvalidParams(format!("since: {e}")))?; let until = q .until .as_deref() .map(parse_rfc3339_seconds) .transpose() .map_err(|e| XrpcError::InvalidParams(format!("until: {e}")))?; if let (Some(s), Some(u)) = (since, until) && s > u { return Err(XrpcError::InvalidParams("since must be <= until".into())); } Ok(SearchFilters { nsid: q.nsid.clone(), author: q.author.clone(), repo: q.repo.clone(), since, until, })}
fn parse_rfc3339_seconds(raw: &str) -> Result<i64, String> { chrono::DateTime::parse_from_rfc3339(raw) .map(|dt| dt.timestamp()) .map_err(|e| format!("expected RFC3339, got {raw}: {e}"))}
async fn hydrate_search_hit( state: &AppState, hit: SearchHit,) -> Result<Option<SearchHitView>, XrpcError> { let SearchHit { uri, nsid, score } = hit; let body = resolve_for_view(state, &nsid, uri).await?; let record = decode_record(&nsid, &body.value) .map_err(|err| XrpcError::InvalidRecord(err.to_string()))?; let Some(value) = SearchableRecord::try_from_record(record) else { return Ok(None); }; let Some(value) = value.normalize(&state.resolver).await else { return Ok(None); }; Ok(Some(SearchHitView { uri: body.uri.clone(), cid: Some(body.cid.clone()), nsid, score, value, }))}
fn map_search_err(err: SearchError) -> XrpcError { use SearchError as E; match err { E::Query(e) => XrpcError::InvalidParams(format!("query: {e}")), e @ (E::Tantivy(_) | E::InvalidUri(_) | E::InvalidNsid(_) | E::MissingField(_) | E::Cancelled(_)) => XrpcError::Internal(format!("search: {e}")), }}fn map_actor_search_err(err: ActorIndexError) -> XrpcError { XrpcError::Internal(format!("actor search: {err}"))}
fn map_proxy_error(err: KnotProxyError) -> XrpcError { match err { KnotProxyError::CircuitOpen => { XrpcError::UpstreamUnavailable("knot circuit breaker open".into()) } KnotProxyError::BlockedHost { host, reason } => { XrpcError::InvalidRecord(format!("knot host {host} is {reason} address space")) } KnotProxyError::PlaintextHttp { host } => { XrpcError::InvalidRecord(format!("knot host {host} requires https")) } KnotProxyError::Connect(e) => XrpcError::UpstreamUnavailable(format!("connect: {e}")), KnotProxyError::Timeout(e) => { XrpcError::UpstreamUnavailable(format!("upstream timeout: {e}")) } KnotProxyError::Redirect(e) => XrpcError::UpstreamUnavailable(format!("redirect: {e}")), KnotProxyError::Transport(e) => XrpcError::UpstreamUnavailable(format!("transport: {e}")), KnotProxyError::Upstream(s) => XrpcError::UpstreamUnavailable(format!("status {s}")), }}
fn filter_request_headers( client: &HeaderMap, socket: SocketPeer, address: &ClientAddress,) -> HeaderMap { let forwarded = RANGE_OR_CONDITIONAL_HEADERS .iter() .fold(HeaderMap::new(), |mut acc, name| { if let Some(value) = client.get(*name) { acc.insert((*name).clone(), value.clone()); } acc }); address .of(client, socket) .into_iter() .fold(forwarded, |mut acc, address| { acc.insert(X_FORWARDED_FOR.clone(), address); acc })}
fn upstream_to_axum(resp: ProxyResponse) -> Response { let status = resp.status(); let upstream_headers = resp.headers().clone(); let body = Body::from_stream(resp.into_body_stream()); let mut response = Response::builder() .status(status) .body(body) .expect("response body construction must succeed"); let response_headers = response.headers_mut(); PASSTHROUGH_HEADERS.iter().for_each(|name| { if let Some(value) = upstream_headers.get(*name) { response_headers.insert((*name).clone(), value.clone()); } }); response}
async fn proxy_mirror_handler( state: AppState, headers: HeaderMap, socket: SocketPeer, params: ProxyParams, nsid: Nsid<DefaultStr>,) -> Result<Response, XrpcError> { let mirror = state .mirror .as_ref() .ok_or_else(|| XrpcError::UpstreamUnavailable("mirror.url is unset".into()))?; let allowed = filter_request_headers(&headers, socket, &state.client_address); let forward: Vec<(&str, &str)> = params .iter() .map(|(k, v)| (k.as_str(), v.as_str())) .collect(); mirror .forward_raw(&nsid, &forward, allowed) .await .map(upstream_to_axum) .map_err(map_proxy_error)}
async fn proxy_mirror_write_handler( state: AppState, headers: HeaderMap, body: Bytes, nsid: Nsid<DefaultStr>,) -> Result<Response, XrpcError> { let mirror = state .mirror .as_ref() .ok_or_else(|| XrpcError::UpstreamUnavailable("mirror.url is unset".into()))?; let token = headers .get(AUTHORIZATION) .ok_or_else(|| XrpcError::AuthRequired("the mirror needs a service-auth token".into()))?;
let url = format!("{}xrpc/{nsid}", mirror.host().url().as_str(),); let request = reqwest::Client::new() .post(url) .header(AUTHORIZATION, token) .header( CONTENT_TYPE, headers .get(CONTENT_TYPE) .cloned() .unwrap_or(axum::http::HeaderValue::from_static("application/json")), ) .body(body);
let upstream = request .send() .await .map_err(|e| XrpcError::UpstreamUnavailable(format!("mirror: {e}")))?;
let status = upstream.status(); let upstream_headers = upstream.headers().clone(); let mut response = Response::builder() .status(status) .body(Body::from_stream(upstream.bytes_stream())) .expect("response body construction must succeed"); let response_headers = response.headers_mut(); PASSTHROUGH_HEADERS.iter().for_each(|name| { if let Some(value) = upstream_headers.get(*name) { response_headers.insert((*name).clone(), value.clone()); } }); Ok(response)}