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