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