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
740#[derive(Serialize)]
741#[serde(rename_all = "camelCase")]
742struct RecordView<V> {
743 uri: AtUri<DefaultStr>,
744 cid: Option<Cid<DefaultStr>>,
745 value: V,
746}
747
748#[derive(Serialize)]
749#[serde(rename_all = "camelCase")]
750struct StatefulItem<V> {
751 #[serde(flatten)]
752 view: RecordView<V>,
753 state: &'static str,
754 #[serde(skip_serializing_if = "Option::is_none")]
755 state_updated_at: Option<String>,
756 comment_count: u64,
757}
758
759fn format_micros(micros: u64) -> String {
760 let signed = i64::try_from(micros).ok();
761 let rfc = signed
762 .and_then(chrono::DateTime::<chrono::Utc>::from_timestamp_micros)
763 .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Micros, true));
764 rfc.unwrap_or_else(|| micros.to_string())
765}
766
767pub(crate) fn source_authority_did(source: &AtUri<DefaultStr>) -> Option<Did<DefaultStr>> {
768 match source.authority() {
769 AtIdentifier::Did(d) => Some(d.clone().into_static()),
770 AtIdentifier::Handle(_) => None,
771 }
772}
773
774fn enrich_issue_view(
775 state: &AppState,
776 view: RecordView<Issue<DefaultStr>>,
777) -> StatefulItem<Issue<DefaultStr>> {
778 let issue_author = source_authority_did(&view.uri);
779 let repo_did = view.value.repo.clone();
780 enrich_view(
781 &state.edges,
782 nsid_static("sh.tangled.feed.comment"),
783 &state.issue_states,
784 view,
785 move |src| accept_state_source(src, issue_author.as_ref(), &repo_did),
786 )
787}
788
789fn enrich_pull_view(
790 state: &AppState,
791 view: RecordView<Pull<DefaultStr>>,
792) -> StatefulItem<Pull<DefaultStr>> {
793 let pull_author = source_authority_did(&view.uri);
794 let target_repo = view.value.target.repo.clone();
795 enrich_view(
796 &state.edges,
797 nsid_static("sh.tangled.feed.comment"),
798 &state.pull_statuses,
799 view,
800 move |src| accept_state_source(src, pull_author.as_ref(), &target_repo),
801 )
802}
803
804pub(crate) fn accept_state_source(
805 source: &AtUri<DefaultStr>,
806 entity_author: Option<&Did<DefaultStr>>,
807 repo_owner: &Did<DefaultStr>,
808) -> bool {
809 let Some(src) = source_authority_did(source) else {
810 return false;
811 };
812 Some(&src) == entity_author || &src == repo_owner
813}
814
815fn enrich_view<V, K, F>(
816 edges: &EdgeStore,
817 comment_nsid: Nsid<DefaultStr>,
818 states: &StateIndex<K>,
819 view: RecordView<V>,
820 accept: F,
821) -> StatefulItem<V>
822where
823 K: StateKind + Default,
824 F: Fn(&AtUri<DefaultStr>) -> bool,
825{
826 let comment_count = edges.count(&EdgeKey::new(
827 comment_nsid,
828 SubjectRef::Uri(view.uri.clone()),
829 ));
830 let (state, state_updated_at) = states
831 .latest_by(&view.uri, accept)
832 .map_or((K::default().wire(), None), |(kind, micros)| {
833 (kind.wire(), Some(format_micros(micros)))
834 });
835 StatefulItem {
836 view,
837 state,
838 state_updated_at,
839 comment_count,
840 }
841}
842
843#[derive(Serialize)]
844#[serde(rename_all = "camelCase")]
845struct CountResponse {
846 count: u64,
847 distinct_authors: u64,
848}
849
850#[derive(Serialize)]
851#[serde(rename_all = "camelCase")]
852struct EdgeUriResponse {
853 uri: AtUri<DefaultStr>,
854}
855
856#[derive(Serialize)]
857#[serde(rename_all = "camelCase")]
858struct SearchHitView {
859 uri: AtUri<DefaultStr>,
860 cid: Option<Cid<DefaultStr>>,
861 nsid: Nsid<DefaultStr>,
862 score: f32,
863 value: SearchableRecord,
864}
865
866fn map_slingshot(err: SlingshotError) -> XrpcError {
867 use SlingshotError as E;
868 match err {
869 E::NotFound => XrpcError::NotFound,
870 e @ (E::Decode(_)
871 | E::MissingField(_)
872 | E::InvalidAtUri(_)
873 | E::InvalidCid(_)
874 | E::UriMismatch { .. }) => XrpcError::InvalidRecord(e.to_string()),
875 e @ (E::Network(_)
876 | E::Build(_)
877 | E::Upstream(_)
878 | E::BodyTooLarge { .. }
879 | E::BadScheme(_)) => XrpcError::UpstreamUnavailable(e.to_string()),
880 }
881}
882
883fn parse_uri(raw: &str) -> Result<AtUri<DefaultStr>, XrpcError> {
884 AtUri::<DefaultStr>::new_owned(raw).map_err(|e| XrpcError::InvalidParams(format!("uri: {e}")))
885}
886
887#[derive(Clone, Copy, Debug, Eq, PartialEq)]
888pub enum SubjectShape {
889 BareDid,
890 Collection(&'static str),
891 BareDidOrOneOfCollections(&'static [&'static str]),
892 OneOfCollections(&'static [&'static str]),
893 AnyAtUri,
894}
895
896pub trait HasSubject {
897 const SHAPE: SubjectShape;
898}
899
900pub trait MirrorOf {
901 type Record: XrpcResp;
902 const EDGE_KIND: &'static str;
903 const SHAPE: SubjectShape;
904}
905
906pub struct StarBy;
907pub struct ReactionBy;
908pub struct FollowBy;
909pub struct VouchBy;
910pub struct RefUpdateBy;
911pub struct KnotMemberBy;
912pub struct LabelOpBy;
913pub struct PipelineBy;
914pub struct PipelineStatusBy;
915pub struct ArtifactBy;
916pub struct CollaboratorBy;
917pub struct FeedCommentBy;
918pub struct IssueBy;
919pub struct IssueStateBy;
920pub struct PullBy;
921pub struct PullStatusBy;
922pub struct SpindleMemberBy;
923
924impl MirrorOf for StarBy {
925 type Record = StarRecord;
926 const EDGE_KIND: &'static str = "sh.tangled.feed.star.by";
927 const SHAPE: SubjectShape = SubjectShape::BareDid;
928}
929impl MirrorOf for ReactionBy {
930 type Record = ReactionRecord;
931 const EDGE_KIND: &'static str = "sh.tangled.feed.reaction.by";
932 const SHAPE: SubjectShape = SubjectShape::BareDid;
933}
934impl MirrorOf for FollowBy {
935 type Record = FollowRecord;
936 const EDGE_KIND: &'static str = "sh.tangled.graph.follow.by";
937 const SHAPE: SubjectShape = SubjectShape::BareDid;
938}
939impl MirrorOf for VouchBy {
940 type Record = VouchRecord;
941 const EDGE_KIND: &'static str = "sh.tangled.graph.vouch.by";
942 const SHAPE: SubjectShape = SubjectShape::BareDid;
943}
944impl MirrorOf for RefUpdateBy {
945 type Record = RefUpdateRecord;
946 const EDGE_KIND: &'static str = "sh.tangled.git.refUpdate.by";
947 const SHAPE: SubjectShape = SubjectShape::BareDid;
948}
949impl MirrorOf for KnotMemberBy {
950 type Record = KnotMemberRecord;
951 const EDGE_KIND: &'static str = "sh.tangled.knot.member.by";
952 const SHAPE: SubjectShape = SubjectShape::BareDid;
953}
954impl MirrorOf for LabelOpBy {
955 type Record = LabelOpRecord;
956 const EDGE_KIND: &'static str = "sh.tangled.label.op.by";
957 const SHAPE: SubjectShape = SubjectShape::BareDid;
958}
959impl MirrorOf for PipelineBy {
960 type Record = PipelineRecord;
961 const EDGE_KIND: &'static str = "sh.tangled.pipeline.by";
962 const SHAPE: SubjectShape = SubjectShape::BareDid;
963}
964impl MirrorOf for PipelineStatusBy {
965 type Record = PipelineStatusRecord;
966 const EDGE_KIND: &'static str = "sh.tangled.pipeline.status.by";
967 const SHAPE: SubjectShape = SubjectShape::BareDid;
968}
969impl MirrorOf for ArtifactBy {
970 type Record = ArtifactRecord;
971 const EDGE_KIND: &'static str = "sh.tangled.repo.artifact.by";
972 const SHAPE: SubjectShape = SubjectShape::BareDid;
973}
974impl MirrorOf for CollaboratorBy {
975 type Record = CollaboratorRecord;
976 const EDGE_KIND: &'static str = "sh.tangled.repo.collaborator.by";
977 const SHAPE: SubjectShape = SubjectShape::BareDid;
978}
979impl MirrorOf for FeedCommentBy {
980 type Record = FeedCommentRecord;
981 const EDGE_KIND: &'static str = "sh.tangled.feed.comment.by";
982 const SHAPE: SubjectShape = SubjectShape::BareDid;
983}
984impl MirrorOf for IssueBy {
985 type Record = IssueRecord;
986 const EDGE_KIND: &'static str = "sh.tangled.repo.issue.by";
987 const SHAPE: SubjectShape = SubjectShape::BareDid;
988}
989impl MirrorOf for IssueStateBy {
990 type Record = IssueStateRecord;
991 const EDGE_KIND: &'static str = "sh.tangled.repo.issue.state.by";
992 const SHAPE: SubjectShape = SubjectShape::BareDid;
993}
994impl MirrorOf for PullBy {
995 type Record = PullRecord;
996 const EDGE_KIND: &'static str = "sh.tangled.repo.pull.by";
997 const SHAPE: SubjectShape = SubjectShape::BareDid;
998}
999impl MirrorOf for PullStatusBy {
1000 type Record = PullStatusRecord;
1001 const EDGE_KIND: &'static str = "sh.tangled.repo.pull.status.by";
1002 const SHAPE: SubjectShape = SubjectShape::BareDid;
1003}
1004impl MirrorOf for SpindleMemberBy {
1005 type Record = SpindleMemberRecord;
1006 const EDGE_KIND: &'static str = "sh.tangled.spindle.member.by";
1007 const SHAPE: SubjectShape = SubjectShape::BareDid;
1008}
1009
1010impl HasSubject for StarRecord {
1011 const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.string"]);
1012}
1013impl HasSubject for FollowRecord {
1014 const SHAPE: SubjectShape = SubjectShape::BareDid;
1015}
1016impl HasSubject for IssueRecord {
1017 const SHAPE: SubjectShape = SubjectShape::BareDid;
1018}
1019impl HasSubject for PullRecord {
1020 const SHAPE: SubjectShape = SubjectShape::BareDid;
1021}
1022impl HasSubject for FeedCommentRecord {
1023 const SHAPE: SubjectShape = SubjectShape::OneOfCollections(&[
1024 "sh.tangled.repo.issue",
1025 "sh.tangled.repo.pull",
1026 "sh.tangled.string",
1027 ]);
1028}
1029impl HasSubject for LabelDefinitionRecord {
1030 const SHAPE: SubjectShape = SubjectShape::BareDid;
1031}
1032impl HasSubject for LabelOpRecord {
1033 const SHAPE: SubjectShape =
1034 SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull"]);
1035}
1036impl HasSubject for PipelineRecord {
1037 const SHAPE: SubjectShape = SubjectShape::BareDid;
1038}
1039impl HasSubject for PipelineStatusRecord {
1040 const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.pipeline");
1041}
1042impl HasSubject for ArtifactRecord {
1043 const SHAPE: SubjectShape = SubjectShape::BareDid;
1044}
1045impl HasSubject for KnotMemberRecord {
1046 const SHAPE: SubjectShape = SubjectShape::BareDid;
1047}
1048impl HasSubject for SpindleMemberRecord {
1049 const SHAPE: SubjectShape = SubjectShape::BareDid;
1050}
1051impl HasSubject for TangledStringRecord {
1052 const SHAPE: SubjectShape = SubjectShape::BareDid;
1053}
1054impl HasSubject for ReactionRecord {
1055 const SHAPE: SubjectShape = SubjectShape::AnyAtUri;
1056}
1057impl HasSubject for RefUpdateRecord {
1058 const SHAPE: SubjectShape = SubjectShape::BareDid;
1059}
1060impl HasSubject for CollaboratorRecord {
1061 const SHAPE: SubjectShape = SubjectShape::BareDid;
1062}
1063impl HasSubject for IssueStateRecord {
1064 const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.repo.issue");
1065}
1066impl HasSubject for PullStatusRecord {
1067 const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.repo.pull");
1068}
1069impl HasSubject for KnotRecord {
1070 const SHAPE: SubjectShape = SubjectShape::BareDid;
1071}
1072impl HasSubject for SpindleRecord {
1073 const SHAPE: SubjectShape = SubjectShape::BareDid;
1074}
1075impl HasSubject for PublicKeyRecord {
1076 const SHAPE: SubjectShape = SubjectShape::BareDid;
1077}
1078impl HasSubject for RepoRecord {
1079 const SHAPE: SubjectShape = SubjectShape::BareDid;
1080}
1081impl HasSubject for VouchRecord {
1082 const SHAPE: SubjectShape = SubjectShape::BareDid;
1083}
1084
1085fn parse_subject(raw: &SubjectQuery, shape: SubjectShape) -> Result<SubjectRef, XrpcError> {
1086 let uri = match raw {
1087 SubjectQuery::Did(did) => {
1088 return match shape {
1089 SubjectShape::BareDid | SubjectShape::BareDidOrOneOfCollections(_) => {
1090 Ok(SubjectRef::Did(did.clone()))
1091 }
1092 SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!(
1093 "subject must be at://<did>/{expected}/<rkey>, got bare did"
1094 ))),
1095 SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
1096 "subject must be at://<did>/<nsid>/<rkey> with nsid in [{}], got bare did",
1097 allowed.join(", "),
1098 ))),
1099 SubjectShape::AnyAtUri => Err(XrpcError::InvalidParams(
1100 "subject must be at-uri form, got bare did".into(),
1101 )),
1102 };
1103 }
1104 SubjectQuery::Uri(uri) => uri,
1105 };
1106 if matches!(uri.authority(), AtIdentifier::Handle(_)) {
1107 return Err(XrpcError::InvalidParams(
1108 "subject authority must be a did, not a handle".into(),
1109 ));
1110 }
1111 let Some(collection) = uri.collection() else {
1112 return Err(XrpcError::InvalidParams(
1113 "subject must be a bare did or full at://<did>/<nsid>/<rkey>".into(),
1114 ));
1115 };
1116 let c = collection.as_ref();
1117 match shape {
1118 SubjectShape::BareDid => Err(XrpcError::InvalidParams(format!(
1119 "subject must be a bare did, got at-uri with collection {c}"
1120 ))),
1121 SubjectShape::Collection(expected) if c == expected => {
1122 require_rkey(uri, expected)?;
1123 Ok(SubjectRef::Uri(uri.clone()))
1124 }
1125 SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!(
1126 "subject must be at://<did>/{expected}/<rkey>, got collection {c}"
1127 ))),
1128 SubjectShape::OneOfCollections(allowed) if allowed.contains(&c) => {
1129 require_rkey(uri, c)?;
1130 Ok(SubjectRef::Uri(uri.clone()))
1131 }
1132 SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
1133 "subject must be at://<did>/<nsid>/<rkey> with nsid in [{}], got collection {c}",
1134 allowed.join(", "),
1135 ))),
1136 SubjectShape::BareDidOrOneOfCollections(allowed) if allowed.contains(&c) => {
1137 require_rkey(uri, c)?;
1138 Ok(SubjectRef::Uri(uri.clone()))
1139 }
1140 SubjectShape::BareDidOrOneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
1141 "subject must be a bare did or at://<did>/<nsid>/<rkey> with nsid in [{}], got collection {c}",
1142 allowed.join(", "),
1143 ))),
1144 SubjectShape::AnyAtUri => Ok(SubjectRef::Uri(uri.clone())),
1145 }
1146}
1147
1148fn require_rkey(uri: &AtUri<DefaultStr>, expected: &str) -> Result<(), XrpcError> {
1149 uri.rkey().map(|_| ()).ok_or_else(|| {
1150 XrpcError::InvalidParams(format!(
1151 "subject must be at://<did>/{expected}/<rkey>; missing rkey"
1152 ))
1153 })
1154}
1155
1156fn parse_cursor(raw: Option<&str>) -> Result<PageCursor, XrpcError> {
1157 PageCursor::from_token(raw)
1158 .map_err(|e: CursorParseError| XrpcError::InvalidParams(format!("cursor: {e}")))
1159}
1160
1161fn parse_limit(raw: Option<u32>) -> Result<PageLimit, XrpcError> {
1162 PageLimit::new(raw.unwrap_or(DEFAULT_LIMIT))
1163 .map_err(|e| XrpcError::InvalidParams(format!("limit: {e}")))
1164}
1165
1166pub(crate) fn at_uri_owned_by(uri: &AtUri<DefaultStr>, author: &Did<DefaultStr>) -> bool {
1167 match uri.authority() {
1168 AtIdentifier::Did(d) => d.as_ref() == author.as_ref(),
1169 AtIdentifier::Handle(_) => false,
1170 }
1171}
1172
1173async fn resolve_for_view(
1174 state: &AppState,
1175 expected_nsid: &Nsid<DefaultStr>,
1176 uri: AtUri<DefaultStr>,
1177) -> Result<Arc<RecordBody>, XrpcError> {
1178 let raw = uri.as_ref().to_owned();
1179 resolve(state, ExpectedNsid::new(expected_nsid.clone()), uri)
1180 .await
1181 .map(|(body, _did)| body)
1182 .map_err(|e| match e {
1183 XrpcError::NotFound => XrpcError::UpstreamGone(raw),
1184 other => other,
1185 })
1186}
1187
1188async fn resolve(
1189 state: &AppState,
1190 expected: ExpectedNsid,
1191 uri: AtUri<DefaultStr>,
1192) -> Result<(Arc<RecordBody>, Did<DefaultStr>), XrpcError> {
1193 let collection = uri
1194 .collection()
1195 .ok_or_else(|| XrpcError::InvalidParams("uri missing collection".into()))?;
1196 if !expected.accepts(collection.as_ref()) {
1197 return Err(XrpcError::InvalidParams(format!(
1198 "collection mismatch: expected {}, got {}",
1199 expected.as_str(),
1200 collection.as_ref()
1201 )));
1202 }
1203 let rkey = uri
1204 .rkey()
1205 .ok_or_else(|| XrpcError::InvalidParams("uri missing rkey".into()))?;
1206 let did_ref = match uri.authority() {
1207 AtIdentifier::Did(d) => d,
1208 AtIdentifier::Handle(_) => {
1209 return Err(XrpcError::InvalidParams(
1210 "uri authority must be a did, not a handle".into(),
1211 ));
1212 }
1213 };
1214 let did: Did<DefaultStr> = did_ref.clone().into_static();
1215
1216 if let Some(hit) = state.records.get(&uri) {
1217 return Ok((hit, did));
1218 }
1219 let body = state
1220 .slingshot
1221 .get_record(&did_ref, &collection, &rkey)
1222 .await
1223 .map_err(map_slingshot)?;
1224 verify_type_tag(&body, &expected)?;
1225 state.records.put(uri, body.clone());
1226 Ok((body, did))
1227}
1228
1229#[derive(Deserialize)]
1230struct TypeTag<'a> {
1231 #[serde(rename = "$type", borrow)]
1232 ty: &'a str,
1233}
1234
1235fn verify_type_tag(body: &RecordBody, expected: &ExpectedNsid) -> Result<(), XrpcError> {
1236 let bytes = body.value.as_ref();
1237 let ty: std::borrow::Cow<'_, str> = match serde_json::from_slice::<TypeTag>(bytes) {
1238 Ok(t) => std::borrow::Cow::Borrowed(t.ty),
1239 Err(_) => {
1240 let value: serde_json::Value = serde_json::from_slice(bytes)
1241 .map_err(|e| XrpcError::InvalidRecord(format!("$type peek: {e}")))?;
1242 value
1243 .as_object()
1244 .and_then(|m| m.get("$type"))
1245 .and_then(|v| v.as_str())
1246 .map(|s| std::borrow::Cow::Owned(s.to_owned()))
1247 .ok_or_else(|| XrpcError::InvalidRecord("$type peek: missing $type field".into()))?
1248 }
1249 };
1250 if !expected.accepts(ty.as_ref()) {
1251 return Err(XrpcError::InvalidRecord(format!(
1252 "$type mismatch: expected {}, got {}",
1253 expected.as_str(),
1254 ty
1255 )));
1256 }
1257 Ok(())
1258}
1259
1260fn wire_type_nsid(bytes: &[u8]) -> Option<Nsid<DefaultStr>> {
1261 let ty = serde_json::from_slice::<TypeTag>(bytes).ok()?.ty;
1262 Nsid::<DefaultStr>::new_owned(ty).ok()
1263}
1264
1265async fn deserialize_or_upgrade<V>(
1266 state: &AppState,
1267 nsid: &Nsid<DefaultStr>,
1268 bytes: &[u8],
1269) -> Result<V, XrpcError>
1270where
1271 V: serde::de::DeserializeOwned,
1272{
1273 match serde_json::from_slice::<V>(bytes) {
1274 Ok(v) => Ok(v),
1275 Err(canon_err) => {
1276 let normalized = normalize_record_fields(bytes);
1277 let working: &[u8] = normalized.as_deref().unwrap_or(bytes);
1278 if normalized.is_some()
1279 && let Ok(v) = serde_json::from_slice::<V>(working)
1280 {
1281 return Ok(v);
1282 }
1283 let scrubbed = scrub_record_bytes(nsid, working);
1284 let retry_bytes: &[u8] = scrubbed.as_deref().unwrap_or(working);
1285 if scrubbed.is_some()
1286 && let Ok(v) = serde_json::from_slice::<V>(retry_bytes)
1287 {
1288 return Ok(v);
1289 }
1290 let wire_nsid = wire_type_nsid(retry_bytes).unwrap_or_else(|| nsid.clone());
1291 match upgrade_wire_bytes(&wire_nsid, retry_bytes, &state.resolver).await {
1292 Ok(canon_bytes) => serde_json::from_slice(&canon_bytes)
1293 .map_err(|e| XrpcError::InvalidRecord(e.to_string())),
1294 Err(_) => Err(XrpcError::InvalidRecord(canon_err.to_string())),
1295 }
1296 }
1297 }
1298}
1299
1300async fn fetch_from_uri<R, V>(
1301 state: &AppState,
1302 uri: AtUri<DefaultStr>,
1303) -> Result<(Arc<RecordBody>, V), XrpcError>
1304where
1305 R: XrpcResp,
1306 V: serde::de::DeserializeOwned + NormalizeRepoRefs,
1307{
1308 let raw = uri.as_str().to_owned();
1309 let nsid = nsid_static(R::NSID);
1310 let (body, _did) = resolve(state, ExpectedNsid::new(nsid.clone()), uri).await?;
1311 let value: V = deserialize_or_upgrade(state, &nsid, &body.value).await?;
1312 let value = value
1313 .normalize(&state.resolver)
1314 .await
1315 .ok_or(XrpcError::UpstreamGone(raw))?;
1316 Ok((body, value))
1317}
1318
1319async fn fetch<R, V>(
1320 state: &AppState,
1321 uri: &AtUri<DefaultStr>,
1322) -> Result<(Arc<RecordBody>, V), XrpcError>
1323where
1324 R: XrpcResp,
1325 V: serde::de::DeserializeOwned + NormalizeRepoRefs,
1326{
1327 fetch_from_uri::<R, V>(state, uri.clone()).await
1328}
1329
1330async fn get_repo(
1331 State(state): State<AppState>,
1332 XrpcQuery(q): XrpcQuery<GetRepoQuery>,
1333) -> Result<Json<RepoGetRecordOutput<DefaultStr>>, XrpcError> {
1334 let (body, value) = fetch::<RepoRecord, Repo<DefaultStr>>(&state, &q.repo).await?;
1335 Ok(Json(RepoGetRecordOutput {
1336 cid: Some(body.cid.clone()),
1337 uri: body.uri.clone(),
1338 value,
1339 }))
1340}
1341
1342async fn get_repo_by_repo_did(
1343 State(state): State<AppState>,
1344 XrpcQuery(q): XrpcQuery<GetRepoByRepoDidQuery>,
1345) -> Result<Json<RepoGetRecordOutput<DefaultStr>>, XrpcError> {
1346 let ident = state
1347 .resolver
1348 .lookup_by_repo_did(&q.repo_did)
1349 .await
1350 .ok_or(XrpcError::NotFound)?;
1351 let uri = AtUri::<DefaultStr>::from_parts_owned(
1352 ident.owner.as_str(),
1353 RepoRecord::NSID,
1354 ident.rkey.as_str(),
1355 )
1356 .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail");
1357 let (body, value) = fetch_from_uri::<RepoRecord, Repo<DefaultStr>>(&state, uri).await?;
1358 Ok(Json(RepoGetRecordOutput {
1359 cid: Some(body.cid.clone()),
1360 uri: body.uri.clone(),
1361 value,
1362 }))
1363}
1364
1365async fn get_profile(
1366 State(state): State<AppState>,
1367 XrpcQuery(q): XrpcQuery<GetProfileQuery>,
1368) -> Result<Json<ProfileGetRecordOutput<DefaultStr>>, XrpcError> {
1369 let (body, value) = fetch::<ProfileRecord, Profile<DefaultStr>>(&state, &q.actor).await?;
1370 Ok(Json(ProfileGetRecordOutput {
1371 cid: Some(body.cid.clone()),
1372 uri: body.uri.clone(),
1373 value,
1374 }))
1375}
1376
1377async fn get_issue(
1378 State(state): State<AppState>,
1379 XrpcQuery(q): XrpcQuery<GetIssueQuery>,
1380) -> Result<Json<IssueGetRecordOutput<DefaultStr>>, XrpcError> {
1381 let (body, value) = fetch::<IssueRecord, Issue<DefaultStr>>(&state, &q.issue).await?;
1382 Ok(Json(IssueGetRecordOutput {
1383 cid: Some(body.cid.clone()),
1384 uri: body.uri.clone(),
1385 value,
1386 }))
1387}
1388
1389async fn get_pull(
1390 State(state): State<AppState>,
1391 XrpcQuery(q): XrpcQuery<GetPullQuery>,
1392) -> Result<Json<PullGetRecordOutput<DefaultStr>>, XrpcError> {
1393 let (body, value) = fetch::<PullRecord, Pull<DefaultStr>>(&state, &q.pull).await?;
1394 Ok(Json(PullGetRecordOutput {
1395 cid: Some(body.cid.clone()),
1396 uri: body.uri.clone(),
1397 value,
1398 }))
1399}
1400
1401async fn get_repos(
1402 State(state): State<AppState>,
1403 RawQuery(query): RawQuery,
1404) -> Result<Response, XrpcError> {
1405 let uris = collect_repeated(query.as_deref(), BULK_REPOS_KEY);
1406 bulk_fetch::<RepoRecord, Repo<DefaultStr>>(&state, uris).await
1407}
1408
1409async fn get_profiles(
1410 State(state): State<AppState>,
1411 RawQuery(query): RawQuery,
1412) -> Result<Response, XrpcError> {
1413 let uris = collect_repeated(query.as_deref(), BULK_PROFILES_KEY);
1414 bulk_fetch::<ProfileRecord, Profile<DefaultStr>>(&state, uris).await
1415}
1416
1417async fn get_issues(
1418 State(state): State<AppState>,
1419 RawQuery(query): RawQuery,
1420) -> Result<Response, XrpcError> {
1421 let uris = collect_repeated(query.as_deref(), BULK_ISSUES_KEY);
1422 bulk_fetch::<IssueRecord, Issue<DefaultStr>>(&state, uris).await
1423}
1424
1425async fn get_pulls(
1426 State(state): State<AppState>,
1427 RawQuery(query): RawQuery,
1428) -> Result<Response, XrpcError> {
1429 let uris = collect_repeated(query.as_deref(), BULK_PULLS_KEY);
1430 bulk_fetch::<PullRecord, Pull<DefaultStr>>(&state, uris).await
1431}
1432
1433const BULK_REPOS_KEY: &str = "repos";
1434const BULK_PROFILES_KEY: &str = "actors";
1435const BULK_ISSUES_KEY: &str = "issues";
1436const BULK_PULLS_KEY: &str = "pulls";
1437const BULK_LIMIT: usize = 50;
1438
1439fn collect_repeated(query: Option<&str>, key: &str) -> Vec<String> {
1440 let Some(q) = query else {
1441 return Vec::new();
1442 };
1443 form_urlencoded::parse(q.as_bytes())
1444 .filter_map(|(k, v)| (k == key).then(|| v.into_owned()))
1445 .collect()
1446}
1447
1448async fn hydrate_record_view<V>(
1449 state: &AppState,
1450 nsid: &Nsid<DefaultStr>,
1451 uri: AtUri<DefaultStr>,
1452 sort_micros: u64,
1453) -> Result<Option<RecordView<V>>, XrpcError>
1454where
1455 V: serde::de::DeserializeOwned + NormalizeRepoRefs,
1456{
1457 if let Some(source) = decode_knot_owned_source(&uri) {
1458 return synthesize_knot_owned_view::<V>(state, uri, source, sort_micros).await;
1459 }
1460 let body = resolve_for_view(state, nsid, uri).await?;
1461 let value: V = deserialize_or_upgrade::<V>(state, nsid, &body.value).await?;
1462 let Some(value) = value.normalize(&state.resolver).await else {
1463 return Ok(None);
1464 };
1465 Ok(Some(RecordView {
1466 uri: body.uri.clone(),
1467 cid: Some(body.cid.clone()),
1468 value,
1469 }))
1470}
1471
1472async fn synthesize_knot_owned_view<V>(
1473 state: &AppState,
1474 uri: AtUri<DefaultStr>,
1475 source: KnotOwnedSource,
1476 sort_micros: u64,
1477) -> Result<Option<RecordView<V>>, XrpcError>
1478where
1479 V: serde::de::DeserializeOwned + NormalizeRepoRefs,
1480{
1481 let Some(body) = synth_knot_owned_value(source, sort_micros) else {
1482 return Ok(None);
1483 };
1484 let Ok(value) = serde_json::from_value::<V>(body) else {
1485 return Ok(None);
1486 };
1487 let Some(value) = value.normalize(&state.resolver).await else {
1488 return Ok(None);
1489 };
1490 Ok(Some(RecordView {
1491 uri,
1492 cid: None,
1493 value,
1494 }))
1495}
1496
1497fn synth_knot_owned_value(source: KnotOwnedSource, sort_micros: u64) -> Option<serde_json::Value> {
1498 let created_at = micros_to_rfc3339(sort_micros)?;
1499 match source {
1500 KnotOwnedSource::Member { knot, subject } => Some(serde_json::json!({
1501 "domain": knot_did_host(&knot)?,
1502 "subject": subject.as_ref(),
1503 "createdAt": created_at,
1504 })),
1505 KnotOwnedSource::Collaborator { repo, subject } => Some(serde_json::json!({
1506 "repo": repo.as_ref(),
1507 "subject": subject.as_ref(),
1508 "createdAt": created_at,
1509 })),
1510 }
1511}
1512
1513fn micros_to_rfc3339(micros: u64) -> Option<String> {
1514 let micros = i64::try_from(micros).ok()?;
1515 chrono::DateTime::from_timestamp_micros(micros)
1516 .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Micros, true))
1517}
1518
1519#[derive(Clone, Copy)]
1520enum HitProvenance {
1521 ClientSupplied,
1522 Indexed,
1523}
1524
1525fn is_index_evictable(err: &XrpcError) -> bool {
1526 matches!(
1527 err,
1528 XrpcError::NotFound
1529 | XrpcError::UpstreamGone(_)
1530 | XrpcError::InvalidRecord(_)
1531 | XrpcError::InvalidParams(_)
1532 )
1533}
1534
1535fn drop_unhydratable<V>(
1536 provenance: HitProvenance,
1537 nsid: &Nsid<DefaultStr>,
1538 uri: &AtUri<DefaultStr>,
1539 result: Result<Option<V>, XrpcError>,
1540) -> Result<Option<V>, XrpcError> {
1541 match result {
1542 Ok(view) => Ok(view),
1543 Err(err @ (XrpcError::NotFound | XrpcError::UpstreamGone(_))) => {
1544 tracing::debug!(
1545 uri = %uri,
1546 nsid = %nsid.as_ref(),
1547 error = %err,
1548 "dropping gone hit during hydration",
1549 );
1550 Ok(None)
1551 }
1552 Err(err @ XrpcError::UpstreamUnavailable(_)) => {
1553 tracing::warn!(
1554 uri = %uri,
1555 nsid = %nsid.as_ref(),
1556 error = %err,
1557 "dropping hit, upstream unavailable during hydration",
1558 );
1559 Ok(None)
1560 }
1561 Err(err @ XrpcError::InvalidRecord(_)) => {
1562 tracing::warn!(
1563 uri = %uri,
1564 nsid = %nsid.as_ref(),
1565 error = %err,
1566 "dropping invalid hit during hydration",
1567 );
1568 Ok(None)
1569 }
1570 Err(err @ XrpcError::InvalidParams(_)) => match provenance {
1571 HitProvenance::ClientSupplied => Err(err),
1572 HitProvenance::Indexed => {
1573 tracing::warn!(
1574 uri = %uri,
1575 nsid = %nsid.as_ref(),
1576 error = %err,
1577 "dropping malformed indexed hit during hydration",
1578 );
1579 Ok(None)
1580 }
1581 },
1582 Err(err @ (XrpcError::Internal(_) | XrpcError::Overloaded)) => Err(err),
1583 }
1584}
1585
1586fn hydrate_stream<T, Fut, V>(
1587 items: impl IntoIterator<Item = T>,
1588 produce: impl FnMut(T) -> Fut,
1589) -> impl Stream<Item = Result<V, XrpcError>>
1590where
1591 Fut: Future<Output = Result<Option<V>, XrpcError>>,
1592{
1593 stream::iter(items)
1594 .map(produce)
1595 .buffered(FETCH_CONCURRENCY)
1596 .try_filter_map(|view| async move { Ok(view) })
1597}
1598
1599fn hydrate_record_stream<V>(
1600 state: &AppState,
1601 nsid: Nsid<DefaultStr>,
1602 items: Vec<EdgeItem>,
1603 provenance: HitProvenance,
1604) -> impl Stream<Item = Result<RecordView<V>, XrpcError>> + Send + 'static
1605where
1606 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1607{
1608 let owned = state.clone();
1609 hydrate_stream(items, move |item| {
1610 let owned = owned.clone();
1611 let nsid = nsid.clone();
1612 async move {
1613 let EdgeItem { uri, sort_micros } = item;
1614 let result = hydrate_record_view::<V>(&owned, &nsid, uri.clone(), sort_micros).await;
1615 if matches!(provenance, HitProvenance::Indexed)
1616 && let Err(err) = &result
1617 && is_index_evictable(err)
1618 {
1619 owned.edges.remove_source(&uri);
1620 }
1621 drop_unhydratable(provenance, &nsid, &uri, result)
1622 }
1623 })
1624}
1625
1626enum PagePhase {
1627 Head,
1628 Body { first: bool },
1629 Done,
1630}
1631
1632struct PageState<S> {
1633 items: std::pin::Pin<Box<S>>,
1634 phase: PagePhase,
1635 array_key: &'static str,
1636 tail: Vec<u8>,
1637 permit: Option<HeavyPermit>,
1638}
1639
1640fn paged_tail(cursor: Option<String>) -> Vec<u8> {
1641 let encoded = serde_json::to_string(&cursor).unwrap_or_else(|_| "null".to_owned());
1642 format!("],\"cursor\":{encoded}}}").into_bytes()
1643}
1644
1645fn unpaged_tail() -> Vec<u8> {
1646 b"]}".to_vec()
1647}
1648
1649fn json_stream<V, S>(
1650 array_key: &'static str,
1651 items: S,
1652 tail: Vec<u8>,
1653 permit: Option<HeavyPermit>,
1654) -> Response
1655where
1656 V: Serialize + Send + 'static,
1657 S: Stream<Item = Result<V, XrpcError>> + Send + 'static,
1658{
1659 let init = PageState {
1660 items: Box::pin(items),
1661 phase: PagePhase::Head,
1662 array_key,
1663 tail,
1664 permit,
1665 };
1666 let chunks = stream::unfold(init, |mut st| async move {
1667 match st.phase {
1668 PagePhase::Head => {
1669 let head = format!("{{\"{}\":[", st.array_key).into_bytes();
1670 st.phase = PagePhase::Body { first: true };
1671 Some((Ok::<Vec<u8>, Infallible>(head), st))
1672 }
1673 PagePhase::Body { first } => match st.items.next().await {
1674 Some(Ok(view)) => match serde_json::to_vec(&view) {
1675 Ok(encoded) => {
1676 let mut chunk = Vec::with_capacity(encoded.len() + 1);
1677 if !first {
1678 chunk.push(b',');
1679 }
1680 chunk.extend_from_slice(&encoded);
1681 st.phase = PagePhase::Body { first: false };
1682 Some((Ok(chunk), st))
1683 }
1684 Err(e) => {
1685 tracing::warn!(error = %e, "skipping hit, serialize failed mid-stream");
1686 Some((Ok(Vec::new()), st))
1687 }
1688 },
1689 Some(Err(e)) => {
1690 tracing::warn!(error = %e, "ending page early, hydration failed mid-stream");
1691 let tail = std::mem::take(&mut st.tail);
1692 st.phase = PagePhase::Done;
1693 Some((Ok(tail), st))
1694 }
1695 None => {
1696 let tail = std::mem::take(&mut st.tail);
1697 st.phase = PagePhase::Done;
1698 Some((Ok(tail), st))
1699 }
1700 },
1701 PagePhase::Done => {
1702 drop(st.permit.take());
1703 None
1704 }
1705 }
1706 });
1707 (
1708 [(CONTENT_TYPE, "application/json")],
1709 Body::from_stream(chunks),
1710 )
1711 .into_response()
1712}
1713
1714async fn bulk_fetch<R, V>(state: &AppState, uris: Vec<String>) -> Result<Response, XrpcError>
1715where
1716 R: XrpcResp,
1717 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1718{
1719 if uris.is_empty() {
1720 return Err(XrpcError::InvalidParams("at least one uri required".into()));
1721 }
1722 if uris.len() > BULK_LIMIT {
1723 return Err(XrpcError::InvalidParams(format!(
1724 "at most {BULK_LIMIT} uris per request"
1725 )));
1726 }
1727 let parsed: Vec<AtUri<DefaultStr>> = uris
1728 .iter()
1729 .map(|s| parse_uri(s))
1730 .collect::<Result<_, _>>()?;
1731 let nsid = nsid_static(R::NSID);
1732 if let Some(bad) = parsed
1733 .iter()
1734 .find(|uri| uri.collection().is_none_or(|c| c.as_ref() != nsid.as_ref()))
1735 {
1736 return Err(XrpcError::InvalidParams(format!(
1737 "uri collection must be {}, got {}",
1738 nsid.as_ref(),
1739 bad.as_ref()
1740 )));
1741 }
1742 let permit = state.heavy_permit()?;
1743 let items = parsed
1744 .into_iter()
1745 .map(|uri| EdgeItem {
1746 uri,
1747 sort_micros: 0,
1748 })
1749 .collect();
1750 let views = hydrate_record_stream::<V>(state, nsid, items, HitProvenance::ClientSupplied);
1751 Ok(json_stream::<RecordView<V>, _>(
1752 "items",
1753 views,
1754 unpaged_tail(),
1755 permit,
1756 ))
1757}
1758
1759fn record_edge_page<R, F>(
1760 state: &AppState,
1761 q: &TypedListQuery<F>,
1762) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>
1763where
1764 R: XrpcResp + HasSubject,
1765 F: ListFilter,
1766{
1767 let subject = parse_subject(&q.subject, R::SHAPE)?;
1768 let cursor = parse_cursor(q.cursor.as_deref())?;
1769 let limit = parse_limit(q.limit)?;
1770 let dir = q.dir();
1771 let nsid = nsid_static(R::NSID);
1772 let page = if q.filter.is_identity() {
1773 let key = EdgeKey::new(nsid.clone(), subject);
1774 state.edges.list(&key, cursor, limit, dir)
1775 } else {
1776 let pred = q.filter.predicate(state, &subject);
1777 let key = EdgeKey::new(nsid.clone(), subject);
1778 state.edges.list_filtered(&key, cursor, limit, dir, pred)
1779 };
1780 Ok((page, nsid))
1781}
1782
1783async fn list_records<R, V, F>(
1784 state: &AppState,
1785 q: TypedListQuery<F>,
1786) -> Result<Response, XrpcError>
1787where
1788 R: XrpcResp + HasSubject,
1789 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1790 F: ListFilter,
1791{
1792 let (page, nsid) = record_edge_page::<R, F>(state, &q)?;
1793 let permit = state.heavy_permit()?;
1794 let views = hydrate_record_stream::<V>(state, nsid, page.items, HitProvenance::Indexed);
1795 Ok(json_stream::<RecordView<V>, _>(
1796 "items",
1797 views,
1798 paged_tail(page.next.map(PageToken::encode_token)),
1799 permit,
1800 ))
1801}
1802
1803fn count_for<R: XrpcResp + HasSubject>(
1804 state: &AppState,
1805 q: CountQuery,
1806) -> Result<CountResponse, XrpcError> {
1807 let subject = parse_subject(&q.subject, R::SHAPE)?;
1808 let key = EdgeKey::new(nsid_static(R::NSID), subject);
1809 Ok(CountResponse {
1810 count: state.edges.count(&key),
1811 distinct_authors: state.edges.count_distinct_authors(&key),
1812 })
1813}
1814
1815// does `actor` have an edge of this kind pointing at `subject`, returns its own uri
1816fn get_for<R: XrpcResp + HasSubject>(
1817 state: &AppState,
1818 q: GetEdgeQuery,
1819) -> Result<EdgeUriResponse, XrpcError> {
1820 let subject = parse_subject(&q.subject, R::SHAPE)?;
1821 let key = EdgeKey::new(nsid_static(R::NSID), subject);
1822 let uri = state
1823 .edges
1824 .viewer_source(&key, q.actor.as_str())
1825 .ok_or(XrpcError::NotFound)?;
1826 Ok(EdgeUriResponse { uri })
1827}
1828
1829fn mirror_edge_page<M, F>(
1830 state: &AppState,
1831 q: &TypedListQuery<F>,
1832) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>
1833where
1834 M: MirrorOf,
1835 F: ListFilter,
1836{
1837 let subject = parse_subject(&q.subject, M::SHAPE)?;
1838 let cursor = parse_cursor(q.cursor.as_deref())?;
1839 let limit = parse_limit(q.limit)?;
1840 let dir = q.dir();
1841 let edge_nsid = nsid_static(M::EDGE_KIND);
1842 let page = if q.filter.is_identity() {
1843 let key = EdgeKey::new(edge_nsid, subject);
1844 state.edges.list(&key, cursor, limit, dir)
1845 } else {
1846 let pred = q.filter.predicate(state, &subject);
1847 let key = EdgeKey::new(edge_nsid, subject);
1848 state.edges.list_filtered(&key, cursor, limit, dir, pred)
1849 };
1850 let record_nsid = nsid_static(<M::Record as XrpcResp>::NSID);
1851 Ok((page, record_nsid))
1852}
1853
1854async fn list_mirror<M, V, F>(state: &AppState, q: TypedListQuery<F>) -> Result<Response, XrpcError>
1855where
1856 M: MirrorOf,
1857 V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
1858 F: ListFilter,
1859{
1860 let (page, record_nsid) = mirror_edge_page::<M, F>(state, &q)?;
1861 let permit = state.heavy_permit()?;
1862 let views = hydrate_record_stream::<V>(state, record_nsid, page.items, HitProvenance::Indexed);
1863 Ok(json_stream::<RecordView<V>, _>(
1864 "items",
1865 views,
1866 paged_tail(page.next.map(PageToken::encode_token)),
1867 permit,
1868 ))
1869}
1870
1871fn count_mirror<M: MirrorOf>(state: &AppState, q: CountQuery) -> Result<CountResponse, XrpcError> {
1872 let subject = parse_subject(&q.subject, M::SHAPE)?;
1873 let key = EdgeKey::new(nsid_static(M::EDGE_KIND), subject);
1874 Ok(CountResponse {
1875 count: state.edges.count(&key),
1876 distinct_authors: state.edges.count_distinct_authors(&key),
1877 })
1878}
1879
1880async fn list_stars(
1881 State(state): State<AppState>,
1882 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
1883) -> Result<Response, XrpcError> {
1884 list_records::<StarRecord, Star<DefaultStr>, _>(&state, q).await
1885}
1886
1887async fn count_stars(
1888 State(state): State<AppState>,
1889 XrpcQuery(q): XrpcQuery<CountQuery>,
1890) -> Result<Json<CountResponse>, XrpcError> {
1891 count_for::<StarRecord>(&state, q).map(Json)
1892}
1893
1894async fn get_star(
1895 State(state): State<AppState>,
1896 XrpcQuery(q): XrpcQuery<GetEdgeQuery>,
1897) -> Result<Json<EdgeUriResponse>, XrpcError> {
1898 get_for::<StarRecord>(&state, q).map(Json)
1899}
1900
1901async fn list_follows(
1902 State(state): State<AppState>,
1903 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
1904) -> Result<Response, XrpcError> {
1905 list_records::<FollowRecord, Follow<DefaultStr>, _>(&state, q).await
1906}
1907
1908async fn count_follows(
1909 State(state): State<AppState>,
1910 XrpcQuery(q): XrpcQuery<CountQuery>,
1911) -> Result<Json<CountResponse>, XrpcError> {
1912 count_for::<FollowRecord>(&state, q).map(Json)
1913}
1914
1915async fn get_follow(
1916 State(state): State<AppState>,
1917 XrpcQuery(q): XrpcQuery<GetEdgeQuery>,
1918) -> Result<Json<EdgeUriResponse>, XrpcError> {
1919 get_for::<FollowRecord>(&state, q).map(Json)
1920}
1921
1922async fn list_issues(
1923 State(state): State<AppState>,
1924 XrpcQuery(q): XrpcQuery<TypedListQuery<IssueFilter>>,
1925) -> Result<Response, XrpcError> {
1926 let (page, nsid) = record_edge_page::<IssueRecord, _>(&state, &q)?;
1927 let permit = state.heavy_permit()?;
1928 let owned = state.clone();
1929 let items = hydrate_record_stream::<Issue<DefaultStr>>(
1930 &state,
1931 nsid,
1932 page.items,
1933 HitProvenance::Indexed,
1934 )
1935 .map(move |view| view.map(|v| enrich_issue_view(&owned, v)));
1936 Ok(json_stream::<StatefulItem<Issue<DefaultStr>>, _>(
1937 "items",
1938 items,
1939 paged_tail(page.next.map(PageToken::encode_token)),
1940 permit,
1941 ))
1942}
1943
1944async fn count_issues(
1945 State(state): State<AppState>,
1946 XrpcQuery(q): XrpcQuery<CountQuery>,
1947) -> Result<Json<CountResponse>, XrpcError> {
1948 count_for::<IssueRecord>(&state, q).map(Json)
1949}
1950
1951async fn list_pulls(
1952 State(state): State<AppState>,
1953 XrpcQuery(q): XrpcQuery<TypedListQuery<PullFilter>>,
1954) -> Result<Response, XrpcError> {
1955 let (page, nsid) = record_edge_page::<PullRecord, _>(&state, &q)?;
1956 let permit = state.heavy_permit()?;
1957 let owned = state.clone();
1958 let items =
1959 hydrate_record_stream::<Pull<DefaultStr>>(&state, nsid, page.items, HitProvenance::Indexed)
1960 .map(move |view| view.map(|v| enrich_pull_view(&owned, v)));
1961 Ok(json_stream::<StatefulItem<Pull<DefaultStr>>, _>(
1962 "items",
1963 items,
1964 paged_tail(page.next.map(PageToken::encode_token)),
1965 permit,
1966 ))
1967}
1968
1969async fn count_pulls(
1970 State(state): State<AppState>,
1971 XrpcQuery(q): XrpcQuery<CountQuery>,
1972) -> Result<Json<CountResponse>, XrpcError> {
1973 count_for::<PullRecord>(&state, q).map(Json)
1974}
1975
1976async fn list_feed_comments(
1977 State(state): State<AppState>,
1978 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
1979) -> Result<Response, XrpcError> {
1980 list_records::<FeedCommentRecord, FeedComment<DefaultStr>, _>(&state, q).await
1981}
1982
1983async fn count_feed_comments(
1984 State(state): State<AppState>,
1985 XrpcQuery(q): XrpcQuery<CountQuery>,
1986) -> Result<Json<CountResponse>, XrpcError> {
1987 count_for::<FeedCommentRecord>(&state, q).map(Json)
1988}
1989
1990async fn list_reactions(
1991 State(state): State<AppState>,
1992 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
1993) -> Result<Response, XrpcError> {
1994 list_records::<ReactionRecord, Reaction<DefaultStr>, _>(&state, q).await
1995}
1996
1997async fn count_reactions(
1998 State(state): State<AppState>,
1999 XrpcQuery(q): XrpcQuery<CountQuery>,
2000) -> Result<Json<CountResponse>, XrpcError> {
2001 count_for::<ReactionRecord>(&state, q).map(Json)
2002}
2003
2004async fn list_ref_updates(
2005 State(state): State<AppState>,
2006 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2007) -> Result<Response, XrpcError> {
2008 list_records::<RefUpdateRecord, RefUpdate<DefaultStr>, _>(&state, q).await
2009}
2010
2011async fn count_ref_updates(
2012 State(state): State<AppState>,
2013 XrpcQuery(q): XrpcQuery<CountQuery>,
2014) -> Result<Json<CountResponse>, XrpcError> {
2015 count_for::<RefUpdateRecord>(&state, q).map(Json)
2016}
2017
2018async fn list_collaborators(
2019 State(state): State<AppState>,
2020 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2021) -> Result<Response, XrpcError> {
2022 list_records::<CollaboratorRecord, Collaborator<DefaultStr>, _>(&state, q).await
2023}
2024
2025async fn count_collaborators(
2026 State(state): State<AppState>,
2027 XrpcQuery(q): XrpcQuery<CountQuery>,
2028) -> Result<Json<CountResponse>, XrpcError> {
2029 count_for::<CollaboratorRecord>(&state, q).map(Json)
2030}
2031
2032async fn list_issue_states(
2033 State(state): State<AppState>,
2034 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2035) -> Result<Response, XrpcError> {
2036 list_records::<IssueStateRecord, IssueState<DefaultStr>, _>(&state, q).await
2037}
2038
2039async fn count_issue_states(
2040 State(state): State<AppState>,
2041 XrpcQuery(q): XrpcQuery<CountQuery>,
2042) -> Result<Json<CountResponse>, XrpcError> {
2043 count_for::<IssueStateRecord>(&state, q).map(Json)
2044}
2045
2046async fn list_pull_statuses(
2047 State(state): State<AppState>,
2048 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2049) -> Result<Response, XrpcError> {
2050 list_records::<PullStatusRecord, PullStatus<DefaultStr>, _>(&state, q).await
2051}
2052
2053async fn count_pull_statuses(
2054 State(state): State<AppState>,
2055 XrpcQuery(q): XrpcQuery<CountQuery>,
2056) -> Result<Json<CountResponse>, XrpcError> {
2057 count_for::<PullStatusRecord>(&state, q).map(Json)
2058}
2059
2060async fn list_repos(
2061 State(state): State<AppState>,
2062 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2063) -> Result<Response, XrpcError> {
2064 list_records::<RepoRecord, Repo<DefaultStr>, _>(&state, q).await
2065}
2066
2067async fn count_repos(
2068 State(state): State<AppState>,
2069 XrpcQuery(q): XrpcQuery<CountQuery>,
2070) -> Result<Json<CountResponse>, XrpcError> {
2071 count_for::<RepoRecord>(&state, q).map(Json)
2072}
2073
2074async fn list_knots(
2075 State(state): State<AppState>,
2076 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2077) -> Result<Response, XrpcError> {
2078 list_records::<KnotRecord, Knot<DefaultStr>, _>(&state, q).await
2079}
2080
2081async fn count_knots(
2082 State(state): State<AppState>,
2083 XrpcQuery(q): XrpcQuery<CountQuery>,
2084) -> Result<Json<CountResponse>, XrpcError> {
2085 count_for::<KnotRecord>(&state, q).map(Json)
2086}
2087
2088async fn list_spindles(
2089 State(state): State<AppState>,
2090 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2091) -> Result<Response, XrpcError> {
2092 list_records::<SpindleRecord, Spindle<DefaultStr>, _>(&state, q).await
2093}
2094
2095async fn count_spindles(
2096 State(state): State<AppState>,
2097 XrpcQuery(q): XrpcQuery<CountQuery>,
2098) -> Result<Json<CountResponse>, XrpcError> {
2099 count_for::<SpindleRecord>(&state, q).map(Json)
2100}
2101
2102async fn list_public_keys(
2103 State(state): State<AppState>,
2104 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2105) -> Result<Response, XrpcError> {
2106 list_records::<PublicKeyRecord, PublicKey<DefaultStr>, _>(&state, q).await
2107}
2108
2109async fn count_public_keys(
2110 State(state): State<AppState>,
2111 XrpcQuery(q): XrpcQuery<CountQuery>,
2112) -> Result<Json<CountResponse>, XrpcError> {
2113 count_for::<PublicKeyRecord>(&state, q).map(Json)
2114}
2115
2116async fn list_vouches(
2117 State(state): State<AppState>,
2118 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2119) -> Result<Response, XrpcError> {
2120 list_records::<VouchRecord, Vouch<DefaultStr>, _>(&state, q).await
2121}
2122
2123async fn count_vouches(
2124 State(state): State<AppState>,
2125 XrpcQuery(q): XrpcQuery<CountQuery>,
2126) -> Result<Json<CountResponse>, XrpcError> {
2127 count_for::<VouchRecord>(&state, q).map(Json)
2128}
2129
2130async fn list_stars_by(
2131 State(state): State<AppState>,
2132 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2133) -> Result<Response, XrpcError> {
2134 list_mirror::<StarBy, Star<DefaultStr>, _>(&state, q).await
2135}
2136async fn count_stars_by(
2137 State(state): State<AppState>,
2138 XrpcQuery(q): XrpcQuery<CountQuery>,
2139) -> Result<Json<CountResponse>, XrpcError> {
2140 count_mirror::<StarBy>(&state, q).map(Json)
2141}
2142
2143async fn list_reactions_by(
2144 State(state): State<AppState>,
2145 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2146) -> Result<Response, XrpcError> {
2147 list_mirror::<ReactionBy, Reaction<DefaultStr>, _>(&state, q).await
2148}
2149async fn count_reactions_by(
2150 State(state): State<AppState>,
2151 XrpcQuery(q): XrpcQuery<CountQuery>,
2152) -> Result<Json<CountResponse>, XrpcError> {
2153 count_mirror::<ReactionBy>(&state, q).map(Json)
2154}
2155
2156async fn list_follows_by(
2157 State(state): State<AppState>,
2158 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2159) -> Result<Response, XrpcError> {
2160 list_mirror::<FollowBy, Follow<DefaultStr>, _>(&state, q).await
2161}
2162async fn count_follows_by(
2163 State(state): State<AppState>,
2164 XrpcQuery(q): XrpcQuery<CountQuery>,
2165) -> Result<Json<CountResponse>, XrpcError> {
2166 count_mirror::<FollowBy>(&state, q).map(Json)
2167}
2168
2169async fn list_vouches_by(
2170 State(state): State<AppState>,
2171 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2172) -> Result<Response, XrpcError> {
2173 list_mirror::<VouchBy, Vouch<DefaultStr>, _>(&state, q).await
2174}
2175async fn count_vouches_by(
2176 State(state): State<AppState>,
2177 XrpcQuery(q): XrpcQuery<CountQuery>,
2178) -> Result<Json<CountResponse>, XrpcError> {
2179 count_mirror::<VouchBy>(&state, q).map(Json)
2180}
2181
2182async fn list_ref_updates_by(
2183 State(state): State<AppState>,
2184 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2185) -> Result<Response, XrpcError> {
2186 list_mirror::<RefUpdateBy, RefUpdate<DefaultStr>, _>(&state, q).await
2187}
2188async fn count_ref_updates_by(
2189 State(state): State<AppState>,
2190 XrpcQuery(q): XrpcQuery<CountQuery>,
2191) -> Result<Json<CountResponse>, XrpcError> {
2192 count_mirror::<RefUpdateBy>(&state, q).map(Json)
2193}
2194
2195async fn list_knot_members_by(
2196 State(state): State<AppState>,
2197 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2198) -> Result<Response, XrpcError> {
2199 list_mirror::<KnotMemberBy, KnotMember<DefaultStr>, _>(&state, q).await
2200}
2201async fn count_knot_members_by(
2202 State(state): State<AppState>,
2203 XrpcQuery(q): XrpcQuery<CountQuery>,
2204) -> Result<Json<CountResponse>, XrpcError> {
2205 count_mirror::<KnotMemberBy>(&state, q).map(Json)
2206}
2207
2208async fn list_label_ops_by(
2209 State(state): State<AppState>,
2210 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2211) -> Result<Response, XrpcError> {
2212 list_mirror::<LabelOpBy, LabelOp<DefaultStr>, _>(&state, q).await
2213}
2214async fn count_label_ops_by(
2215 State(state): State<AppState>,
2216 XrpcQuery(q): XrpcQuery<CountQuery>,
2217) -> Result<Json<CountResponse>, XrpcError> {
2218 count_mirror::<LabelOpBy>(&state, q).map(Json)
2219}
2220
2221async fn list_pipelines_by(
2222 State(state): State<AppState>,
2223 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2224) -> Result<Response, XrpcError> {
2225 list_mirror::<PipelineBy, Pipeline<DefaultStr>, _>(&state, q).await
2226}
2227async fn count_pipelines_by(
2228 State(state): State<AppState>,
2229 XrpcQuery(q): XrpcQuery<CountQuery>,
2230) -> Result<Json<CountResponse>, XrpcError> {
2231 count_mirror::<PipelineBy>(&state, q).map(Json)
2232}
2233
2234async fn list_pipeline_statuses_by(
2235 State(state): State<AppState>,
2236 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2237) -> Result<Response, XrpcError> {
2238 list_mirror::<PipelineStatusBy, PipelineStatus<DefaultStr>, _>(&state, q).await
2239}
2240async fn count_pipeline_statuses_by(
2241 State(state): State<AppState>,
2242 XrpcQuery(q): XrpcQuery<CountQuery>,
2243) -> Result<Json<CountResponse>, XrpcError> {
2244 count_mirror::<PipelineStatusBy>(&state, q).map(Json)
2245}
2246
2247async fn list_artifacts_by(
2248 State(state): State<AppState>,
2249 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2250) -> Result<Response, XrpcError> {
2251 list_mirror::<ArtifactBy, Artifact<DefaultStr>, _>(&state, q).await
2252}
2253async fn count_artifacts_by(
2254 State(state): State<AppState>,
2255 XrpcQuery(q): XrpcQuery<CountQuery>,
2256) -> Result<Json<CountResponse>, XrpcError> {
2257 count_mirror::<ArtifactBy>(&state, q).map(Json)
2258}
2259
2260async fn list_collaborators_by(
2261 State(state): State<AppState>,
2262 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2263) -> Result<Response, XrpcError> {
2264 list_mirror::<CollaboratorBy, Collaborator<DefaultStr>, _>(&state, q).await
2265}
2266async fn count_collaborators_by(
2267 State(state): State<AppState>,
2268 XrpcQuery(q): XrpcQuery<CountQuery>,
2269) -> Result<Json<CountResponse>, XrpcError> {
2270 count_mirror::<CollaboratorBy>(&state, q).map(Json)
2271}
2272
2273async fn list_issues_by(
2274 State(state): State<AppState>,
2275 XrpcQuery(q): XrpcQuery<TypedListQuery<IssueFilter>>,
2276) -> Result<Response, XrpcError> {
2277 let (page, nsid) = mirror_edge_page::<IssueBy, _>(&state, &q)?;
2278 let permit = state.heavy_permit()?;
2279 let owned = state.clone();
2280 let items = hydrate_record_stream::<Issue<DefaultStr>>(
2281 &state,
2282 nsid,
2283 page.items,
2284 HitProvenance::Indexed,
2285 )
2286 .map(move |view| view.map(|v| enrich_issue_view(&owned, v)));
2287 Ok(json_stream::<StatefulItem<Issue<DefaultStr>>, _>(
2288 "items",
2289 items,
2290 paged_tail(page.next.map(PageToken::encode_token)),
2291 permit,
2292 ))
2293}
2294async fn count_issues_by(
2295 State(state): State<AppState>,
2296 XrpcQuery(q): XrpcQuery<CountQuery>,
2297) -> Result<Json<CountResponse>, XrpcError> {
2298 count_mirror::<IssueBy>(&state, q).map(Json)
2299}
2300
2301async fn list_feed_comments_by(
2302 State(state): State<AppState>,
2303 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2304) -> Result<Response, XrpcError> {
2305 list_mirror::<FeedCommentBy, FeedComment<DefaultStr>, _>(&state, q).await
2306}
2307async fn count_feed_comments_by(
2308 State(state): State<AppState>,
2309 XrpcQuery(q): XrpcQuery<CountQuery>,
2310) -> Result<Json<CountResponse>, XrpcError> {
2311 count_mirror::<FeedCommentBy>(&state, q).map(Json)
2312}
2313
2314async fn list_issue_states_by(
2315 State(state): State<AppState>,
2316 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2317) -> Result<Response, XrpcError> {
2318 list_mirror::<IssueStateBy, IssueState<DefaultStr>, _>(&state, q).await
2319}
2320async fn count_issue_states_by(
2321 State(state): State<AppState>,
2322 XrpcQuery(q): XrpcQuery<CountQuery>,
2323) -> Result<Json<CountResponse>, XrpcError> {
2324 count_mirror::<IssueStateBy>(&state, q).map(Json)
2325}
2326
2327async fn list_pulls_by(
2328 State(state): State<AppState>,
2329 XrpcQuery(q): XrpcQuery<TypedListQuery<PullFilter>>,
2330) -> Result<Response, XrpcError> {
2331 let (page, nsid) = mirror_edge_page::<PullBy, _>(&state, &q)?;
2332 let permit = state.heavy_permit()?;
2333 let owned = state.clone();
2334 let items =
2335 hydrate_record_stream::<Pull<DefaultStr>>(&state, nsid, page.items, HitProvenance::Indexed)
2336 .map(move |view| view.map(|v| enrich_pull_view(&owned, v)));
2337 Ok(json_stream::<StatefulItem<Pull<DefaultStr>>, _>(
2338 "items",
2339 items,
2340 paged_tail(page.next.map(PageToken::encode_token)),
2341 permit,
2342 ))
2343}
2344async fn count_pulls_by(
2345 State(state): State<AppState>,
2346 XrpcQuery(q): XrpcQuery<CountQuery>,
2347) -> Result<Json<CountResponse>, XrpcError> {
2348 count_mirror::<PullBy>(&state, q).map(Json)
2349}
2350
2351async fn list_pull_statuses_by(
2352 State(state): State<AppState>,
2353 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2354) -> Result<Response, XrpcError> {
2355 list_mirror::<PullStatusBy, PullStatus<DefaultStr>, _>(&state, q).await
2356}
2357async fn count_pull_statuses_by(
2358 State(state): State<AppState>,
2359 XrpcQuery(q): XrpcQuery<CountQuery>,
2360) -> Result<Json<CountResponse>, XrpcError> {
2361 count_mirror::<PullStatusBy>(&state, q).map(Json)
2362}
2363
2364async fn list_spindle_members_by(
2365 State(state): State<AppState>,
2366 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2367) -> Result<Response, XrpcError> {
2368 list_mirror::<SpindleMemberBy, SpindleMember<DefaultStr>, _>(&state, q).await
2369}
2370async fn count_spindle_members_by(
2371 State(state): State<AppState>,
2372 XrpcQuery(q): XrpcQuery<CountQuery>,
2373) -> Result<Json<CountResponse>, XrpcError> {
2374 count_mirror::<SpindleMemberBy>(&state, q).map(Json)
2375}
2376
2377async fn list_label_definitions(
2378 State(state): State<AppState>,
2379 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2380) -> Result<Response, XrpcError> {
2381 list_records::<LabelDefinitionRecord, LabelDefinition<DefaultStr>, _>(&state, q).await
2382}
2383
2384async fn count_label_definitions(
2385 State(state): State<AppState>,
2386 XrpcQuery(q): XrpcQuery<CountQuery>,
2387) -> Result<Json<CountResponse>, XrpcError> {
2388 count_for::<LabelDefinitionRecord>(&state, q).map(Json)
2389}
2390
2391async fn list_label_ops(
2392 State(state): State<AppState>,
2393 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2394) -> Result<Response, XrpcError> {
2395 list_records::<LabelOpRecord, LabelOp<DefaultStr>, _>(&state, q).await
2396}
2397
2398async fn count_label_ops(
2399 State(state): State<AppState>,
2400 XrpcQuery(q): XrpcQuery<CountQuery>,
2401) -> Result<Json<CountResponse>, XrpcError> {
2402 count_for::<LabelOpRecord>(&state, q).map(Json)
2403}
2404
2405async fn list_pipelines(
2406 State(state): State<AppState>,
2407 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2408) -> Result<Response, XrpcError> {
2409 list_records::<PipelineRecord, Pipeline<DefaultStr>, _>(&state, q).await
2410}
2411
2412async fn count_pipelines(
2413 State(state): State<AppState>,
2414 XrpcQuery(q): XrpcQuery<CountQuery>,
2415) -> Result<Json<CountResponse>, XrpcError> {
2416 count_for::<PipelineRecord>(&state, q).map(Json)
2417}
2418
2419async fn list_pipeline_statuses(
2420 State(state): State<AppState>,
2421 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2422) -> Result<Response, XrpcError> {
2423 list_records::<PipelineStatusRecord, PipelineStatus<DefaultStr>, _>(&state, q).await
2424}
2425
2426async fn count_pipeline_statuses(
2427 State(state): State<AppState>,
2428 XrpcQuery(q): XrpcQuery<CountQuery>,
2429) -> Result<Json<CountResponse>, XrpcError> {
2430 count_for::<PipelineStatusRecord>(&state, q).map(Json)
2431}
2432
2433async fn list_artifacts(
2434 State(state): State<AppState>,
2435 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2436) -> Result<Response, XrpcError> {
2437 list_records::<ArtifactRecord, Artifact<DefaultStr>, _>(&state, q).await
2438}
2439
2440async fn count_artifacts(
2441 State(state): State<AppState>,
2442 XrpcQuery(q): XrpcQuery<CountQuery>,
2443) -> Result<Json<CountResponse>, XrpcError> {
2444 count_for::<ArtifactRecord>(&state, q).map(Json)
2445}
2446
2447async fn list_knot_members(
2448 State(state): State<AppState>,
2449 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2450) -> Result<Response, XrpcError> {
2451 list_records::<KnotMemberRecord, KnotMember<DefaultStr>, _>(&state, q).await
2452}
2453
2454async fn count_knot_members(
2455 State(state): State<AppState>,
2456 XrpcQuery(q): XrpcQuery<CountQuery>,
2457) -> Result<Json<CountResponse>, XrpcError> {
2458 count_for::<KnotMemberRecord>(&state, q).map(Json)
2459}
2460
2461async fn list_spindle_members(
2462 State(state): State<AppState>,
2463 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2464) -> Result<Response, XrpcError> {
2465 list_records::<SpindleMemberRecord, SpindleMember<DefaultStr>, _>(&state, q).await
2466}
2467
2468async fn count_spindle_members(
2469 State(state): State<AppState>,
2470 XrpcQuery(q): XrpcQuery<CountQuery>,
2471) -> Result<Json<CountResponse>, XrpcError> {
2472 count_for::<SpindleMemberRecord>(&state, q).map(Json)
2473}
2474
2475async fn list_strings(
2476 State(state): State<AppState>,
2477 XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
2478) -> Result<Response, XrpcError> {
2479 list_records::<TangledStringRecord, TangledString<DefaultStr>, _>(&state, q).await
2480}
2481
2482async fn count_strings(
2483 State(state): State<AppState>,
2484 XrpcQuery(q): XrpcQuery<CountQuery>,
2485) -> Result<Json<CountResponse>, XrpcError> {
2486 count_for::<TangledStringRecord>(&state, q).map(Json)
2487}
2488
2489#[derive(Deserialize)]
2490struct ResolveMiniDocParams {
2491 identifier: AtIdentifier<DefaultStr>,
2492}
2493
2494async fn resolve_mini_doc(
2495 State(state): State<AppState>,
2496 XrpcQuery(q): XrpcQuery<ResolveMiniDocParams>,
2497) -> Result<Response, XrpcError> {
2498 let body = state
2499 .slingshot
2500 .resolve_mini_doc(&q.identifier)
2501 .await
2502 .map_err(map_slingshot)?;
2503 Ok((StatusCode::OK, [(CONTENT_TYPE, "application/json")], body).into_response())
2504}
2505
2506async fn get_coverage(State(state): State<AppState>) -> Json<CoverageEnvelope> {
2507 Json(state.coverage.snapshot().into())
2508}
2509
2510async fn search_query(
2511 State(state): State<AppState>,
2512 XrpcQuery(q): XrpcQuery<SearchQueryParams>,
2513) -> Result<Response, XrpcError> {
2514 if q.q.trim().is_empty() {
2515 return Err(XrpcError::InvalidParams("q must not be empty".into()));
2516 }
2517 let cursor = SearchCursor::from_token(q.cursor.as_deref())
2518 .map_err(|e| XrpcError::InvalidParams(format!("cursor: {e}")))?;
2519 let limit = parse_limit(q.limit)?;
2520 let filters = build_search_filters(&q)?;
2521 let permit = state.heavy_permit()?;
2522 let page = state
2523 .search
2524 .search(&q.q, filters, cursor, limit.get())
2525 .await
2526 .map_err(map_search_err)?;
2527 let next = page.next.map(SearchOffset::encode_token);
2528 let owned = state.clone();
2529 let hits = hydrate_stream(page.hits, move |hit| {
2530 let owned = owned.clone();
2531 let uri = hit.uri.clone();
2532 let nsid = hit.nsid.clone();
2533 async move {
2534 let result = hydrate_search_hit(&owned, hit).await;
2535 drop_unhydratable(HitProvenance::Indexed, &nsid, &uri, result)
2536 }
2537 });
2538 Ok(json_stream::<SearchHitView, _>(
2539 "hits",
2540 hits,
2541 paged_tail(next),
2542 permit,
2543 ))
2544}
2545
2546fn build_search_filters(q: &SearchQueryParams) -> Result<SearchFilters, XrpcError> {
2547 let since = q
2548 .since
2549 .as_deref()
2550 .map(parse_rfc3339_seconds)
2551 .transpose()
2552 .map_err(|e| XrpcError::InvalidParams(format!("since: {e}")))?;
2553 let until = q
2554 .until
2555 .as_deref()
2556 .map(parse_rfc3339_seconds)
2557 .transpose()
2558 .map_err(|e| XrpcError::InvalidParams(format!("until: {e}")))?;
2559 if let (Some(s), Some(u)) = (since, until)
2560 && s > u
2561 {
2562 return Err(XrpcError::InvalidParams("since must be <= until".into()));
2563 }
2564 Ok(SearchFilters {
2565 nsid: q.nsid.clone(),
2566 author: q.author.clone(),
2567 repo: q.repo.clone(),
2568 since,
2569 until,
2570 })
2571}
2572
2573fn parse_rfc3339_seconds(raw: &str) -> Result<i64, String> {
2574 chrono::DateTime::parse_from_rfc3339(raw)
2575 .map(|dt| dt.timestamp())
2576 .map_err(|e| format!("expected RFC3339, got {raw}: {e}"))
2577}
2578
2579async fn hydrate_search_hit(
2580 state: &AppState,
2581 hit: SearchHit,
2582) -> Result<Option<SearchHitView>, XrpcError> {
2583 let SearchHit { uri, nsid, score } = hit;
2584 let body = resolve_for_view(state, &nsid, uri).await?;
2585 let record = decode_canon_or_upgrade(&nsid, &body.value, &state.resolver)
2586 .await
2587 .map_err(|err| XrpcError::InvalidRecord(err.to_string()))?;
2588 let Some(value) = SearchableRecord::try_from_record(record) else {
2589 return Ok(None);
2590 };
2591 let Some(value) = value.normalize(&state.resolver).await else {
2592 return Ok(None);
2593 };
2594 Ok(Some(SearchHitView {
2595 uri: body.uri.clone(),
2596 cid: Some(body.cid.clone()),
2597 nsid,
2598 score,
2599 value,
2600 }))
2601}
2602
2603fn map_search_err(err: SearchError) -> XrpcError {
2604 use SearchError as E;
2605 match err {
2606 E::Query(e) => XrpcError::InvalidParams(format!("query: {e}")),
2607 e @ (E::Tantivy(_)
2608 | E::InvalidUri(_)
2609 | E::InvalidNsid(_)
2610 | E::MissingField(_)
2611 | E::Cancelled(_)) => XrpcError::Internal(format!("search: {e}")),
2612 }
2613}
2614
2615fn map_proxy_error(err: KnotProxyError) -> XrpcError {
2616 match err {
2617 KnotProxyError::CircuitOpen => {
2618 XrpcError::UpstreamUnavailable("knot circuit breaker open".into())
2619 }
2620 KnotProxyError::BlockedHost { host, reason } => {
2621 XrpcError::InvalidRecord(format!("knot host {host} is {reason} address space"))
2622 }
2623 KnotProxyError::PlaintextHttp { host } => {
2624 XrpcError::InvalidRecord(format!("knot host {host} requires https"))
2625 }
2626 KnotProxyError::Connect(e) => XrpcError::UpstreamUnavailable(format!("connect: {e}")),
2627 KnotProxyError::Timeout(e) => {
2628 XrpcError::UpstreamUnavailable(format!("upstream timeout: {e}"))
2629 }
2630 KnotProxyError::Redirect(e) => XrpcError::UpstreamUnavailable(format!("redirect: {e}")),
2631 KnotProxyError::Transport(e) => XrpcError::UpstreamUnavailable(format!("transport: {e}")),
2632 KnotProxyError::Upstream(s) => XrpcError::UpstreamUnavailable(format!("status {s}")),
2633 }
2634}
2635
2636fn validate_client_supplied_knot(state: &AppState, host: &KnotHost) -> Result<(), XrpcError> {
2637 let host_str = || host.url().host_str().unwrap_or_default().to_owned();
2638 if state.knots.requires_https() && host.url().scheme() != "https" {
2639 return Err(XrpcError::InvalidParams(format!(
2640 "knot host {} must be https",
2641 host_str(),
2642 )));
2643 }
2644 if state.knots.allows_private_hosts() {
2645 return Ok(());
2646 }
2647 match host.private_literal_reason() {
2648 None => Ok(()),
2649 Some(reason) => Err(XrpcError::InvalidParams(format!(
2650 "knot host {} blocked: {} address space",
2651 host_str(),
2652 reason,
2653 ))),
2654 }
2655}
2656
2657async fn resolve_knot_target(
2658 state: &AppState,
2659 repo_uri: AtUri<DefaultStr>,
2660) -> Result<(KnotHost, RepoSlug), 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((host, slug))
2673}
2674
2675fn pick_human_slug(rkey: Option<&Rkey<DefaultStr>>, name: Option<&str>) -> Option<String> {
2676 match rkey {
2677 Some(r) if jacquard_common::types::tid::Tid::new(r.as_ref()).is_ok() => {
2678 Some(name.unwrap_or(r.as_ref()).to_owned())
2679 }
2680 Some(r) => Some(r.as_ref().to_owned()),
2681 None => name.map(str::to_owned),
2682 }
2683}
2684
2685fn filter_request_headers(client: &HeaderMap) -> HeaderMap {
2686 FORWARDED_REQUEST_HEADERS
2687 .iter()
2688 .fold(HeaderMap::new(), |mut acc, name| {
2689 if let Some(value) = client.get(*name) {
2690 acc.insert((*name).clone(), value.clone());
2691 }
2692 acc
2693 })
2694}
2695
2696fn upstream_to_axum(resp: ProxyResponse) -> Response {
2697 let status = resp.status();
2698 let upstream_headers = resp.headers().clone();
2699 let body = Body::from_stream(resp.into_body_stream());
2700 let mut response = Response::builder()
2701 .status(status)
2702 .body(body)
2703 .expect("response body construction must succeed");
2704 let response_headers = response.headers_mut();
2705 PASSTHROUGH_HEADERS.iter().for_each(|name| {
2706 if let Some(value) = upstream_headers.get(*name) {
2707 response_headers.insert((*name).clone(), value.clone());
2708 }
2709 });
2710 response
2711}
2712
2713async fn dispatch_proxy(
2714 state: AppState,
2715 headers: HeaderMap,
2716 nsid: Nsid<DefaultStr>,
2717 host: KnotHost,
2718 params: ProxyParams,
2719) -> Result<Response, XrpcError> {
2720 let forward: Vec<(&str, &str)> = params
2721 .iter()
2722 .map(|(k, v)| (k.as_str(), v.as_str()))
2723 .collect();
2724 let allowed = filter_request_headers(&headers);
2725 let upstream = state
2726 .knots
2727 .forward(&host, &nsid, &forward, allowed)
2728 .await
2729 .map_err(map_proxy_error)?;
2730 Ok(upstream_to_axum(upstream))
2731}
2732
2733fn extract_param(
2734 params: ProxyParams,
2735 key: &str,
2736) -> Result<Option<(String, ProxyParams)>, XrpcError> {
2737 let (matching, rest): (ProxyParams, ProxyParams) =
2738 params.into_iter().partition(|(k, _)| k == key);
2739 match matching.as_slice() {
2740 [] => Ok(None),
2741 [_] => Ok(matching.into_iter().next().map(|(_, v)| (v, rest))),
2742 _ => Err(XrpcError::InvalidParams(format!(
2743 "{key} parameter must appear at most once, got {}",
2744 matching.len(),
2745 ))),
2746 }
2747}
2748
2749async fn proxy_repo_handler(
2750 state: AppState,
2751 headers: HeaderMap,
2752 params: ProxyParams,
2753 nsid: Nsid<DefaultStr>,
2754) -> Result<Response, XrpcError> {
2755 let (repo_raw, rest) = extract_param(params, REPO_PARAM)?
2756 .ok_or_else(|| XrpcError::InvalidParams("missing repo".into()))?;
2757 let repo_uri = parse_uri(&repo_raw)?;
2758 let (host, slug) = resolve_knot_target(&state, repo_uri).await?;
2759 let forward = rest
2760 .into_iter()
2761 .chain(std::iter::once((
2762 REPO_PARAM.to_owned(),
2763 slug.as_str().to_owned(),
2764 )))
2765 .collect();
2766 dispatch_proxy(state, headers, nsid, host, forward).await
2767}
2768
2769async fn proxy_knot_handler(
2770 state: AppState,
2771 headers: HeaderMap,
2772 params: ProxyParams,
2773 nsid: Nsid<DefaultStr>,
2774) -> Result<Response, XrpcError> {
2775 let (knot_raw, forward) = extract_param(params, KNOT_HOST_PARAM)?
2776 .ok_or_else(|| XrpcError::InvalidParams("missing knot".into()))?;
2777 let host =
2778 KnotHost::parse(&knot_raw).map_err(|e| XrpcError::InvalidParams(format!("knot: {e}")))?;
2779 validate_client_supplied_knot(&state, &host)?;
2780 dispatch_proxy(state, headers, nsid, host, forward).await
2781}