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