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