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