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