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