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