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