alice.pds.demo.boltless.dev
This repository has no description
1use std::future::Future;
2use std::sync::Arc;
3
4use bobbin_resolver::{
5 NormalizeRepoRefs, decode_canon_or_upgrade, normalize_record_fields, scrub_record_bytes,
6 upgrade_wire_bytes,
7};
8
9use axum::{
10 Router,
11 body::Body,
12 extract::{FromRequestParts, Query, RawQuery, State, rejection::QueryRejection},
13 http::{
14 HeaderMap, HeaderName, StatusCode,
15 header::{
16 ACCEPT_RANGES, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LENGTH,
17 CONTENT_RANGE, CONTENT_TYPE, ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, IF_RANGE,
18 LAST_MODIFIED, RANGE,
19 },
20 request::Parts,
21 },
22 response::{IntoResponse, Json, Response},
23 routing::get,
24};
25use bobbin_edge_index::{
26 Coverage, CoverageWatch, CursorParseError, EdgeItem, EdgePage, EdgeStore, IssueStateKind,
27 PageCursor, PageLimit, PageToken, PullStatusKind, SortDir, StateIndex, StateKind,
28};
29use bobbin_knot_proxy::{KnotHost, KnotProxy, KnotProxyError, ProxyResponse, RepoSlug};
30use bobbin_record_lru::RecordStore;
31use bobbin_resolver::RepoIdResolver;
32use bobbin_runtime::ReqwestHttp;
33use bobbin_search::{
34 SearchCursor, SearchError, SearchFilters, SearchHit, SearchOffset, SearchReader,
35};
36use bobbin_slingshot_client::{SlingshotClient, SlingshotError};
37use bobbin_types::edges::REPO_SOURCE_EDGE_KIND;
38use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static};
39use bobbin_types::knot_acl::{KnotOwnedSource, decode_knot_owned_source, knot_did_host};
40use bobbin_types::record::RecordBody;
41use bobbin_types::search::SearchableRecord;
42use bobbin_types::sh_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord};
43use bobbin_types::sh_tangled::feed::comment::{
44 Comment as FeedComment, CommentRecord as FeedCommentRecord,
45};
46use bobbin_types::sh_tangled::feed::reaction::{Reaction, ReactionRecord};
47use bobbin_types::sh_tangled::feed::star::{Star, StarRecord};
48use bobbin_types::sh_tangled::git::ref_update::{RefUpdate, RefUpdateRecord};
49use bobbin_types::sh_tangled::graph::follow::{Follow, FollowRecord};
50use bobbin_types::sh_tangled::graph::vouch::{Vouch, VouchRecord};
51use bobbin_types::sh_tangled::knot::member::{
52 Member as KnotMember, MemberRecord as KnotMemberRecord,
53};
54use bobbin_types::sh_tangled::knot::{Knot, KnotRecord};
55use bobbin_types::sh_tangled::label::definition::{
56 Definition as LabelDefinition, DefinitionRecord as LabelDefinitionRecord,
57};
58use bobbin_types::sh_tangled::label::op::{Op as LabelOp, OpRecord as LabelOpRecord};
59use bobbin_types::sh_tangled::pipeline::status::{
60 Status as PipelineStatus, StatusRecord as PipelineStatusRecord,
61};
62use bobbin_types::sh_tangled::pipeline::{Pipeline, PipelineRecord};
63use bobbin_types::sh_tangled::public_key::{PublicKey, PublicKeyGetRecordOutput, PublicKeyRecord};
64use bobbin_types::sh_tangled::repo::artifact::{Artifact, ArtifactRecord};
65use bobbin_types::sh_tangled::repo::collaborator::{Collaborator, CollaboratorRecord};
66use bobbin_types::sh_tangled::repo::issue::state::{
67 State as IssueState, StateRecord as IssueStateRecord,
68};
69use bobbin_types::sh_tangled::repo::issue::{Issue, IssueGetRecordOutput, IssueRecord};
70use bobbin_types::sh_tangled::repo::pull::status::{
71 Status as PullStatus, StatusRecord as PullStatusRecord,
72};
73use bobbin_types::sh_tangled::repo::pull::{Pull, PullGetRecordOutput, PullRecord};
74use bobbin_types::sh_tangled::repo::{Repo, RepoGetRecordOutput, RepoRecord};
75use bobbin_types::sh_tangled::spindle::member::{
76 Member as SpindleMember, MemberRecord as SpindleMemberRecord,
77};
78use bobbin_types::sh_tangled::spindle::{Spindle, SpindleRecord};
79use bobbin_types::sh_tangled::string::{
80 TangledString, TangledStringGetRecordOutput, TangledStringRecord,
81};
82use futures::Stream;
83use futures::stream::{self, StreamExt, TryStreamExt};
84use jacquard_common::types::did::Did;
85use jacquard_common::types::ident::AtIdentifier;
86use jacquard_common::types::nsid::Nsid;
87use jacquard_common::types::recordkey::Rkey;
88use jacquard_common::types::string::{AtUri, Cid};
89use jacquard_common::xrpc::XrpcResp;
90use jacquard_common::{DefaultStr, IntoStatic};
91use jacquard_identity::JacquardResolver;
92use serde::{Deserialize, Serialize};
93use std::convert::Infallible;
94use std::time::Duration;
95use thiserror::Error;
96use url::form_urlencoded;
97
98use tower_http::classify::ServerErrorsFailureClass;
99use tower_http::trace::{DefaultMakeSpan, OnFailure, OnResponse, TraceLayer};
100use tracing::{Level, Span};
101
102mod backpressure;
103mod enrich;
104mod feed;
105mod filter;
106mod recordpath;
107
108pub use backpressure::{
109 HeavyLimiter, HeavyPermit, MaxInFlight, PerRequestAnonBytes, PressureVerdict, ReservedFloor,
110};
111use filter::{CountFilter, IssueFilter, ListFilter, NoFilter, PullFilter};
112
113const DEFAULT_LIMIT: u32 = 50;
114const FETCH_CONCURRENCY: usize = 8;
115
116pub type Directory = JacquardResolver<ReqwestHttp>;
117
118pub fn default_directory() -> Directory {
119 JacquardResolver::new(ReqwestHttp::new(reqwest::Client::new()), Default::default())
120}
121
122#[derive(Clone)]
123pub struct AppState {
124 pub records: Arc<dyn RecordStore>,
125 pub slingshot: SlingshotClient,
126 pub edges: Arc<EdgeStore>,
127 pub issue_states: Arc<StateIndex<IssueStateKind>>,
128 pub pull_statuses: Arc<StateIndex<PullStatusKind>>,
129 pub coverage: Arc<CoverageWatch>,
130 pub knots: Arc<KnotProxy>,
131 pub search: Arc<dyn SearchReader>,
132 pub resolver: Arc<RepoIdResolver>,
133 pub directory: Arc<Directory>,
134 pub limiter: Option<Arc<HeavyLimiter>>,
135 enrich_router: Arc<std::sync::OnceLock<Router>>,
136}
137
138impl AppState {
139 #[allow(clippy::too_many_arguments)]
140 pub fn new(
141 records: Arc<dyn RecordStore>,
142 slingshot: SlingshotClient,
143 edges: Arc<EdgeStore>,
144 issue_states: Arc<StateIndex<IssueStateKind>>,
145 pull_statuses: Arc<StateIndex<PullStatusKind>>,
146 coverage: Arc<CoverageWatch>,
147 knots: Arc<KnotProxy>,
148 search: Arc<dyn SearchReader>,
149 resolver: Arc<RepoIdResolver>,
150 directory: Arc<Directory>,
151 ) -> Self {
152 Self {
153 records,
154 slingshot,
155 edges,
156 issue_states,
157 pull_statuses,
158 coverage,
159 knots,
160 search,
161 resolver,
162 directory,
163 limiter: None,
164 enrich_router: Arc::new(std::sync::OnceLock::new()),
165 }
166 }
167
168 pub fn with_limiter(mut self, limiter: Option<Arc<HeavyLimiter>>) -> Self {
169 self.limiter = limiter;
170 self
171 }
172
173 /// for internal xrpc dispatch [`enrich`]
174 pub fn self_router(&self) -> Router {
175 self.enrich_router
176 .get_or_init(|| router(self.clone()))
177 .clone()
178 }
179
180 pub(crate) fn heavy_permit(&self) -> Result<Option<HeavyPermit>, XrpcError> {
181 self.limiter.as_ref().map(|l| l.try_enter()).transpose()
182 }
183}
184
185pub fn router(state: AppState) -> Router {
186 Router::new()
187 .route("/xrpc/sh.tangled.repo.getRepo", get(get_repo))
188 .route("/xrpc/sh.tangled.repo.getRepos", get(get_repos))
189 .route(
190 "/xrpc/sh.tangled.repo.getRepoByRepoDid",
191 get(get_repo_by_repo_did),
192 )
193 .route(
194 "/xrpc/sh.tangled.repo.getReposByRepoDids",
195 get(get_repos_by_repo_dids),
196 )
197 .route("/xrpc/sh.tangled.repo.getRepoByName", get(get_repo_by_name))
198 .route("/xrpc/sh.tangled.actor.getProfile", get(get_profile))
199 .route("/xrpc/sh.tangled.actor.getProfiles", get(get_profiles))
200 .route("/xrpc/sh.tangled.repo.getIssue", get(get_issue))
201 .route("/xrpc/sh.tangled.repo.getIssues", get(get_issues))
202 .route("/xrpc/sh.tangled.repo.getPull", get(get_pull))
203 .route("/xrpc/sh.tangled.repo.getPulls", get(get_pulls))
204 .route("/xrpc/sh.tangled.feed.listStars", get(list_stars))
205 .route("/xrpc/sh.tangled.feed.countStars", get(count_stars))
206 .route("/xrpc/sh.tangled.feed.getStar", get(get_star))
207 .route("/xrpc/sh.tangled.graph.listFollows", get(list_follows))
208 .route("/xrpc/sh.tangled.graph.countFollows", get(count_follows))
209 .route("/xrpc/sh.tangled.graph.getFollow", get(get_follow))
210 .route("/xrpc/sh.tangled.repo.listIssues", get(list_issues))
211 .route("/xrpc/sh.tangled.repo.countIssues", get(count_issues))
212 .route("/xrpc/sh.tangled.repo.listPulls", get(list_pulls))
213 .route("/xrpc/sh.tangled.repo.countPulls", get(count_pulls))
214 .route(
215 "/xrpc/sh.tangled.feed.listComments",
216 get(list_feed_comments),
217 )
218 .route(
219 "/xrpc/sh.tangled.feed.countComments",
220 get(count_feed_comments),
221 )
222 .route("/xrpc/sh.tangled.feed.listReactions", get(list_reactions))
223 .route("/xrpc/sh.tangled.feed.countReactions", get(count_reactions))
224 .route("/xrpc/sh.tangled.git.listRefUpdates", get(list_ref_updates))
225 .route(
226 "/xrpc/sh.tangled.git.countRefUpdates",
227 get(count_ref_updates),
228 )
229 .route(
230 "/xrpc/sh.tangled.repo.listCollaborators",
231 get(list_collaborators),
232 )
233 .route(
234 "/xrpc/sh.tangled.repo.countCollaborators",
235 get(count_collaborators),
236 )
237 .route(
238 "/xrpc/sh.tangled.repo.issue.listStates",
239 get(list_issue_states),
240 )
241 .route(
242 "/xrpc/sh.tangled.repo.issue.countStates",
243 get(count_issue_states),
244 )
245 .route(
246 "/xrpc/sh.tangled.repo.pull.listStatuses",
247 get(list_pull_statuses),
248 )
249 .route(
250 "/xrpc/sh.tangled.repo.pull.countStatuses",
251 get(count_pull_statuses),
252 )
253 .route("/xrpc/sh.tangled.repo.listRepos", get(list_repos))
254 .route("/xrpc/sh.tangled.repo.countRepos", get(count_repos))
255 .route("/xrpc/sh.tangled.repo.countForks", get(count_forks))
256 .route("/xrpc/sh.tangled.knot.listKnots", get(list_knots))
257 .route("/xrpc/sh.tangled.knot.countKnots", get(count_knots))
258 .route("/xrpc/sh.tangled.spindle.listSpindles", get(list_spindles))
259 .route(
260 "/xrpc/sh.tangled.spindle.countSpindles",
261 get(count_spindles),
262 )
263 .route(
264 "/xrpc/sh.tangled.publicKey.getPublicKey",
265 get(get_public_key),
266 )
267 .route("/xrpc/sh.tangled.publicKey.listKeys", get(list_public_keys))
268 .route(
269 "/xrpc/sh.tangled.publicKey.countKeys",
270 get(count_public_keys),
271 )
272 .route("/xrpc/sh.tangled.graph.listVouches", get(list_vouches))
273 .route("/xrpc/sh.tangled.graph.countVouches", get(count_vouches))
274 .route("/xrpc/sh.tangled.feed.getTimeline", get(feed::get_timeline))
275 .route("/xrpc/sh.tangled.feed.listStarsBy", get(list_stars_by))
276 .route("/xrpc/sh.tangled.feed.countStarsBy", get(count_stars_by))
277 .route(
278 "/xrpc/sh.tangled.feed.listReactionsBy",
279 get(list_reactions_by),
280 )
281 .route(
282 "/xrpc/sh.tangled.feed.countReactionsBy",
283 get(count_reactions_by),
284 )
285 .route("/xrpc/sh.tangled.graph.listFollowsBy", get(list_follows_by))
286 .route(
287 "/xrpc/sh.tangled.graph.countFollowsBy",
288 get(count_follows_by),
289 )
290 .route("/xrpc/sh.tangled.graph.listVouchesBy", get(list_vouches_by))
291 .route(
292 "/xrpc/sh.tangled.graph.countVouchesBy",
293 get(count_vouches_by),
294 )
295 .route(
296 "/xrpc/sh.tangled.git.listRefUpdatesBy",
297 get(list_ref_updates_by),
298 )
299 .route(
300 "/xrpc/sh.tangled.git.countRefUpdatesBy",
301 get(count_ref_updates_by),
302 )
303 .route(
304 "/xrpc/sh.tangled.knot.listMembersBy",
305 get(list_knot_members_by),
306 )
307 .route(
308 "/xrpc/sh.tangled.knot.countMembersBy",
309 get(count_knot_members_by),
310 )
311 .route("/xrpc/sh.tangled.label.listOpsBy", get(list_label_ops_by))
312 .route("/xrpc/sh.tangled.label.countOpsBy", get(count_label_ops_by))
313 .route(
314 "/xrpc/sh.tangled.pipeline.listPipelinesBy",
315 get(list_pipelines_by),
316 )
317 .route(
318 "/xrpc/sh.tangled.pipeline.countPipelinesBy",
319 get(count_pipelines_by),
320 )
321 .route(
322 "/xrpc/sh.tangled.pipeline.listStatusesBy",
323 get(list_pipeline_statuses_by),
324 )
325 .route(
326 "/xrpc/sh.tangled.pipeline.countStatusesBy",
327 get(count_pipeline_statuses_by),
328 )
329 .route(
330 "/xrpc/sh.tangled.repo.listArtifactsBy",
331 get(list_artifacts_by),
332 )
333 .route(
334 "/xrpc/sh.tangled.repo.countArtifactsBy",
335 get(count_artifacts_by),
336 )
337 .route(
338 "/xrpc/sh.tangled.repo.listCollaboratorsBy",
339 get(list_collaborators_by),
340 )
341 .route(
342 "/xrpc/sh.tangled.repo.countCollaboratorsBy",
343 get(count_collaborators_by),
344 )
345 .route("/xrpc/sh.tangled.repo.listIssuesBy", get(list_issues_by))
346 .route("/xrpc/sh.tangled.repo.countIssuesBy", get(count_issues_by))
347 .route(
348 "/xrpc/sh.tangled.feed.listCommentsBy",
349 get(list_feed_comments_by),
350 )
351 .route(
352 "/xrpc/sh.tangled.feed.countCommentsBy",
353 get(count_feed_comments_by),
354 )
355 .route(
356 "/xrpc/sh.tangled.repo.issue.listStatesBy",
357 get(list_issue_states_by),
358 )
359 .route(
360 "/xrpc/sh.tangled.repo.issue.countStatesBy",
361 get(count_issue_states_by),
362 )
363 .route("/xrpc/sh.tangled.repo.listPullsBy", get(list_pulls_by))
364 .route("/xrpc/sh.tangled.repo.countPullsBy", get(count_pulls_by))
365 .route(
366 "/xrpc/sh.tangled.repo.pull.listStatusesBy",
367 get(list_pull_statuses_by),
368 )
369 .route(
370 "/xrpc/sh.tangled.repo.pull.countStatusesBy",
371 get(count_pull_statuses_by),
372 )
373 .route(
374 "/xrpc/sh.tangled.spindle.listMembersBy",
375 get(list_spindle_members_by),
376 )
377 .route(
378 "/xrpc/sh.tangled.spindle.countMembersBy",
379 get(count_spindle_members_by),
380 )
381 .route(
382 "/xrpc/sh.tangled.label.listDefinitions",
383 get(list_label_definitions),
384 )
385 .route(
386 "/xrpc/sh.tangled.label.countDefinitions",
387 get(count_label_definitions),
388 )
389 .route("/xrpc/sh.tangled.label.listOps", get(list_label_ops))
390 .route("/xrpc/sh.tangled.label.countOps", get(count_label_ops))
391 .route(
392 "/xrpc/sh.tangled.pipeline.listPipelines",
393 get(list_pipelines),
394 )
395 .route(
396 "/xrpc/sh.tangled.pipeline.countPipelines",
397 get(count_pipelines),
398 )
399 .route(
400 "/xrpc/sh.tangled.pipeline.listStatuses",
401 get(list_pipeline_statuses),
402 )
403 .route(
404 "/xrpc/sh.tangled.pipeline.countStatuses",
405 get(count_pipeline_statuses),
406 )
407 .route("/xrpc/sh.tangled.repo.listArtifacts", get(list_artifacts))
408 .route("/xrpc/sh.tangled.repo.countArtifacts", get(count_artifacts))
409 .route("/xrpc/sh.tangled.knot.listMembers", get(list_knot_members))
410 .route(
411 "/xrpc/sh.tangled.knot.countMembers",
412 get(count_knot_members),
413 )
414 .route(
415 "/xrpc/sh.tangled.spindle.listMembers",
416 get(list_spindle_members),
417 )
418 .route(
419 "/xrpc/sh.tangled.spindle.countMembers",
420 get(count_spindle_members),
421 )
422 .route("/xrpc/sh.tangled.string.getString", get(get_string))
423 .route("/xrpc/sh.tangled.string.listStrings", get(list_strings))
424 .route("/xrpc/sh.tangled.string.countStrings", get(count_strings))
425 .route("/xrpc/sh.tangled.search.query", get(search_query))
426 .route(
427 "/xrpc/sh.tangled.query.enrichResponse",
428 axum::routing::post(enrich::enrich),
429 )
430 .route("/xrpc/sh.tangled.bobbin.getCoverage", get(get_coverage))
431 .route(
432 "/xrpc/com.bad-example.identity.resolveMiniDoc",
433 get(resolve_mini_doc),
434 )
435 .merge(knot_proxied_routes())
436 .layer(
437 TraceLayer::new_for_http()
438 .make_span_with(DefaultMakeSpan::new().level(Level::INFO))
439 .on_request(())
440 .on_response(LatencyFreeTrace)
441 .on_failure(LatencyFreeTrace),
442 )
443 .with_state(state)
444}
445
446#[derive(Clone, Copy, Debug)]
447struct LatencyFreeTrace;
448
449impl<B> OnResponse<B> for LatencyFreeTrace {
450 fn on_response(self, response: &Response<B>, _latency: Duration, _span: &Span) {
451 tracing::event!(
452 target: "tower_http::trace::on_response",
453 Level::INFO,
454 status = response.status().as_u16(),
455 "request completed",
456 );
457 }
458}
459
460impl OnFailure<ServerErrorsFailureClass> for LatencyFreeTrace {
461 fn on_failure(&mut self, error: ServerErrorsFailureClass, _latency: Duration, _span: &Span) {
462 tracing::event!(
463 target: "tower_http::trace::on_failure",
464 Level::WARN,
465 error = %error,
466 "request failed",
467 );
468 }
469}
470
471const REPO_PROXIED_NSIDS: &[&str] = &[
472 "sh.tangled.repo.archive",
473 "sh.tangled.repo.blob",
474 "sh.tangled.repo.branch",
475 "sh.tangled.repo.branches",
476 "sh.tangled.repo.compare",
477 "sh.tangled.repo.describeRepo",
478 "sh.tangled.repo.diff",
479 "sh.tangled.repo.getDefaultBranch",
480 "sh.tangled.repo.languages",
481 "sh.tangled.repo.listSecrets",
482 "sh.tangled.repo.log",
483 "sh.tangled.repo.tag",
484 "sh.tangled.repo.tags",
485 "sh.tangled.repo.tree",
486];
487
488const KNOT_PROXIED_NSIDS: &[&str] = &[
489 "sh.tangled.owner",
490 "sh.tangled.knot.version",
491 "sh.tangled.knot.listKeys",
492];
493
494const PASSTHROUGH_HEADERS: &[&HeaderName] = &[
495 &CONTENT_TYPE,
496 &CONTENT_LENGTH,
497 &CONTENT_ENCODING,
498 &ETAG,
499 &CACHE_CONTROL,
500 &LAST_MODIFIED,
501 &CONTENT_DISPOSITION,
502 &ACCEPT_RANGES,
503 &CONTENT_RANGE,
504];
505
506const FORWARDED_REQUEST_HEADERS: &[&HeaderName] =
507 &[&RANGE, &IF_RANGE, &IF_NONE_MATCH, &IF_MODIFIED_SINCE];
508
509const KNOT_HOST_PARAM: &str = "knot";
510const REPO_PARAM: &str = "repo";
511
512type ProxyParams = Vec<(String, String)>;
513
514fn knot_proxied_routes() -> Router<AppState> {
515 let with_repo = register_proxied(Router::new(), REPO_PROXIED_NSIDS, proxy_repo_handler);
516 register_proxied(with_repo, KNOT_PROXIED_NSIDS, proxy_knot_handler)
517}
518
519fn register_proxied<H, Fut>(
520 router: Router<AppState>,
521 nsids: &[&'static str],
522 handler: H,
523) -> Router<AppState>
524where
525 H: Fn(AppState, HeaderMap, ProxyParams, Nsid<DefaultStr>) -> Fut
526 + Clone
527 + Send
528 + Sync
529 + 'static,
530 Fut: Future<Output = Result<Response, XrpcError>> + Send + 'static,
531{
532 nsids.iter().fold(router, |router, &nsid_lit| {
533 let handler = handler.clone();
534 let nsid = nsid_static(nsid_lit);
535 router.route(
536 &format!("/xrpc/{nsid_lit}"),
537 get(
538 move |State(state): State<AppState>,
539 headers: HeaderMap,
540 Query(params): Query<ProxyParams>| {
541 handler(state, headers, params, nsid.clone())
542 },
543 ),
544 )
545 })
546}
547
548#[derive(Clone, Debug)]
549pub enum SubjectQuery {
550 Did(Did<DefaultStr>),
551 Uri(AtUri<DefaultStr>),
552}
553
554impl<'de> Deserialize<'de> for SubjectQuery {
555 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
556 where
557 D: serde::Deserializer<'de>,
558 {
559 let raw = String::deserialize(deserializer)?;
560 if let Ok(did) = Did::<DefaultStr>::new_owned(&raw) {
561 return Ok(Self::Did(did));
562 }
563 AtUri::<DefaultStr>::new_owned(&raw)
564 .map(Self::Uri)
565 .map_err(serde::de::Error::custom)
566 }
567}
568
569#[derive(Clone, Debug, Eq, PartialEq)]
570pub struct ExpectedNsid {
571 canon: Nsid<DefaultStr>,
572 aliases: &'static [&'static str],
573}
574
575const FEED_COMMENT_LEGACY_ALIASES: &[&str] = &[
576 "sh.tangled.repo.issue.comment",
577 "sh.tangled.repo.pull.comment",
578];
579
580fn aliases_for(nsid: &str) -> &'static [&'static str] {
581 match nsid {
582 "sh.tangled.feed.comment" => FEED_COMMENT_LEGACY_ALIASES,
583 _ => &[],
584 }
585}
586
587impl ExpectedNsid {
588 pub fn new(nsid: Nsid<DefaultStr>) -> Self {
589 let aliases = aliases_for(nsid.as_ref());
590 Self {
591 canon: nsid,
592 aliases,
593 }
594 }
595
596 pub fn from_static(s: &'static str) -> Self {
597 let canon = nsid_static(s);
598 let aliases = aliases_for(s);
599 Self { canon, aliases }
600 }
601
602 pub fn as_nsid(&self) -> &Nsid<DefaultStr> {
603 &self.canon
604 }
605
606 pub fn as_str(&self) -> &str {
607 self.canon.as_ref()
608 }
609
610 fn accepts(&self, other: &str) -> bool {
611 other == self.canon.as_ref() || self.aliases.contains(&other)
612 }
613}
614
615#[derive(Debug, Deserialize)]
616struct GetRepoQuery {
617 repo: AtUri<DefaultStr>,
618}
619
620#[derive(Debug, Deserialize)]
621struct GetRepoByRepoDidQuery {
622 #[serde(rename = "repoDid")]
623 repo_did: Did<DefaultStr>,
624}
625
626#[derive(Debug, Deserialize)]
627struct GetRepoByNameQuery {
628 owner: Did<DefaultStr>,
629 name: String,
630}
631
632#[derive(Debug, Deserialize)]
633struct GetProfileQuery {
634 actor: AtUri<DefaultStr>,
635}
636
637#[derive(Debug, Deserialize)]
638struct GetIssueQuery {
639 issue: AtUri<DefaultStr>,
640}
641
642#[derive(Debug, Deserialize)]
643struct GetPullQuery {
644 pull: AtUri<DefaultStr>,
645}
646
647#[derive(Debug, Deserialize)]
648struct GetStringQuery {
649 string: AtUri<DefaultStr>,
650}
651
652#[derive(Debug, Deserialize)]
653struct GetPublicKeyQuery {
654 #[serde(rename = "publicKey")]
655 public_key: AtUri<DefaultStr>,
656}
657
658#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)]
659#[serde(rename_all = "lowercase")]
660enum Order {
661 Asc,
662 #[default]
663 Desc,
664}
665
666impl From<Order> for SortDir {
667 fn from(o: Order) -> Self {
668 match o {
669 Order::Asc => SortDir::Asc,
670 Order::Desc => SortDir::Desc,
671 }
672 }
673}
674
675#[derive(Debug, Deserialize)]
676struct TypedListQuery<F> {
677 subject: SubjectQuery,
678 cursor: Option<String>,
679 limit: Option<u32>,
680 #[serde(default)]
681 order: Order,
682 #[serde(flatten)]
683 filter: F,
684}
685
686impl<F> TypedListQuery<F> {
687 fn dir(&self) -> SortDir {
688 self.order.into()
689 }
690}
691
692#[derive(Debug, Deserialize)]
693struct CountQuery {
694 subject: SubjectQuery,
695}
696
697#[derive(Debug, Deserialize)]
698struct TypedCountQuery<F> {
699 subject: SubjectQuery,
700 #[serde(flatten)]
701 filter: F,
702}
703
704#[derive(Debug, Deserialize)]
705struct GetEdgeQuery {
706 actor: Did<DefaultStr>,
707 subject: SubjectQuery,
708}
709
710#[derive(Debug, Deserialize)]
711struct SearchQueryParams {
712 q: String,
713 nsid: Option<Nsid<DefaultStr>>,
714 author: Option<Did<DefaultStr>>,
715 repo: Option<Did<DefaultStr>>,
716 since: Option<String>,
717 until: Option<String>,
718 cursor: Option<String>,
719 limit: Option<u32>,
720}
721
722pub struct XrpcQuery<T>(pub T);
723
724impl<S, T> FromRequestParts<S> for XrpcQuery<T>
725where
726 S: Send + Sync,
727 T: serde::de::DeserializeOwned + Send + 'static,
728{
729 type Rejection = XrpcError;
730
731 async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
732 Query::<T>::from_request_parts(parts, state)
733 .await
734 .map(|Query(t)| Self(t))
735 .map_err(|rej: QueryRejection| XrpcError::InvalidParams(rej.body_text()))
736 }
737}
738
739#[derive(Debug, Error)]
740pub enum XrpcError {
741 #[error("invalid request: {0}")]
742 InvalidParams(String),
743 #[error("record not found")]
744 NotFound,
745 #[error("upstream unavailable: {0}")]
746 UpstreamUnavailable(String),
747 #[error("upstream gone: {0}")]
748 UpstreamGone(String),
749 #[error("invalid record: {0}")]
750 InvalidRecord(String),
751 #[error("internal: {0}")]
752 Internal(String),
753 #[error("overloaded, shedding under memory pressure")]
754 Overloaded,
755}
756
757impl XrpcError {
758 pub fn overloaded() -> Self {
759 Self::Overloaded
760 }
761}
762
763#[derive(Serialize)]
764struct ErrorBody {
765 error: &'static str,
766 message: String,
767}
768
769impl IntoResponse for XrpcError {
770 fn into_response(self) -> Response {
771 let (status, error) = match &self {
772 Self::InvalidParams(_) => (StatusCode::BAD_REQUEST, "InvalidRequest"),
773 Self::NotFound => (StatusCode::NOT_FOUND, "RecordNotFound"),
774 Self::UpstreamUnavailable(_) => (StatusCode::BAD_GATEWAY, "UpstreamFailed"),
775 Self::UpstreamGone(_) => (StatusCode::BAD_GATEWAY, "UpstreamGone"),
776 Self::InvalidRecord(_) => (StatusCode::BAD_GATEWAY, "InvalidRecord"),
777 Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "InternalError"),
778 Self::Overloaded => (StatusCode::SERVICE_UNAVAILABLE, "Overloaded"),
779 };
780 let body = ErrorBody {
781 error,
782 message: self.to_string(),
783 };
784 (status, Json(body)).into_response()
785 }
786}
787
788#[derive(Serialize)]
789#[serde(rename_all = "camelCase")]
790struct CoverageEnvelope {
791 ready: bool,
792 events_processed: u64,
793 last_cursor: u64,
794}
795
796impl From<Coverage> for CoverageEnvelope {
797 fn from(c: Coverage) -> Self {
798 Self {
799 ready: c.is_ready(),
800 events_processed: c.events_processed(),
801 last_cursor: c.last_cursor().raw(),
802 }
803 }
804}
805
806struct Deduped<T>(T);
807
808impl<T: Serialize> Serialize for Deduped<T> {
809 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
810 where
811 S: serde::Serializer,
812 {
813 serde_json::to_value(&self.0)
814 .map_err(serde::ser::Error::custom)?
815 .serialize(serializer)
816 }
817}
818
819#[derive(Serialize)]
820#[serde(rename_all = "camelCase")]
821struct RecordView<V> {
822 uri: AtUri<DefaultStr>,
823 cid: Option<Cid<DefaultStr>>,
824 value: V,
825}
826
827#[derive(Serialize)]
828#[serde(rename_all = "camelCase")]
829struct StatefulItem<V> {
830 #[serde(flatten)]
831 view: RecordView<V>,
832 state: &'static str,
833 #[serde(skip_serializing_if = "Option::is_none")]
834 state_updated_at: Option<String>,
835 comment_count: u64,
836}
837
838fn format_micros(micros: u64) -> String {
839 let signed = i64::try_from(micros).ok();
840 let rfc = signed
841 .and_then(chrono::DateTime::<chrono::Utc>::from_timestamp_micros)
842 .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Micros, true));
843 rfc.unwrap_or_else(|| micros.to_string())
844}
845
846pub(crate) fn source_authority_did(source: &AtUri<DefaultStr>) -> Option<Did<DefaultStr>> {
847 match source.authority() {
848 AtIdentifier::Did(d) => Some(d.clone().into_static()),
849 AtIdentifier::Handle(_) => None,
850 }
851}
852
853fn enrich_issue_view(
854 state: &AppState,
855 view: RecordView<Issue<DefaultStr>>,
856) -> StatefulItem<Issue<DefaultStr>> {
857 let issue_author = source_authority_did(&view.uri);
858 let repo_did = view.value.repo.clone();
859 enrich_view(
860 &state.edges,
861 nsid_static("sh.tangled.feed.comment"),
862 &state.issue_states,
863 view,
864 move |src| accept_state_source(src, issue_author.as_ref(), &repo_did),
865 )
866}
867
868fn enrich_pull_view(
869 state: &AppState,
870 view: RecordView<Pull<DefaultStr>>,
871) -> StatefulItem<Pull<DefaultStr>> {
872 let pull_author = source_authority_did(&view.uri);
873 let target_repo = view.value.target.repo.clone();
874 enrich_view(
875 &state.edges,
876 nsid_static("sh.tangled.feed.comment"),
877 &state.pull_statuses,
878 view,
879 move |src| accept_state_source(src, pull_author.as_ref(), &target_repo),
880 )
881}
882
883pub(crate) fn accept_state_source(
884 source: &AtUri<DefaultStr>,
885 entity_author: Option<&Did<DefaultStr>>,
886 repo_owner: &Did<DefaultStr>,
887) -> bool {
888 let Some(src) = source_authority_did(source) else {
889 return false;
890 };
891 Some(&src) == entity_author || &src == repo_owner
892}
893
894fn enrich_view<V, K, F>(
895 edges: &EdgeStore,
896 comment_nsid: Nsid<DefaultStr>,
897 states: &StateIndex<K>,
898 view: RecordView<V>,
899 accept: F,
900) -> StatefulItem<V>
901where
902 K: StateKind + Default,
903 F: Fn(&AtUri<DefaultStr>) -> bool,
904{
905 let comment_count = edges.count(&EdgeKey::new(
906 comment_nsid,
907 SubjectRef::Uri(view.uri.clone()),
908 ));
909 let (state, state_updated_at) = states
910 .latest_by(&view.uri, accept)
911 .map_or((K::default().wire(), None), |(kind, micros)| {
912 (kind.wire(), Some(format_micros(micros)))
913 });
914 StatefulItem {
915 view,
916 state,
917 state_updated_at,
918 comment_count,
919 }
920}
921
922#[derive(Clone, Copy, Debug, Serialize)]
923#[serde(rename_all = "camelCase")]
924struct CountResponse {
925 count: u64,
926 distinct_authors: u64,
927}
928
929#[derive(Serialize)]
930#[serde(rename_all = "camelCase")]
931struct EdgeUriResponse {
932 uri: AtUri<DefaultStr>,
933}
934
935#[derive(Serialize)]
936#[serde(rename_all = "camelCase")]
937struct SearchHitView {
938 uri: AtUri<DefaultStr>,
939 cid: Option<Cid<DefaultStr>>,
940 nsid: Nsid<DefaultStr>,
941 score: f32,
942 value: SearchableRecord,
943}
944
945fn map_slingshot(err: SlingshotError) -> XrpcError {
946 use SlingshotError as E;
947 match err {
948 E::NotFound => XrpcError::NotFound,
949 e @ (E::Decode(_)
950 | E::MissingField(_)
951 | E::InvalidAtUri(_)
952 | E::InvalidCid(_)
953 | E::UriMismatch { .. }) => XrpcError::InvalidRecord(e.to_string()),
954 e @ (E::Network(_)
955 | E::Build(_)
956 | E::Upstream(_)
957 | E::BodyTooLarge { .. }
958 | E::BadScheme(_)) => XrpcError::UpstreamUnavailable(e.to_string()),
959 }
960}
961
962fn parse_uri(raw: &str) -> Result<AtUri<DefaultStr>, XrpcError> {
963 AtUri::<DefaultStr>::new_owned(raw).map_err(|e| XrpcError::InvalidParams(format!("uri: {e}")))
964}
965
966#[derive(Clone, Copy, Debug, Eq, PartialEq)]
967pub enum SubjectShape {
968 BareDid,
969 Collection(&'static str),
970 BareDidOrOneOfCollections(&'static [&'static str]),
971 OneOfCollections(&'static [&'static str]),
972 AnyAtUri,
973}
974
975pub trait HasSubject {
976 const SHAPE: SubjectShape;
977}
978
979pub trait MirrorOf {
980 type Record: XrpcResp;
981 const EDGE_KIND: &'static str;
982 const SHAPE: SubjectShape;
983}
984
985macro_rules! edge_kinds {
986 ($($collection:literal => $record:ty, $shape:expr $(, mirror $by:ident)? ;)*) => {
987 $(
988 impl HasSubject for $record {
989 const SHAPE: SubjectShape = $shape;
990 }
991 )*
992 $($(
993 pub struct $by;
994 impl MirrorOf for $by {
995 type Record = $record;
996 const EDGE_KIND: &'static str = concat!($collection, ".by");
997 const SHAPE: SubjectShape = SubjectShape::BareDid;
998 }
999 )?)*
1000
1001 pub(crate) fn subject_shape(collection: &str) -> Option<(&'static str, SubjectShape)> {
1002 Some(match collection {
1003 $($collection => ($collection, <$record as HasSubject>::SHAPE),)*
1004 _ => return None,
1005 })
1006 }
1007
1008 pub(crate) fn mirror_kind(collection: &str) -> Option<&'static str> {
1009 Some(match collection {
1010 $($($collection => <$by as MirrorOf>::EDGE_KIND,)?)*
1011 _ => return None,
1012 })
1013 }
1014 };
1015}
1016
1017edge_kinds! {
1018 "sh.tangled.feed.star" => StarRecord, SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.string"]), mirror StarBy;
1019 "sh.tangled.feed.comment" => FeedCommentRecord, SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull", "sh.tangled.string"]), mirror FeedCommentBy;
1020 "sh.tangled.feed.reaction" => ReactionRecord, SubjectShape::AnyAtUri, mirror ReactionBy;
1021 "sh.tangled.graph.follow" => FollowRecord, SubjectShape::BareDid, mirror FollowBy;
1022 "sh.tangled.graph.vouch" => VouchRecord, SubjectShape::BareDid, mirror VouchBy;
1023 "sh.tangled.git.refUpdate" => RefUpdateRecord, SubjectShape::BareDid, mirror RefUpdateBy;
1024 "sh.tangled.knot" => KnotRecord, SubjectShape::BareDid;
1025 "sh.tangled.knot.member" => KnotMemberRecord, SubjectShape::BareDid, mirror KnotMemberBy;
1026 "sh.tangled.label.definition" => LabelDefinitionRecord, SubjectShape::BareDid;
1027 "sh.tangled.label.op" => LabelOpRecord, SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull"]), mirror LabelOpBy;
1028 "sh.tangled.pipeline" => PipelineRecord, SubjectShape::BareDid, mirror PipelineBy;
1029 "sh.tangled.pipeline.status" => PipelineStatusRecord, SubjectShape::Collection("sh.tangled.pipeline"), mirror PipelineStatusBy;
1030 "sh.tangled.publicKey" => PublicKeyRecord, SubjectShape::BareDid;
1031 "sh.tangled.repo" => RepoRecord, SubjectShape::BareDid;
1032 "sh.tangled.repo.artifact" => ArtifactRecord, SubjectShape::BareDid, mirror ArtifactBy;
1033 "sh.tangled.repo.collaborator" => CollaboratorRecord, SubjectShape::BareDid, mirror CollaboratorBy;
1034 "sh.tangled.repo.issue" => IssueRecord, SubjectShape::BareDid, mirror IssueBy;
1035 "sh.tangled.repo.issue.state" => IssueStateRecord, SubjectShape::Collection("sh.tangled.repo.issue"), mirror IssueStateBy;
1036 "sh.tangled.repo.pull" => PullRecord, SubjectShape::BareDid, mirror PullBy;
1037 "sh.tangled.repo.pull.status" => PullStatusRecord, SubjectShape::Collection("sh.tangled.repo.pull"), mirror PullStatusBy;
1038 "sh.tangled.spindle" => SpindleRecord, SubjectShape::BareDid;
1039 "sh.tangled.spindle.member" => SpindleMemberRecord, SubjectShape::BareDid, mirror SpindleMemberBy;
1040 "sh.tangled.string" => TangledStringRecord, SubjectShape::BareDid;
1041}
1042
1043// the fork edge is not a collection so it is not in the table above
1044const FORK_SUBJECT_SHAPE: SubjectShape = SubjectShape::BareDid;
1045
1046fn parse_subject(raw: &SubjectQuery, shape: SubjectShape) -> Result<SubjectRef, XrpcError> {
1047 let uri = match raw {
1048 SubjectQuery::Did(did) => {
1049 return match shape {
1050 SubjectShape::BareDid | SubjectShape::BareDidOrOneOfCollections(_) => {
1051 Ok(SubjectRef::Did(did.clone()))
1052 }
1053 SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!(
1054 "subject must be at://<did>/{expected}/<rkey>, got bare did"
1055 ))),
1056 SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
1057 "subject must be at://<did>/<nsid>/<rkey> with nsid in [{}], got bare did",
1058 allowed.join(", "),
1059 ))),
1060 SubjectShape::AnyAtUri => Err(XrpcError::InvalidParams(
1061 "subject must be at-uri form, got bare did".into(),
1062 )),
1063 };
1064 }
1065 SubjectQuery::Uri(uri) => uri,
1066 };
1067 if matches!(uri.authority(), AtIdentifier::Handle(_)) {
1068 return Err(XrpcError::InvalidParams(
1069 "subject authority must be a did, not a handle".into(),
1070 ));
1071 }
1072 let Some(collection) = uri.collection() else {
1073 return Err(XrpcError::InvalidParams(
1074 "subject must be a bare did or full at://<did>/<nsid>/<rkey>".into(),
1075 ));
1076 };
1077 let c = collection.as_ref();
1078 match shape {
1079 SubjectShape::BareDid => Err(XrpcError::InvalidParams(format!(
1080 "subject must be a bare did, got at-uri with collection {c}"
1081 ))),
1082 SubjectShape::Collection(expected) if c == expected => {
1083 require_rkey(uri, expected)?;
1084 Ok(SubjectRef::Uri(uri.clone()))
1085 }
1086 SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!(
1087 "subject must be at://<did>/{expected}/<rkey>, got collection {c}"
1088 ))),
1089 SubjectShape::OneOfCollections(allowed) if allowed.contains(&c) => {
1090 require_rkey(uri, c)?;
1091 Ok(SubjectRef::Uri(uri.clone()))
1092 }
1093 SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
1094 "subject must be at://<did>/<nsid>/<rkey> with nsid in [{}], got collection {c}",
1095 allowed.join(", "),
1096 ))),
1097 SubjectShape::BareDidOrOneOfCollections(allowed) if allowed.contains(&c) => {
1098 require_rkey(uri, c)?;
1099 Ok(SubjectRef::Uri(uri.clone()))
1100 }
1101 SubjectShape::BareDidOrOneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
1102 "subject must be a bare did or at://<did>/<nsid>/<rkey> with nsid in [{}], got collection {c}",
1103 allowed.join(", "),
1104 ))),
1105 SubjectShape::AnyAtUri => Ok(SubjectRef::Uri(uri.clone())),
1106 }
1107}
1108
1109fn require_rkey(uri: &AtUri<DefaultStr>, expected: &str) -> Result<(), XrpcError> {
1110 uri.rkey().map(|_| ()).ok_or_else(|| {
1111 XrpcError::InvalidParams(format!(
1112 "subject must be at://<did>/{expected}/<rkey>; missing rkey"
1113 ))
1114 })
1115}
1116
1117pub(crate) fn parse_cursor(raw: Option<&str>) -> Result<PageCursor, XrpcError> {
1118 PageCursor::from_token(raw)
1119 .map_err(|e: CursorParseError| XrpcError::InvalidParams(format!("cursor: {e}")))
1120}
1121
1122pub(crate) fn parse_limit(raw: Option<u32>) -> Result<PageLimit, XrpcError> {
1123 PageLimit::new(raw.unwrap_or(DEFAULT_LIMIT))
1124 .map_err(|e| XrpcError::InvalidParams(format!("limit: {e}")))
1125}
1126
1127pub(crate) fn at_uri_owned_by(uri: &AtUri<DefaultStr>, author: &Did<DefaultStr>) -> bool {
1128 match uri.authority() {
1129 AtIdentifier::Did(d) => d.as_ref() == author.as_ref(),
1130 AtIdentifier::Handle(_) => false,
1131 }
1132}
1133
1134async fn resolve_for_view(
1135 state: &AppState,
1136 expected_nsid: &Nsid<DefaultStr>,
1137 uri: AtUri<DefaultStr>,
1138) -> Result<Arc<RecordBody>, XrpcError> {
1139 let raw = uri.as_ref().to_owned();
1140 resolve(state, ExpectedNsid::new(expected_nsid.clone()), uri)
1141 .await
1142 .map(|(body, _did)| body)
1143 .map_err(|e| match e {
1144 XrpcError::NotFound => XrpcError::UpstreamGone(raw),
1145 other => other,
1146 })
1147}
1148
1149async fn resolve(
1150 state: &AppState,
1151 expected: ExpectedNsid,
1152 uri: AtUri<DefaultStr>,
1153) -> Result<(Arc<RecordBody>, Did<DefaultStr>), XrpcError> {
1154 let collection = uri
1155 .collection()
1156 .ok_or_else(|| XrpcError::InvalidParams("uri missing collection".into()))?;
1157 if !expected.accepts(collection.as_ref()) {
1158 return Err(XrpcError::InvalidParams(format!(
1159 "collection mismatch: expected {}, got {}",
1160 expected.as_str(),
1161 collection.as_ref()
1162 )));
1163 }
1164 let rkey = uri
1165 .rkey()
1166 .ok_or_else(|| XrpcError::InvalidParams("uri missing rkey".into()))?;
1167 let did_ref = match uri.authority() {
1168 AtIdentifier::Did(d) => d,
1169 AtIdentifier::Handle(_) => {
1170 return Err(XrpcError::InvalidParams(
1171 "uri authority must be a did, not a handle".into(),
1172 ));
1173 }
1174 };
1175 let did: Did<DefaultStr> = did_ref.clone().into_static();
1176
1177 if let Some(hit) = state.records.get(&uri) {
1178 return Ok((hit, did));
1179 }
1180 let body = state
1181 .slingshot
1182 .get_record(&did_ref, &collection, &rkey)
1183 .await
1184 .map_err(map_slingshot)?;
1185 verify_type_tag(&body, &expected)?;
1186 state.records.put(uri, body.clone());
1187 Ok((body, did))
1188}
1189
1190#[derive(Deserialize)]
1191struct TypeTag<'a> {
1192 #[serde(rename = "$type", borrow)]
1193 ty: &'a str,
1194}
1195
1196fn verify_type_tag(body: &RecordBody, expected: &ExpectedNsid) -> Result<(), XrpcError> {
1197 let bytes = body.value.as_ref();
1198 let ty: std::borrow::Cow<'_, str> = match serde_json::from_slice::<TypeTag>(bytes) {
1199 Ok(t) => std::borrow::Cow::Borrowed(t.ty),
1200 Err(_) => {
1201 let value: serde_json::Value = serde_json::from_slice(bytes)
1202 .map_err(|e| XrpcError::InvalidRecord(format!("$type peek: {e}")))?;
1203 value
1204 .as_object()
1205 .and_then(|m| m.get("$type"))
1206 .and_then(|v| v.as_str())
1207 .map(|s| std::borrow::Cow::Owned(s.to_owned()))
1208 .ok_or_else(|| XrpcError::InvalidRecord("$type peek: missing $type field".into()))?
1209 }
1210 };
1211 if !expected.accepts(ty.as_ref()) {
1212 return Err(XrpcError::InvalidRecord(format!(
1213 "$type mismatch: expected {}, got {}",
1214 expected.as_str(),
1215 ty
1216 )));
1217 }
1218 Ok(())
1219}
1220
1221fn wire_type_nsid(bytes: &[u8]) -> Option<Nsid<DefaultStr>> {
1222 let ty = serde_json::from_slice::<TypeTag>(bytes).ok()?.ty;
1223 Nsid::<DefaultStr>::new_owned(ty).ok()
1224}
1225
1226async fn deserialize_or_upgrade<V>(
1227 state: &AppState,
1228 nsid: &Nsid<DefaultStr>,
1229 bytes: &[u8],
1230) -> Result<V, XrpcError>
1231where
1232 V: serde::de::DeserializeOwned,
1233{
1234 match serde_json::from_slice::<V>(bytes) {
1235 Ok(v) => Ok(v),
1236 Err(canon_err) => {
1237 let normalized = normalize_record_fields(bytes);
1238 let working: &[u8] = normalized.as_deref().unwrap_or(bytes);
1239 if normalized.is_some()
1240 && let Ok(v) = serde_json::from_slice::<V>(working)
1241 {
1242 return Ok(v);
1243 }
1244 let scrubbed = scrub_record_bytes(nsid, working);
1245 let retry_bytes: &[u8] = scrubbed.as_deref().unwrap_or(working);
1246 if scrubbed.is_some()
1247 && let Ok(v) = serde_json::from_slice::<V>(retry_bytes)
1248 {
1249 return Ok(v);
1250 }
1251 let wire_nsid = wire_type_nsid(retry_bytes).unwrap_or_else(|| nsid.clone());
1252 match upgrade_wire_bytes(&wire_nsid, retry_bytes, &state.resolver).await {
1253 Ok(canon_bytes) => serde_json::from_slice(&canon_bytes)
1254 .map_err(|e| XrpcError::InvalidRecord(e.to_string())),
1255 Err(_) => Err(XrpcError::InvalidRecord(canon_err.to_string())),
1256 }
1257 }
1258 }
1259}
1260
1261async fn fetch_from_uri<R, V>(
1262 state: &AppState,
1263 uri: AtUri<DefaultStr>,
1264) -> Result<(Arc<RecordBody>, V), XrpcError>
1265where
1266 R: XrpcResp,
1267 V: serde::de::DeserializeOwned + NormalizeRepoRefs,
1268{
1269 let raw = uri.as_str().to_owned();
1270 let nsid = nsid_static(R::NSID);
1271 let (body, _did) = resolve(state, ExpectedNsid::new(nsid.clone()), uri).await?;
1272 let value: V = deserialize_or_upgrade(state, &nsid, &body.value).await?;
1273 let value = value
1274 .normalize(&state.resolver)
1275 .await
1276 .ok_or(XrpcError::UpstreamGone(raw))?;
1277 Ok((body, value))
1278}
1279
1280pub(crate) async fn fetch<R, V>(
1281 state: &AppState,
1282 uri: &AtUri<DefaultStr>,
1283) -> Result<(Arc<RecordBody>, V), XrpcError>
1284where
1285 R: XrpcResp,
1286 V: serde::de::DeserializeOwned + NormalizeRepoRefs,
1287{
1288 fetch_from_uri::<R, V>(state, uri.clone()).await
1289}
1290
1291async fn get_repo(
1292 State(state): State<AppState>,
1293 XrpcQuery(q): XrpcQuery<GetRepoQuery>,
1294) -> Result<Json<Deduped<RepoGetRecordOutput<DefaultStr>>>, XrpcError> {
1295 let (body, value) = fetch::<RepoRecord, Repo<DefaultStr>>(&state, &q.repo).await?;
1296 Ok(Json(Deduped(RepoGetRecordOutput {
1297 cid: Some(body.cid.clone()),
1298 uri: body.uri.clone(),
1299 value,
1300 })))
1301}
1302
1303async fn get_repo_by_repo_did(
1304 State(state): State<AppState>,
1305 XrpcQuery(q): XrpcQuery<GetRepoByRepoDidQuery>,
1306) -> Result<Json<Deduped<RepoGetRecordOutput<DefaultStr>>>, XrpcError> {
1307 let ident = state
1308 .resolver
1309 .lookup_by_repo_did(&q.repo_did)
1310 .await
1311 .ok_or(XrpcError::NotFound)?;
1312 let uri = AtUri::<DefaultStr>::from_parts_owned(
1313 ident.owner.as_str(),
1314 RepoRecord::NSID,
1315 ident.rkey.as_str(),
1316 )
1317 .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail");
1318 let (body, value) = fetch_from_uri::<RepoRecord, Repo<DefaultStr>>(&state, uri).await?;
1319 Ok(Json(Deduped(RepoGetRecordOutput {
1320 cid: Some(body.cid.clone()),
1321 uri: body.uri.clone(),
1322 value,
1323 })))
1324}
1325
1326async fn get_repo_by_name(
1327 State(state): State<AppState>,
1328 XrpcQuery(q): XrpcQuery<GetRepoByNameQuery>,
1329) -> Result<Json<Deduped<RepoGetRecordOutput<DefaultStr>>>, XrpcError> {
1330 let ident = state
1331 .resolver
1332 .lookup_by_name(&q.owner, &q.name)
1333 .await
1334 .ok_or(XrpcError::NotFound)?;
1335 let uri = AtUri::<DefaultStr>::from_parts_owned(
1336 ident.owner.as_str(),
1337 RepoRecord::NSID,
1338 ident.rkey.as_str(),
1339 )
1340 .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail");
1341 let (body, value) = fetch_from_uri::<RepoRecord, Repo<DefaultStr>>(&state, uri).await?;
1342 Ok(Json(Deduped(RepoGetRecordOutput {
1343 cid: Some(body.cid.clone()),
1344 uri: body.uri.clone(),
1345 value,
1346 })))
1347}
1348
1349async fn get_profile(
1350 State(state): State<AppState>,
1351 XrpcQuery(q): XrpcQuery<GetProfileQuery>,
1352) -> Result<Json<Deduped<ProfileGetRecordOutput<DefaultStr>>>, XrpcError> {
1353 let (body, value) = fetch::<ProfileRecord, Profile<DefaultStr>>(&state, &q.actor).await?;
1354 Ok(Json(Deduped(ProfileGetRecordOutput {
1355 cid: Some(body.cid.clone()),
1356 uri: body.uri.clone(),
1357 value,
1358 })))
1359}
1360
1361async fn get_issue(
1362 State(state): State<AppState>,
1363 XrpcQuery(q): XrpcQuery<GetIssueQuery>,
1364) -> Result<Json<Deduped<IssueGetRecordOutput<DefaultStr>>>, XrpcError> {
1365 let (body, value) = fetch::<IssueRecord, Issue<DefaultStr>>(&state, &q.issue).await?;
1366 Ok(Json(Deduped(IssueGetRecordOutput {
1367 cid: Some(body.cid.clone()),
1368 uri: body.uri.clone(),
1369 value,
1370 })))
1371}
1372
1373async fn get_pull(
1374 State(state): State<AppState>,
1375 XrpcQuery(q): XrpcQuery<GetPullQuery>,
1376) -> Result<Json<Deduped<PullGetRecordOutput<DefaultStr>>>, XrpcError> {
1377 let (body, value) = fetch::<PullRecord, Pull<DefaultStr>>(&state, &q.pull).await?;
1378 Ok(Json(Deduped(PullGetRecordOutput {
1379 cid: Some(body.cid.clone()),
1380 uri: body.uri.clone(),
1381 value,
1382 })))
1383}
1384
1385async fn get_string(
1386 State(state): State<AppState>,
1387 XrpcQuery(q): XrpcQuery<GetStringQuery>,
1388) -> Result<Json<Deduped<TangledStringGetRecordOutput<DefaultStr>>>, XrpcError> {
1389 let (body, value) =
1390 fetch::<TangledStringRecord, TangledString<DefaultStr>>(&state, &q.string).await?;
1391 Ok(Json(Deduped(TangledStringGetRecordOutput {
1392 cid: Some(body.cid.clone()),
1393 uri: body.uri.clone(),
1394 value,
1395 })))
1396}
1397
1398async fn get_public_key(
1399 State(state): State<AppState>,
1400 XrpcQuery(q): XrpcQuery<GetPublicKeyQuery>,
1401) -> Result<Json<Deduped<PublicKeyGetRecordOutput<DefaultStr>>>, XrpcError> {
1402 let (body, value) =
1403 fetch::<PublicKeyRecord, PublicKey<DefaultStr>>(&state, &q.public_key).await?;
1404 Ok(Json(Deduped(PublicKeyGetRecordOutput {
1405 cid: Some(body.cid.clone()),
1406 uri: body.uri.clone(),
1407 value,
1408 })))
1409}
1410
1411async fn get_repos(
1412 State(state): State<AppState>,
1413 RawQuery(query): RawQuery,
1414) -> Result<Response, XrpcError> {
1415 let uris = collect_repeated(query.as_deref(), BULK_REPOS_KEY);
1416 bulk_fetch::<RepoRecord, Repo<DefaultStr>>(&state, uris).await
1417}
1418
1419async fn get_repos_by_repo_dids(
1420 State(state): State<AppState>,
1421 RawQuery(query): RawQuery,
1422) -> Result<Response, XrpcError> {
1423 let dids = collect_repeated(query.as_deref(), BULK_REPO_DIDS_KEY);
1424 if dids.is_empty() {
1425 return Err(XrpcError::InvalidParams("at least one did required".into()));
1426 }
1427 if dids.len() > BULK_LIMIT {
1428 return Err(XrpcError::InvalidParams(format!(
1429 "at most {BULK_LIMIT} dids per request"
1430 )));
1431 }
1432 let dids = dids
1433 .iter()
1434 .map(|s| {
1435 Did::<DefaultStr>::new_owned(s)
1436 .map_err(|_| XrpcError::InvalidParams(format!("invalid did: {s}")))
1437 })
1438 .collect::<Result<Vec<_>, _>>()?;
1439 let mut uris: Vec<AtUri<DefaultStr>> = Vec::new();
1440 for did in &dids {
1441 if let Some(ident) = state.resolver.lookup_by_repo_did(did).await {
1442 uris.push(
1443 AtUri::<DefaultStr>::from_parts_owned(
1444 ident.owner.as_str(),
1445 RepoRecord::NSID,
1446 ident.rkey.as_str(),
1447 )
1448 .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail"),
1449 );
1450 }
1451 }
1452 bulk_stream::<RepoRecord, Repo<DefaultStr>>(&state, uris)
1453}
1454
1455async fn get_profiles(
1456 State(state): State<AppState>,
1457 RawQuery(query): RawQuery,
1458) -> Result<Response, XrpcError> {
1459 let uris = collect_repeated(query.as_deref(), BULK_PROFILES_KEY);
1460 bulk_fetch::<ProfileRecord, Profile<DefaultStr>>(&state, uris).await
1461}
1462
1463async fn get_issues(
1464 State(state): State<AppState>,
1465 RawQuery(query): RawQuery,
1466) -> Result<Response, XrpcError> {
1467 let uris = collect_repeated(query.as_deref(), BULK_ISSUES_KEY);
1468 bulk_fetch::<IssueRecord, Issue<DefaultStr>>(&state, uris).await
1469}
1470
1471async fn get_pulls(
1472 State(state): State<AppState>,
1473 RawQuery(query): RawQuery,
1474) -> Result<Response, XrpcError> {
1475 let uris = collect_repeated(query.as_deref(), BULK_PULLS_KEY);
1476 bulk_fetch::<PullRecord, Pull<DefaultStr>>(&state, uris).await
1477}
1478
1479const BULK_REPOS_KEY: &str = "repos";
1480const BULK_REPO_DIDS_KEY: &str = "dids";
1481const BULK_PROFILES_KEY: &str = "actors";
1482const BULK_ISSUES_KEY: &str = "issues";
1483const BULK_PULLS_KEY: &str = "pulls";
1484const BULK_LIMIT: usize = 50;
1485
1486fn collect_repeated(query: Option<&str>, key: &str) -> Vec<String> {
1487 let Some(q) = query else {
1488 return Vec::new();
1489 };
1490 form_urlencoded::parse(q.as_bytes())
1491 .filter_map(|(k, v)| (k == key).then(|| v.into_owned()))
1492 .collect()
1493}
1494
1495async fn hydrate_record_view<V>(
1496 state: &AppState,
1497 nsid: &Nsid<DefaultStr>,
1498 uri: AtUri<DefaultStr>,
1499 sort_micros: u64,
1500) -> Result<Option<RecordView<V>>, XrpcError>
1501where
1502 V: serde::de::DeserializeOwned + NormalizeRepoRefs,
1503{
1504 if let Some(source) = decode_knot_owned_source(&uri) {
1505 return synthesize_knot_owned_view::<V>(state, uri, source, sort_micros).await;
1506 }
1507 let body = resolve_for_view(state, nsid, uri).await?;
1508 let value: V = deserialize_or_upgrade::<V>(state, nsid, &body.value).await?;
1509 let Some(value) = value.normalize(&state.resolver).await else {
1510 return Ok(None);
1511 };
1512 Ok(Some(RecordView {
1513 uri: body.uri.clone(),
1514 cid: Some(body.cid.clone()),
1515 value,
1516 }))
1517}
1518
1519async fn synthesize_knot_owned_view<V>(
1520 state: &AppState,
1521 uri: AtUri<DefaultStr>,
1522 source: KnotOwnedSource,
1523 sort_micros: u64,
1524) -> Result<Option<RecordView<V>>, XrpcError>
1525where
1526 V: serde::de::DeserializeOwned + NormalizeRepoRefs,
1527{
1528 let Some(body) = synth_knot_owned_value(source, sort_micros) else {
1529 return Ok(None);
1530 };
1531 let Ok(value) = serde_json::from_value::<V>(body) else {
1532 return Ok(None);
1533 };
1534 let Some(value) = value.normalize(&state.resolver).await else {
1535 return Ok(None);
1536 };
1537 Ok(Some(RecordView {
1538 uri,
1539 cid: None,
1540 value,
1541 }))
1542}
1543
1544fn synth_knot_owned_value(source: KnotOwnedSource, sort_micros: u64) -> Option<serde_json::Value> {
1545 let created_at = micros_to_rfc3339(sort_micros)?;
1546 match source {
1547 KnotOwnedSource::Member { knot, subject } => Some(serde_json::json!({
1548 "domain": knot_did_host(&knot)?,
1549 "subject": subject.as_ref(),
1550 "createdAt": created_at,
1551 })),
1552 KnotOwnedSource::Collaborator { repo, subject } => Some(serde_json::json!({
1553 "repo": repo.as_ref(),
1554 "subject": subject.as_ref(),
1555 "createdAt": created_at,
1556 })),
1557 }
1558}
1559
1560fn micros_to_rfc3339(micros: u64) -> Option<String> {
1561 let micros = i64::try_from(micros).ok()?;
1562 chrono::DateTime::from_timestamp_micros(micros)
1563 .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Micros, true))
1564}
1565
1566#[derive(Clone, Copy)]
1567enum HitProvenance {
1568 ClientSupplied,
1569 Indexed,
1570}
1571
1572fn is_index_evictable(err: &XrpcError) -> bool {
1573 matches!(
1574 err,
1575 XrpcError::NotFound
1576 | XrpcError::UpstreamGone(_)
1577 | XrpcError::InvalidRecord(_)
1578 | XrpcError::InvalidParams(_)
1579 )
1580}
1581
1582fn drop_unhydratable<V>(
1583 provenance: HitProvenance,
1584 nsid: &Nsid<DefaultStr>,
1585 uri: &AtUri<DefaultStr>,
1586 result: Result<Option<V>, XrpcError>,
1587) -> Result<Option<V>, XrpcError> {
1588 match result {
1589 Ok(view) => Ok(view),
1590 Err(err @ (XrpcError::NotFound | XrpcError::UpstreamGone(_))) => {
1591 tracing::debug!(
1592 uri = %uri,
1593 nsid = %nsid.as_ref(),
1594 error = %err,
1595 "dropping gone hit during hydration",
1596 );
1597 Ok(None)
1598 }
1599 Err(err @ XrpcError::UpstreamUnavailable(_)) => {
1600 tracing::warn!(
1601 uri = %uri,
1602 nsid = %nsid.as_ref(),
1603 error = %err,
1604 "dropping hit, upstream unavailable during hydration",
1605 );
1606 Ok(None)
1607 }
1608 Err(err @ XrpcError::InvalidRecord(_)) => {
1609 tracing::warn!(
1610 uri = %uri,
1611 nsid = %nsid.as_ref(),
1612 error = %err,
1613 "dropping invalid hit during hydration",
1614 );
1615 Ok(None)
1616 }
1617 Err(err @ XrpcError::InvalidParams(_)) => match provenance {
1618 HitProvenance::ClientSupplied => Err(err),
1619 HitProvenance::Indexed => {
1620 tracing::warn!(
1621 uri = %uri,
1622 nsid = %nsid.as_ref(),
1623 error = %err,
1624 "dropping malformed indexed hit during hydration",
1625 );
1626 Ok(None)
1627 }
1628 },
1629 Err(err @ (XrpcError::Internal(_) | XrpcError::Overloaded)) => Err(err),
1630 }
1631}
1632
1633fn hydrate_stream<T, Fut, V>(
1634 items: impl IntoIterator<Item = T>,
1635 produce: impl FnMut(T) -> Fut,
1636) -> impl Stream<Item = Result<V, XrpcError>>
1637where
1638 Fut: Future<Output = Result<Option<V>, XrpcError>>,
1639{
1640 stream::iter(items)
1641 .map(produce)
1642 .buffered(FETCH_CONCURRENCY)
1643 .try_filter_map(|view| async move { Ok(view) })
1644}
1645
1646fn hydrate_record_stream<V>(
1647 state: &AppState,
1648 nsid: Nsid<DefaultStr>,
1649 items: Vec<EdgeItem>,
1650 provenance: HitProvenance,
1651) -> impl Stream<Item = Result<RecordView<V>, XrpcError>> + Send + 'static
1652where
1653 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1654{
1655 let owned = state.clone();
1656 hydrate_stream(items, move |item| {
1657 let owned = owned.clone();
1658 let nsid = nsid.clone();
1659 async move {
1660 let EdgeItem { uri, sort_micros } = item;
1661 let result = hydrate_record_view::<V>(&owned, &nsid, uri.clone(), sort_micros).await;
1662 if matches!(provenance, HitProvenance::Indexed)
1663 && let Err(err) = &result
1664 && is_index_evictable(err)
1665 {
1666 owned.edges.remove_source(&uri);
1667 }
1668 drop_unhydratable(provenance, &nsid, &uri, result)
1669 }
1670 })
1671}
1672
1673enum PagePhase {
1674 Head,
1675 Body { first: bool },
1676 Done,
1677}
1678
1679struct PageState<S> {
1680 items: std::pin::Pin<Box<S>>,
1681 phase: PagePhase,
1682 array_key: &'static str,
1683 tail: Vec<u8>,
1684 permit: Option<HeavyPermit>,
1685}
1686
1687pub(crate) fn paged_tail(cursor: Option<String>) -> Vec<u8> {
1688 let encoded = serde_json::to_string(&cursor).unwrap_or_else(|_| "null".to_owned());
1689 format!("],\"cursor\":{encoded}}}").into_bytes()
1690}
1691
1692fn unpaged_tail() -> Vec<u8> {
1693 b"]}".to_vec()
1694}
1695
1696pub(crate) fn json_stream<V, S>(
1697 array_key: &'static str,
1698 items: S,
1699 tail: Vec<u8>,
1700 permit: Option<HeavyPermit>,
1701) -> Response
1702where
1703 V: Serialize + Send + 'static,
1704 S: Stream<Item = Result<V, XrpcError>> + Send + 'static,
1705{
1706 let init = PageState {
1707 items: Box::pin(items),
1708 phase: PagePhase::Head,
1709 array_key,
1710 tail,
1711 permit,
1712 };
1713 let chunks = stream::unfold(init, |mut st| async move {
1714 match st.phase {
1715 PagePhase::Head => {
1716 let head = format!("{{\"{}\":[", st.array_key).into_bytes();
1717 st.phase = PagePhase::Body { first: true };
1718 Some((Ok::<Vec<u8>, Infallible>(head), st))
1719 }
1720 PagePhase::Body { first } => match st.items.next().await {
1721 Some(Ok(view)) => match serde_json::to_vec(&Deduped(&view)) {
1722 Ok(encoded) => {
1723 let mut chunk = Vec::with_capacity(encoded.len() + 1);
1724 if !first {
1725 chunk.push(b',');
1726 }
1727 chunk.extend_from_slice(&encoded);
1728 st.phase = PagePhase::Body { first: false };
1729 Some((Ok(chunk), st))
1730 }
1731 Err(e) => {
1732 tracing::warn!(error = %e, "skipping hit, serialize failed mid-stream");
1733 Some((Ok(Vec::new()), st))
1734 }
1735 },
1736 Some(Err(e)) => {
1737 tracing::warn!(error = %e, "ending page early, hydration failed mid-stream");
1738 let tail = std::mem::take(&mut st.tail);
1739 st.phase = PagePhase::Done;
1740 Some((Ok(tail), st))
1741 }
1742 None => {
1743 let tail = std::mem::take(&mut st.tail);
1744 st.phase = PagePhase::Done;
1745 Some((Ok(tail), st))
1746 }
1747 },
1748 PagePhase::Done => {
1749 drop(st.permit.take());
1750 None
1751 }
1752 }
1753 });
1754 (
1755 [(CONTENT_TYPE, "application/json")],
1756 Body::from_stream(chunks),
1757 )
1758 .into_response()
1759}
1760
1761async fn bulk_fetch<R, V>(state: &AppState, uris: Vec<String>) -> Result<Response, XrpcError>
1762where
1763 R: XrpcResp,
1764 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1765{
1766 if uris.is_empty() {
1767 return Err(XrpcError::InvalidParams("at least one uri required".into()));
1768 }
1769 if uris.len() > BULK_LIMIT {
1770 return Err(XrpcError::InvalidParams(format!(
1771 "at most {BULK_LIMIT} uris per request"
1772 )));
1773 }
1774 let parsed: Vec<AtUri<DefaultStr>> = uris
1775 .iter()
1776 .map(|s| parse_uri(s))
1777 .collect::<Result<_, _>>()?;
1778 let nsid = nsid_static(R::NSID);
1779 if let Some(bad) = parsed
1780 .iter()
1781 .find(|uri| uri.collection().is_none_or(|c| c.as_ref() != nsid.as_ref()))
1782 {
1783 return Err(XrpcError::InvalidParams(format!(
1784 "uri collection must be {}, got {}",
1785 nsid.as_ref(),
1786 bad.as_ref()
1787 )));
1788 }
1789 bulk_stream::<R, V>(state, parsed)
1790}
1791
1792fn bulk_stream<R, V>(
1793 state: &AppState,
1794 parsed: Vec<AtUri<DefaultStr>>,
1795) -> Result<Response, XrpcError>
1796where
1797 R: XrpcResp,
1798 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1799{
1800 let nsid = nsid_static(R::NSID);
1801 let permit = state.heavy_permit()?;
1802 let items = parsed
1803 .into_iter()
1804 .map(|uri| EdgeItem {
1805 uri,
1806 sort_micros: 0,
1807 })
1808 .collect();
1809 let views = hydrate_record_stream::<V>(state, nsid, items, HitProvenance::ClientSupplied);
1810 Ok(json_stream::<RecordView<V>, _>(
1811 "items",
1812 views,
1813 unpaged_tail(),
1814 permit,
1815 ))
1816}
1817
1818fn record_edge_page<R, F>(
1819 state: &AppState,
1820 q: &TypedListQuery<F>,
1821) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>
1822where
1823 R: XrpcResp + HasSubject,
1824 F: ListFilter,
1825{
1826 let subject = parse_subject(&q.subject, R::SHAPE)?;
1827 let cursor = parse_cursor(q.cursor.as_deref())?;
1828 let limit = parse_limit(q.limit)?;
1829 let dir = q.dir();
1830 let nsid = nsid_static(R::NSID);
1831 let page = if q.filter.is_identity() {
1832 let key = EdgeKey::new(nsid.clone(), subject);
1833 state.edges.list(&key, cursor, limit, dir)
1834 } else {
1835 let pred = q.filter.predicate(state, &subject);
1836 let key = EdgeKey::new(nsid.clone(), subject);
1837 state.edges.list_filtered(&key, cursor, limit, dir, pred)
1838 };
1839 Ok((page, nsid))
1840}
1841
1842async fn list_records<R, V, F>(
1843 state: &AppState,
1844 q: TypedListQuery<F>,
1845) -> Result<Response, XrpcError>
1846where
1847 R: XrpcResp + HasSubject,
1848 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1849 F: ListFilter,
1850{
1851 let (page, nsid) = record_edge_page::<R, F>(state, &q)?;
1852 let permit = state.heavy_permit()?;
1853 let views = hydrate_record_stream::<V>(state, nsid, page.items, HitProvenance::Indexed);
1854 Ok(json_stream::<RecordView<V>, _>(
1855 "items",
1856 views,
1857 paged_tail(page.next.map(PageToken::encode_token)),
1858 permit,
1859 ))
1860}
1861
1862fn count_for<R: XrpcResp + HasSubject>(
1863 state: &AppState,
1864 q: CountQuery,
1865) -> Result<CountResponse, XrpcError> {
1866 let subject = parse_subject(&q.subject, R::SHAPE)?;
1867 let key = EdgeKey::new(nsid_static(R::NSID), subject);
1868 Ok(CountResponse {
1869 count: state.edges.count(&key),
1870 distinct_authors: state.edges.count_distinct_authors(&key),
1871 })
1872}
1873
1874fn count_typed_for<R, F>(
1875 state: &AppState,
1876 q: TypedCountQuery<F>,
1877) -> Result<CountResponse, XrpcError>
1878where
1879 R: XrpcResp + HasSubject,
1880 F: CountFilter,
1881{
1882 let subject = parse_subject(&q.subject, R::SHAPE)?;
1883 let key = EdgeKey::new(nsid_static(R::NSID), subject);
1884 let counted = q.filter.count(state, &key);
1885 Ok(CountResponse {
1886 count: counted.count.get(),
1887 distinct_authors: counted.distinct_authors.get(),
1888 })
1889}
1890
1891// does `actor` have an edge of this kind pointing at `subject`, returns its own uri
1892fn get_for<R: XrpcResp + HasSubject>(
1893 state: &AppState,
1894 q: GetEdgeQuery,
1895) -> Result<EdgeUriResponse, XrpcError> {
1896 let subject = parse_subject(&q.subject, R::SHAPE)?;
1897 let key = EdgeKey::new(nsid_static(R::NSID), subject);
1898 let uri = state
1899 .edges
1900 .viewer_source(&key, q.actor.as_str())
1901 .ok_or(XrpcError::NotFound)?;
1902 Ok(EdgeUriResponse { uri })
1903}
1904
1905fn mirror_edge_page<M, F>(
1906 state: &AppState,
1907 q: &TypedListQuery<F>,
1908) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>
1909where
1910 M: MirrorOf,
1911 F: ListFilter,
1912{
1913 let subject = parse_subject(&q.subject, M::SHAPE)?;
1914 let cursor = parse_cursor(q.cursor.as_deref())?;
1915 let limit = parse_limit(q.limit)?;
1916 let dir = q.dir();
1917 let edge_nsid = nsid_static(M::EDGE_KIND);
1918 let page = if q.filter.is_identity() {
1919 let key = EdgeKey::new(edge_nsid, subject);
1920 state.edges.list(&key, cursor, limit, dir)
1921 } else {
1922 let pred = q.filter.predicate(state, &subject);
1923 let key = EdgeKey::new(edge_nsid, subject);
1924 state.edges.list_filtered(&key, cursor, limit, dir, pred)
1925 };
1926 let record_nsid = nsid_static(<M::Record as XrpcResp>::NSID);
1927 Ok((page, record_nsid))
1928}
1929
1930async fn list_mirror<M, V, F>(state: &AppState, q: TypedListQuery<F>) -> Result<Response, XrpcError>
1931where
1932 M: MirrorOf,
1933 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1934 F: ListFilter,
1935{
1936 let (page, record_nsid) = mirror_edge_page::<M, F>(state, &q)?;
1937 let permit = state.heavy_permit()?;
1938 let views = hydrate_record_stream::<V>(state, record_nsid, page.items, HitProvenance::Indexed);
1939 Ok(json_stream::<RecordView<V>, _>(
1940 "items",
1941 views,
1942 paged_tail(page.next.map(PageToken::encode_token)),
1943 permit,
1944 ))
1945}
1946
1947fn count_mirror<M: MirrorOf>(state: &AppState, q: CountQuery) -> Result<CountResponse, XrpcError> {
1948 let subject = parse_subject(&q.subject, M::SHAPE)?;
1949 let key = EdgeKey::new(nsid_static(M::EDGE_KIND), subject);
1950 Ok(CountResponse {
1951 count: state.edges.count(&key),
1952 distinct_authors: state.edges.count_distinct_authors(&key),
1953 })
1954}
1955
1956async fn list_stars(
1957 State(state): State<AppState>,
1958 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
1959) -> Result<Response, XrpcError> {
1960 list_records::<StarRecord, Star<DefaultStr>, _>(&state, q).await
1961}
1962
1963async fn count_stars(
1964 State(state): State<AppState>,
1965 XrpcQuery(q): XrpcQuery<CountQuery>,
1966) -> Result<Json<CountResponse>, XrpcError> {
1967 count_for::<StarRecord>(&state, q).map(Json)
1968}
1969
1970async fn get_star(
1971 State(state): State<AppState>,
1972 XrpcQuery(q): XrpcQuery<GetEdgeQuery>,
1973) -> Result<Json<EdgeUriResponse>, XrpcError> {
1974 get_for::<StarRecord>(&state, q).map(Json)
1975}
1976
1977async fn list_follows(
1978 State(state): State<AppState>,
1979 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
1980) -> Result<Response, XrpcError> {
1981 list_records::<FollowRecord, Follow<DefaultStr>, _>(&state, q).await
1982}
1983
1984async fn count_follows(
1985 State(state): State<AppState>,
1986 XrpcQuery(q): XrpcQuery<CountQuery>,
1987) -> Result<Json<CountResponse>, XrpcError> {
1988 count_for::<FollowRecord>(&state, q).map(Json)
1989}
1990
1991async fn get_follow(
1992 State(state): State<AppState>,
1993 XrpcQuery(q): XrpcQuery<GetEdgeQuery>,
1994) -> Result<Json<EdgeUriResponse>, XrpcError> {
1995 get_for::<FollowRecord>(&state, q).map(Json)
1996}
1997
1998async fn list_issues(
1999 State(state): State<AppState>,
2000 XrpcQuery(q): XrpcQuery<TypedListQuery<IssueFilter>>,
2001) -> Result<Response, XrpcError> {
2002 let (page, nsid) = record_edge_page::<IssueRecord, _>(&state, &q)?;
2003 let permit = state.heavy_permit()?;
2004 let owned = state.clone();
2005 let items = hydrate_record_stream::<Issue<DefaultStr>>(
2006 &state,
2007 nsid,
2008 page.items,
2009 HitProvenance::Indexed,
2010 )
2011 .map(move |view| view.map(|v| enrich_issue_view(&owned, v)));
2012 Ok(json_stream::<StatefulItem<Issue<DefaultStr>>, _>(
2013 "items",
2014 items,
2015 paged_tail(page.next.map(PageToken::encode_token)),
2016 permit,
2017 ))
2018}
2019
2020async fn count_issues(
2021 State(state): State<AppState>,
2022 XrpcQuery(q): XrpcQuery<TypedCountQuery<IssueFilter>>,
2023) -> Result<Json<CountResponse>, XrpcError> {
2024 count_typed_for::<IssueRecord, _>(&state, q).map(Json)
2025}
2026
2027async fn list_pulls(
2028 State(state): State<AppState>,
2029 XrpcQuery(q): XrpcQuery<TypedListQuery<PullFilter>>,
2030) -> Result<Response, XrpcError> {
2031 let (page, nsid) = record_edge_page::<PullRecord, _>(&state, &q)?;
2032 let permit = state.heavy_permit()?;
2033 let owned = state.clone();
2034 let items =
2035 hydrate_record_stream::<Pull<DefaultStr>>(&state, nsid, page.items, HitProvenance::Indexed)
2036 .map(move |view| view.map(|v| enrich_pull_view(&owned, v)));
2037 Ok(json_stream::<StatefulItem<Pull<DefaultStr>>, _>(
2038 "items",
2039 items,
2040 paged_tail(page.next.map(PageToken::encode_token)),
2041 permit,
2042 ))
2043}
2044
2045async fn count_pulls(
2046 State(state): State<AppState>,
2047 XrpcQuery(q): XrpcQuery<TypedCountQuery<PullFilter>>,
2048) -> Result<Json<CountResponse>, XrpcError> {
2049 count_typed_for::<PullRecord, _>(&state, q).map(Json)
2050}
2051
2052async fn list_feed_comments(
2053 State(state): State<AppState>,
2054 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2055) -> Result<Response, XrpcError> {
2056 list_records::<FeedCommentRecord, FeedComment<DefaultStr>, _>(&state, q).await
2057}
2058
2059async fn count_feed_comments(
2060 State(state): State<AppState>,
2061 XrpcQuery(q): XrpcQuery<CountQuery>,
2062) -> Result<Json<CountResponse>, XrpcError> {
2063 count_for::<FeedCommentRecord>(&state, q).map(Json)
2064}
2065
2066async fn list_reactions(
2067 State(state): State<AppState>,
2068 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2069) -> Result<Response, XrpcError> {
2070 list_records::<ReactionRecord, Reaction<DefaultStr>, _>(&state, q).await
2071}
2072
2073async fn count_reactions(
2074 State(state): State<AppState>,
2075 XrpcQuery(q): XrpcQuery<CountQuery>,
2076) -> Result<Json<CountResponse>, XrpcError> {
2077 count_for::<ReactionRecord>(&state, q).map(Json)
2078}
2079
2080async fn list_ref_updates(
2081 State(state): State<AppState>,
2082 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2083) -> Result<Response, XrpcError> {
2084 list_records::<RefUpdateRecord, RefUpdate<DefaultStr>, _>(&state, q).await
2085}
2086
2087async fn count_ref_updates(
2088 State(state): State<AppState>,
2089 XrpcQuery(q): XrpcQuery<CountQuery>,
2090) -> Result<Json<CountResponse>, XrpcError> {
2091 count_for::<RefUpdateRecord>(&state, q).map(Json)
2092}
2093
2094async fn list_collaborators(
2095 State(state): State<AppState>,
2096 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2097) -> Result<Response, XrpcError> {
2098 list_records::<CollaboratorRecord, Collaborator<DefaultStr>, _>(&state, q).await
2099}
2100
2101async fn count_collaborators(
2102 State(state): State<AppState>,
2103 XrpcQuery(q): XrpcQuery<CountQuery>,
2104) -> Result<Json<CountResponse>, XrpcError> {
2105 count_for::<CollaboratorRecord>(&state, q).map(Json)
2106}
2107
2108async fn list_issue_states(
2109 State(state): State<AppState>,
2110 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2111) -> Result<Response, XrpcError> {
2112 list_records::<IssueStateRecord, IssueState<DefaultStr>, _>(&state, q).await
2113}
2114
2115async fn count_issue_states(
2116 State(state): State<AppState>,
2117 XrpcQuery(q): XrpcQuery<CountQuery>,
2118) -> Result<Json<CountResponse>, XrpcError> {
2119 count_for::<IssueStateRecord>(&state, q).map(Json)
2120}
2121
2122async fn list_pull_statuses(
2123 State(state): State<AppState>,
2124 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2125) -> Result<Response, XrpcError> {
2126 list_records::<PullStatusRecord, PullStatus<DefaultStr>, _>(&state, q).await
2127}
2128
2129async fn count_pull_statuses(
2130 State(state): State<AppState>,
2131 XrpcQuery(q): XrpcQuery<CountQuery>,
2132) -> Result<Json<CountResponse>, XrpcError> {
2133 count_for::<PullStatusRecord>(&state, q).map(Json)
2134}
2135
2136async fn list_repos(
2137 State(state): State<AppState>,
2138 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2139) -> Result<Response, XrpcError> {
2140 list_records::<RepoRecord, Repo<DefaultStr>, _>(&state, q).await
2141}
2142
2143async fn count_repos(
2144 State(state): State<AppState>,
2145 XrpcQuery(q): XrpcQuery<CountQuery>,
2146) -> Result<Json<CountResponse>, XrpcError> {
2147 count_for::<RepoRecord>(&state, q).map(Json)
2148}
2149
2150// forks are repo records pointing back at a repo, so they get their own edge
2151// kind instead of a collection
2152async fn count_forks(
2153 State(state): State<AppState>,
2154 XrpcQuery(q): XrpcQuery<CountQuery>,
2155) -> Result<Json<CountResponse>, XrpcError> {
2156 let subject = parse_subject(&q.subject, FORK_SUBJECT_SHAPE)?;
2157 let key = EdgeKey::new(nsid_static(REPO_SOURCE_EDGE_KIND), subject);
2158 Ok(Json(CountResponse {
2159 count: state.edges.count(&key),
2160 distinct_authors: state.edges.count_distinct_authors(&key),
2161 }))
2162}
2163
2164async fn list_knots(
2165 State(state): State<AppState>,
2166 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2167) -> Result<Response, XrpcError> {
2168 list_records::<KnotRecord, Knot<DefaultStr>, _>(&state, q).await
2169}
2170
2171async fn count_knots(
2172 State(state): State<AppState>,
2173 XrpcQuery(q): XrpcQuery<CountQuery>,
2174) -> Result<Json<CountResponse>, XrpcError> {
2175 count_for::<KnotRecord>(&state, q).map(Json)
2176}
2177
2178async fn list_spindles(
2179 State(state): State<AppState>,
2180 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2181) -> Result<Response, XrpcError> {
2182 list_records::<SpindleRecord, Spindle<DefaultStr>, _>(&state, q).await
2183}
2184
2185async fn count_spindles(
2186 State(state): State<AppState>,
2187 XrpcQuery(q): XrpcQuery<CountQuery>,
2188) -> Result<Json<CountResponse>, XrpcError> {
2189 count_for::<SpindleRecord>(&state, q).map(Json)
2190}
2191
2192async fn list_public_keys(
2193 State(state): State<AppState>,
2194 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2195) -> Result<Response, XrpcError> {
2196 list_records::<PublicKeyRecord, PublicKey<DefaultStr>, _>(&state, q).await
2197}
2198
2199async fn count_public_keys(
2200 State(state): State<AppState>,
2201 XrpcQuery(q): XrpcQuery<CountQuery>,
2202) -> Result<Json<CountResponse>, XrpcError> {
2203 count_for::<PublicKeyRecord>(&state, q).map(Json)
2204}
2205
2206async fn list_vouches(
2207 State(state): State<AppState>,
2208 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2209) -> Result<Response, XrpcError> {
2210 list_records::<VouchRecord, Vouch<DefaultStr>, _>(&state, q).await
2211}
2212
2213async fn count_vouches(
2214 State(state): State<AppState>,
2215 XrpcQuery(q): XrpcQuery<CountQuery>,
2216) -> Result<Json<CountResponse>, XrpcError> {
2217 count_for::<VouchRecord>(&state, q).map(Json)
2218}
2219
2220async fn list_stars_by(
2221 State(state): State<AppState>,
2222 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2223) -> Result<Response, XrpcError> {
2224 list_mirror::<StarBy, Star<DefaultStr>, _>(&state, q).await
2225}
2226async fn count_stars_by(
2227 State(state): State<AppState>,
2228 XrpcQuery(q): XrpcQuery<CountQuery>,
2229) -> Result<Json<CountResponse>, XrpcError> {
2230 count_mirror::<StarBy>(&state, q).map(Json)
2231}
2232
2233async fn list_reactions_by(
2234 State(state): State<AppState>,
2235 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2236) -> Result<Response, XrpcError> {
2237 list_mirror::<ReactionBy, Reaction<DefaultStr>, _>(&state, q).await
2238}
2239async fn count_reactions_by(
2240 State(state): State<AppState>,
2241 XrpcQuery(q): XrpcQuery<CountQuery>,
2242) -> Result<Json<CountResponse>, XrpcError> {
2243 count_mirror::<ReactionBy>(&state, q).map(Json)
2244}
2245
2246async fn list_follows_by(
2247 State(state): State<AppState>,
2248 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2249) -> Result<Response, XrpcError> {
2250 list_mirror::<FollowBy, Follow<DefaultStr>, _>(&state, q).await
2251}
2252async fn count_follows_by(
2253 State(state): State<AppState>,
2254 XrpcQuery(q): XrpcQuery<CountQuery>,
2255) -> Result<Json<CountResponse>, XrpcError> {
2256 count_mirror::<FollowBy>(&state, q).map(Json)
2257}
2258
2259async fn list_vouches_by(
2260 State(state): State<AppState>,
2261 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2262) -> Result<Response, XrpcError> {
2263 list_mirror::<VouchBy, Vouch<DefaultStr>, _>(&state, q).await
2264}
2265async fn count_vouches_by(
2266 State(state): State<AppState>,
2267 XrpcQuery(q): XrpcQuery<CountQuery>,
2268) -> Result<Json<CountResponse>, XrpcError> {
2269 count_mirror::<VouchBy>(&state, q).map(Json)
2270}
2271
2272async fn list_ref_updates_by(
2273 State(state): State<AppState>,
2274 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2275) -> Result<Response, XrpcError> {
2276 list_mirror::<RefUpdateBy, RefUpdate<DefaultStr>, _>(&state, q).await
2277}
2278async fn count_ref_updates_by(
2279 State(state): State<AppState>,
2280 XrpcQuery(q): XrpcQuery<CountQuery>,
2281) -> Result<Json<CountResponse>, XrpcError> {
2282 count_mirror::<RefUpdateBy>(&state, q).map(Json)
2283}
2284
2285async fn list_knot_members_by(
2286 State(state): State<AppState>,
2287 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2288) -> Result<Response, XrpcError> {
2289 list_mirror::<KnotMemberBy, KnotMember<DefaultStr>, _>(&state, q).await
2290}
2291async fn count_knot_members_by(
2292 State(state): State<AppState>,
2293 XrpcQuery(q): XrpcQuery<CountQuery>,
2294) -> Result<Json<CountResponse>, XrpcError> {
2295 count_mirror::<KnotMemberBy>(&state, q).map(Json)
2296}
2297
2298async fn list_label_ops_by(
2299 State(state): State<AppState>,
2300 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2301) -> Result<Response, XrpcError> {
2302 list_mirror::<LabelOpBy, LabelOp<DefaultStr>, _>(&state, q).await
2303}
2304async fn count_label_ops_by(
2305 State(state): State<AppState>,
2306 XrpcQuery(q): XrpcQuery<CountQuery>,
2307) -> Result<Json<CountResponse>, XrpcError> {
2308 count_mirror::<LabelOpBy>(&state, q).map(Json)
2309}
2310
2311async fn list_pipelines_by(
2312 State(state): State<AppState>,
2313 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2314) -> Result<Response, XrpcError> {
2315 list_mirror::<PipelineBy, Pipeline<DefaultStr>, _>(&state, q).await
2316}
2317async fn count_pipelines_by(
2318 State(state): State<AppState>,
2319 XrpcQuery(q): XrpcQuery<CountQuery>,
2320) -> Result<Json<CountResponse>, XrpcError> {
2321 count_mirror::<PipelineBy>(&state, q).map(Json)
2322}
2323
2324async fn list_pipeline_statuses_by(
2325 State(state): State<AppState>,
2326 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2327) -> Result<Response, XrpcError> {
2328 list_mirror::<PipelineStatusBy, PipelineStatus<DefaultStr>, _>(&state, q).await
2329}
2330async fn count_pipeline_statuses_by(
2331 State(state): State<AppState>,
2332 XrpcQuery(q): XrpcQuery<CountQuery>,
2333) -> Result<Json<CountResponse>, XrpcError> {
2334 count_mirror::<PipelineStatusBy>(&state, q).map(Json)
2335}
2336
2337async fn list_artifacts_by(
2338 State(state): State<AppState>,
2339 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2340) -> Result<Response, XrpcError> {
2341 list_mirror::<ArtifactBy, Artifact<DefaultStr>, _>(&state, q).await
2342}
2343async fn count_artifacts_by(
2344 State(state): State<AppState>,
2345 XrpcQuery(q): XrpcQuery<CountQuery>,
2346) -> Result<Json<CountResponse>, XrpcError> {
2347 count_mirror::<ArtifactBy>(&state, q).map(Json)
2348}
2349
2350async fn list_collaborators_by(
2351 State(state): State<AppState>,
2352 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2353) -> Result<Response, XrpcError> {
2354 list_mirror::<CollaboratorBy, Collaborator<DefaultStr>, _>(&state, q).await
2355}
2356async fn count_collaborators_by(
2357 State(state): State<AppState>,
2358 XrpcQuery(q): XrpcQuery<CountQuery>,
2359) -> Result<Json<CountResponse>, XrpcError> {
2360 count_mirror::<CollaboratorBy>(&state, q).map(Json)
2361}
2362
2363async fn list_issues_by(
2364 State(state): State<AppState>,
2365 XrpcQuery(q): XrpcQuery<TypedListQuery<IssueFilter>>,
2366) -> Result<Response, XrpcError> {
2367 let (page, nsid) = mirror_edge_page::<IssueBy, _>(&state, &q)?;
2368 let permit = state.heavy_permit()?;
2369 let owned = state.clone();
2370 let items = hydrate_record_stream::<Issue<DefaultStr>>(
2371 &state,
2372 nsid,
2373 page.items,
2374 HitProvenance::Indexed,
2375 )
2376 .map(move |view| view.map(|v| enrich_issue_view(&owned, v)));
2377 Ok(json_stream::<StatefulItem<Issue<DefaultStr>>, _>(
2378 "items",
2379 items,
2380 paged_tail(page.next.map(PageToken::encode_token)),
2381 permit,
2382 ))
2383}
2384async fn count_issues_by(
2385 State(state): State<AppState>,
2386 XrpcQuery(q): XrpcQuery<CountQuery>,
2387) -> Result<Json<CountResponse>, XrpcError> {
2388 count_mirror::<IssueBy>(&state, q).map(Json)
2389}
2390
2391async fn list_feed_comments_by(
2392 State(state): State<AppState>,
2393 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2394) -> Result<Response, XrpcError> {
2395 list_mirror::<FeedCommentBy, FeedComment<DefaultStr>, _>(&state, q).await
2396}
2397async fn count_feed_comments_by(
2398 State(state): State<AppState>,
2399 XrpcQuery(q): XrpcQuery<CountQuery>,
2400) -> Result<Json<CountResponse>, XrpcError> {
2401 count_mirror::<FeedCommentBy>(&state, q).map(Json)
2402}
2403
2404async fn list_issue_states_by(
2405 State(state): State<AppState>,
2406 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2407) -> Result<Response, XrpcError> {
2408 list_mirror::<IssueStateBy, IssueState<DefaultStr>, _>(&state, q).await
2409}
2410async fn count_issue_states_by(
2411 State(state): State<AppState>,
2412 XrpcQuery(q): XrpcQuery<CountQuery>,
2413) -> Result<Json<CountResponse>, XrpcError> {
2414 count_mirror::<IssueStateBy>(&state, q).map(Json)
2415}
2416
2417async fn list_pulls_by(
2418 State(state): State<AppState>,
2419 XrpcQuery(q): XrpcQuery<TypedListQuery<PullFilter>>,
2420) -> Result<Response, XrpcError> {
2421 let (page, nsid) = mirror_edge_page::<PullBy, _>(&state, &q)?;
2422 let permit = state.heavy_permit()?;
2423 let owned = state.clone();
2424 let items =
2425 hydrate_record_stream::<Pull<DefaultStr>>(&state, nsid, page.items, HitProvenance::Indexed)
2426 .map(move |view| view.map(|v| enrich_pull_view(&owned, v)));
2427 Ok(json_stream::<StatefulItem<Pull<DefaultStr>>, _>(
2428 "items",
2429 items,
2430 paged_tail(page.next.map(PageToken::encode_token)),
2431 permit,
2432 ))
2433}
2434async fn count_pulls_by(
2435 State(state): State<AppState>,
2436 XrpcQuery(q): XrpcQuery<CountQuery>,
2437) -> Result<Json<CountResponse>, XrpcError> {
2438 count_mirror::<PullBy>(&state, q).map(Json)
2439}
2440
2441async fn list_pull_statuses_by(
2442 State(state): State<AppState>,
2443 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2444) -> Result<Response, XrpcError> {
2445 list_mirror::<PullStatusBy, PullStatus<DefaultStr>, _>(&state, q).await
2446}
2447async fn count_pull_statuses_by(
2448 State(state): State<AppState>,
2449 XrpcQuery(q): XrpcQuery<CountQuery>,
2450) -> Result<Json<CountResponse>, XrpcError> {
2451 count_mirror::<PullStatusBy>(&state, q).map(Json)
2452}
2453
2454async fn list_spindle_members_by(
2455 State(state): State<AppState>,
2456 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2457) -> Result<Response, XrpcError> {
2458 list_mirror::<SpindleMemberBy, SpindleMember<DefaultStr>, _>(&state, q).await
2459}
2460async fn count_spindle_members_by(
2461 State(state): State<AppState>,
2462 XrpcQuery(q): XrpcQuery<CountQuery>,
2463) -> Result<Json<CountResponse>, XrpcError> {
2464 count_mirror::<SpindleMemberBy>(&state, q).map(Json)
2465}
2466
2467async fn list_label_definitions(
2468 State(state): State<AppState>,
2469 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2470) -> Result<Response, XrpcError> {
2471 list_records::<LabelDefinitionRecord, LabelDefinition<DefaultStr>, _>(&state, q).await
2472}
2473
2474async fn count_label_definitions(
2475 State(state): State<AppState>,
2476 XrpcQuery(q): XrpcQuery<CountQuery>,
2477) -> Result<Json<CountResponse>, XrpcError> {
2478 count_for::<LabelDefinitionRecord>(&state, q).map(Json)
2479}
2480
2481async fn list_label_ops(
2482 State(state): State<AppState>,
2483 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2484) -> Result<Response, XrpcError> {
2485 list_records::<LabelOpRecord, LabelOp<DefaultStr>, _>(&state, q).await
2486}
2487
2488async fn count_label_ops(
2489 State(state): State<AppState>,
2490 XrpcQuery(q): XrpcQuery<CountQuery>,
2491) -> Result<Json<CountResponse>, XrpcError> {
2492 count_for::<LabelOpRecord>(&state, q).map(Json)
2493}
2494
2495async fn list_pipelines(
2496 State(state): State<AppState>,
2497 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2498) -> Result<Response, XrpcError> {
2499 list_records::<PipelineRecord, Pipeline<DefaultStr>, _>(&state, q).await
2500}
2501
2502async fn count_pipelines(
2503 State(state): State<AppState>,
2504 XrpcQuery(q): XrpcQuery<CountQuery>,
2505) -> Result<Json<CountResponse>, XrpcError> {
2506 count_for::<PipelineRecord>(&state, q).map(Json)
2507}
2508
2509async fn list_pipeline_statuses(
2510 State(state): State<AppState>,
2511 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2512) -> Result<Response, XrpcError> {
2513 list_records::<PipelineStatusRecord, PipelineStatus<DefaultStr>, _>(&state, q).await
2514}
2515
2516async fn count_pipeline_statuses(
2517 State(state): State<AppState>,
2518 XrpcQuery(q): XrpcQuery<CountQuery>,
2519) -> Result<Json<CountResponse>, XrpcError> {
2520 count_for::<PipelineStatusRecord>(&state, q).map(Json)
2521}
2522
2523async fn list_artifacts(
2524 State(state): State<AppState>,
2525 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2526) -> Result<Response, XrpcError> {
2527 list_records::<ArtifactRecord, Artifact<DefaultStr>, _>(&state, q).await
2528}
2529
2530async fn count_artifacts(
2531 State(state): State<AppState>,
2532 XrpcQuery(q): XrpcQuery<CountQuery>,
2533) -> Result<Json<CountResponse>, XrpcError> {
2534 count_for::<ArtifactRecord>(&state, q).map(Json)
2535}
2536
2537async fn list_knot_members(
2538 State(state): State<AppState>,
2539 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2540) -> Result<Response, XrpcError> {
2541 list_records::<KnotMemberRecord, KnotMember<DefaultStr>, _>(&state, q).await
2542}
2543
2544async fn count_knot_members(
2545 State(state): State<AppState>,
2546 XrpcQuery(q): XrpcQuery<CountQuery>,
2547) -> Result<Json<CountResponse>, XrpcError> {
2548 count_for::<KnotMemberRecord>(&state, q).map(Json)
2549}
2550
2551async fn list_spindle_members(
2552 State(state): State<AppState>,
2553 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2554) -> Result<Response, XrpcError> {
2555 list_records::<SpindleMemberRecord, SpindleMember<DefaultStr>, _>(&state, q).await
2556}
2557
2558async fn count_spindle_members(
2559 State(state): State<AppState>,
2560 XrpcQuery(q): XrpcQuery<CountQuery>,
2561) -> Result<Json<CountResponse>, XrpcError> {
2562 count_for::<SpindleMemberRecord>(&state, q).map(Json)
2563}
2564
2565async fn list_strings(
2566 State(state): State<AppState>,
2567 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2568) -> Result<Response, XrpcError> {
2569 list_records::<TangledStringRecord, TangledString<DefaultStr>, _>(&state, q).await
2570}
2571
2572async fn count_strings(
2573 State(state): State<AppState>,
2574 XrpcQuery(q): XrpcQuery<CountQuery>,
2575) -> Result<Json<CountResponse>, XrpcError> {
2576 count_for::<TangledStringRecord>(&state, q).map(Json)
2577}
2578
2579#[derive(Deserialize)]
2580struct ResolveMiniDocParams {
2581 identifier: AtIdentifier<DefaultStr>,
2582}
2583
2584async fn resolve_mini_doc(
2585 State(state): State<AppState>,
2586 XrpcQuery(q): XrpcQuery<ResolveMiniDocParams>,
2587) -> Result<Response, XrpcError> {
2588 let body = state
2589 .slingshot
2590 .resolve_mini_doc(&q.identifier)
2591 .await
2592 .map_err(map_slingshot)?;
2593 Ok((StatusCode::OK, [(CONTENT_TYPE, "application/json")], body).into_response())
2594}
2595
2596async fn get_coverage(State(state): State<AppState>) -> Json<CoverageEnvelope> {
2597 Json(state.coverage.snapshot().into())
2598}
2599
2600async fn search_query(
2601 State(state): State<AppState>,
2602 XrpcQuery(q): XrpcQuery<SearchQueryParams>,
2603) -> Result<Response, XrpcError> {
2604 if q.q.trim().is_empty() {
2605 return Err(XrpcError::InvalidParams("q must not be empty".into()));
2606 }
2607 let cursor = SearchCursor::from_token(q.cursor.as_deref())
2608 .map_err(|e| XrpcError::InvalidParams(format!("cursor: {e}")))?;
2609 let limit = parse_limit(q.limit)?;
2610 let filters = build_search_filters(&q)?;
2611 let permit = state.heavy_permit()?;
2612 let page = state
2613 .search
2614 .search(&q.q, filters, cursor, limit.get())
2615 .await
2616 .map_err(map_search_err)?;
2617 let next = page.next.map(SearchOffset::encode_token);
2618 let owned = state.clone();
2619 let hits = hydrate_stream(page.hits, move |hit| {
2620 let owned = owned.clone();
2621 let uri = hit.uri.clone();
2622 let nsid = hit.nsid.clone();
2623 async move {
2624 let result = hydrate_search_hit(&owned, hit).await;
2625 drop_unhydratable(HitProvenance::Indexed, &nsid, &uri, result)
2626 }
2627 });
2628 Ok(json_stream::<SearchHitView, _>(
2629 "hits",
2630 hits,
2631 paged_tail(next),
2632 permit,
2633 ))
2634}
2635
2636fn build_search_filters(q: &SearchQueryParams) -> Result<SearchFilters, XrpcError> {
2637 let since = q
2638 .since
2639 .as_deref()
2640 .map(parse_rfc3339_seconds)
2641 .transpose()
2642 .map_err(|e| XrpcError::InvalidParams(format!("since: {e}")))?;
2643 let until = q
2644 .until
2645 .as_deref()
2646 .map(parse_rfc3339_seconds)
2647 .transpose()
2648 .map_err(|e| XrpcError::InvalidParams(format!("until: {e}")))?;
2649 if let (Some(s), Some(u)) = (since, until)
2650 && s > u
2651 {
2652 return Err(XrpcError::InvalidParams("since must be <= until".into()));
2653 }
2654 Ok(SearchFilters {
2655 nsid: q.nsid.clone(),
2656 author: q.author.clone(),
2657 repo: q.repo.clone(),
2658 since,
2659 until,
2660 })
2661}
2662
2663fn parse_rfc3339_seconds(raw: &str) -> Result<i64, String> {
2664 chrono::DateTime::parse_from_rfc3339(raw)
2665 .map(|dt| dt.timestamp())
2666 .map_err(|e| format!("expected RFC3339, got {raw}: {e}"))
2667}
2668
2669async fn hydrate_search_hit(
2670 state: &AppState,
2671 hit: SearchHit,
2672) -> Result<Option<SearchHitView>, XrpcError> {
2673 let SearchHit { uri, nsid, score } = hit;
2674 let body = resolve_for_view(state, &nsid, uri).await?;
2675 let record = decode_canon_or_upgrade(&nsid, &body.value, &state.resolver)
2676 .await
2677 .map_err(|err| XrpcError::InvalidRecord(err.to_string()))?;
2678 let Some(value) = SearchableRecord::try_from_record(record) else {
2679 return Ok(None);
2680 };
2681 let Some(value) = value.normalize(&state.resolver).await else {
2682 return Ok(None);
2683 };
2684 Ok(Some(SearchHitView {
2685 uri: body.uri.clone(),
2686 cid: Some(body.cid.clone()),
2687 nsid,
2688 score,
2689 value,
2690 }))
2691}
2692
2693fn map_search_err(err: SearchError) -> XrpcError {
2694 use SearchError as E;
2695 match err {
2696 E::Query(e) => XrpcError::InvalidParams(format!("query: {e}")),
2697 e @ (E::Tantivy(_)
2698 | E::InvalidUri(_)
2699 | E::InvalidNsid(_)
2700 | E::MissingField(_)
2701 | E::Cancelled(_)) => XrpcError::Internal(format!("search: {e}")),
2702 }
2703}
2704
2705fn map_proxy_error(err: KnotProxyError) -> XrpcError {
2706 match err {
2707 KnotProxyError::CircuitOpen => {
2708 XrpcError::UpstreamUnavailable("knot circuit breaker open".into())
2709 }
2710 KnotProxyError::BlockedHost { host, reason } => {
2711 XrpcError::InvalidRecord(format!("knot host {host} is {reason} address space"))
2712 }
2713 KnotProxyError::PlaintextHttp { host } => {
2714 XrpcError::InvalidRecord(format!("knot host {host} requires https"))
2715 }
2716 KnotProxyError::Connect(e) => XrpcError::UpstreamUnavailable(format!("connect: {e}")),
2717 KnotProxyError::Timeout(e) => {
2718 XrpcError::UpstreamUnavailable(format!("upstream timeout: {e}"))
2719 }
2720 KnotProxyError::Redirect(e) => XrpcError::UpstreamUnavailable(format!("redirect: {e}")),
2721 KnotProxyError::Transport(e) => XrpcError::UpstreamUnavailable(format!("transport: {e}")),
2722 KnotProxyError::Upstream(s) => XrpcError::UpstreamUnavailable(format!("status {s}")),
2723 }
2724}
2725
2726fn validate_client_supplied_knot(state: &AppState, host: &KnotHost) -> Result<(), XrpcError> {
2727 let host_str = || host.url().host_str().unwrap_or_default().to_owned();
2728 if state.knots.requires_https() && host.url().scheme() != "https" {
2729 return Err(XrpcError::InvalidParams(format!(
2730 "knot host {} must be https",
2731 host_str(),
2732 )));
2733 }
2734 if state.knots.allows_private_hosts() {
2735 return Ok(());
2736 }
2737 match host.private_literal_reason() {
2738 None => Ok(()),
2739 Some(reason) => Err(XrpcError::InvalidParams(format!(
2740 "knot host {} blocked: {} address space",
2741 host_str(),
2742 reason,
2743 ))),
2744 }
2745}
2746
2747async fn resolve_knot_target(
2748 state: &AppState,
2749 repo_uri: AtUri<DefaultStr>,
2750) -> Result<(KnotHost, RepoSlug), XrpcError> {
2751 let rkey: Option<Rkey<DefaultStr>> = repo_uri.rkey().map(|r| r.clone().into_static());
2752 let (body, did) = resolve(state, ExpectedNsid::from_static(RepoRecord::NSID), repo_uri).await?;
2753 let value: Repo<DefaultStr> = serde_json::from_slice(&body.value)
2754 .map_err(|e| XrpcError::InvalidRecord(format!("decode repo record: {e}")))?;
2755 let host = KnotHost::parse(value.knot.as_ref())
2756 .map_err(|e| XrpcError::InvalidRecord(format!("knot field: {e}")))?;
2757 let name = pick_human_slug(rkey.as_ref(), value.name.as_deref()).ok_or_else(|| {
2758 XrpcError::InvalidRecord("at-uri missing rkey and record missing name".to_string())
2759 })?;
2760 let slug = RepoSlug::new(&did, &name)
2761 .map_err(|e| XrpcError::InvalidRecord(format!("repo slug: {e}")))?;
2762 Ok((host, slug))
2763}
2764
2765fn pick_human_slug(rkey: Option<&Rkey<DefaultStr>>, name: Option<&str>) -> Option<String> {
2766 match rkey {
2767 Some(r) if jacquard_common::types::tid::Tid::new(r.as_ref()).is_ok() => {
2768 Some(name.unwrap_or(r.as_ref()).to_owned())
2769 }
2770 Some(r) => Some(r.as_ref().to_owned()),
2771 None => name.map(str::to_owned),
2772 }
2773}
2774
2775fn filter_request_headers(client: &HeaderMap) -> HeaderMap {
2776 FORWARDED_REQUEST_HEADERS
2777 .iter()
2778 .fold(HeaderMap::new(), |mut acc, name| {
2779 if let Some(value) = client.get(*name) {
2780 acc.insert((*name).clone(), value.clone());
2781 }
2782 acc
2783 })
2784}
2785
2786fn upstream_to_axum(resp: ProxyResponse) -> Response {
2787 let status = resp.status();
2788 let upstream_headers = resp.headers().clone();
2789 let body = Body::from_stream(resp.into_body_stream());
2790 let mut response = Response::builder()
2791 .status(status)
2792 .body(body)
2793 .expect("response body construction must succeed");
2794 let response_headers = response.headers_mut();
2795 PASSTHROUGH_HEADERS.iter().for_each(|name| {
2796 if let Some(value) = upstream_headers.get(*name) {
2797 response_headers.insert((*name).clone(), value.clone());
2798 }
2799 });
2800 response
2801}
2802
2803async fn dispatch_proxy(
2804 state: AppState,
2805 headers: HeaderMap,
2806 nsid: Nsid<DefaultStr>,
2807 host: KnotHost,
2808 params: ProxyParams,
2809) -> Result<Response, XrpcError> {
2810 let forward: Vec<(&str, &str)> = params
2811 .iter()
2812 .map(|(k, v)| (k.as_str(), v.as_str()))
2813 .collect();
2814 let allowed = filter_request_headers(&headers);
2815 let upstream = state
2816 .knots
2817 .forward(&host, &nsid, &forward, allowed)
2818 .await
2819 .map_err(map_proxy_error)?;
2820 Ok(upstream_to_axum(upstream))
2821}
2822
2823fn extract_param(
2824 params: ProxyParams,
2825 key: &str,
2826) -> Result<Option<(String, ProxyParams)>, XrpcError> {
2827 let (matching, rest): (ProxyParams, ProxyParams) =
2828 params.into_iter().partition(|(k, _)| k == key);
2829 match matching.as_slice() {
2830 [] => Ok(None),
2831 [_] => Ok(matching.into_iter().next().map(|(_, v)| (v, rest))),
2832 _ => Err(XrpcError::InvalidParams(format!(
2833 "{key} parameter must appear at most once, got {}",
2834 matching.len(),
2835 ))),
2836 }
2837}
2838
2839async fn proxy_repo_handler(
2840 state: AppState,
2841 headers: HeaderMap,
2842 params: ProxyParams,
2843 nsid: Nsid<DefaultStr>,
2844) -> Result<Response, XrpcError> {
2845 let (repo_raw, rest) = extract_param(params, REPO_PARAM)?
2846 .ok_or_else(|| XrpcError::InvalidParams("missing repo".into()))?;
2847 let repo_uri = parse_uri(&repo_raw)?;
2848 let (host, slug) = resolve_knot_target(&state, repo_uri).await?;
2849 let forward = rest
2850 .into_iter()
2851 .chain(std::iter::once((
2852 REPO_PARAM.to_owned(),
2853 slug.as_str().to_owned(),
2854 )))
2855 .collect();
2856 dispatch_proxy(state, headers, nsid, host, forward).await
2857}
2858
2859async fn proxy_knot_handler(
2860 state: AppState,
2861 headers: HeaderMap,
2862 params: ProxyParams,
2863 nsid: Nsid<DefaultStr>,
2864) -> Result<Response, XrpcError> {
2865 let (knot_raw, forward) = extract_param(params, KNOT_HOST_PARAM)?
2866 .ok_or_else(|| XrpcError::InvalidParams("missing knot".into()))?;
2867 let host =
2868 KnotHost::parse(&knot_raw).map_err(|e| XrpcError::InvalidParams(format!("knot: {e}")))?;
2869 validate_client_supplied_knot(&state, &host)?;
2870 dispatch_proxy(state, headers, nsid, host, forward).await
2871}