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