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