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