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