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