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