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