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