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