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