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