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