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