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