alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::sync::Arc;
2
3use axum::Json;
4use axum::extract::{FromRequestParts, Query, State};
5use axum::response::{IntoResponse, Response};
6use http::StatusCode;
7use http::request::Parts;
8use serde::{Deserialize, Serialize};
9
10use knot_cobs::Grant;
11use knot_index::Resolved;
12use knot_runtime::{Clock, HttpTransport};
13use knot_types::{AccountDid, RepoDid};
14
15use crate::XrpcState;
16use crate::error::XrpcError;
17use crate::query::{Limit, Offset, Order, Total, ValidatedQuery, next_cursor};
18use crate::wire::rfc3339;
19
20pub(crate) const LIST_MEMBERS_ROUTE: &str = "/xrpc/sh.tangled.knot.listMembers";
21pub(crate) const LIST_COLLABORATORS_ROUTE: &str = "/xrpc/sh.tangled.repo.listCollaborators";
22
23const DEFAULT_LIMIT: usize = 50;
24const MAX_LIMIT: usize = 1000;
25
26#[derive(Deserialize)]
27pub(crate) struct Paging {
28 #[serde(default)]
29 limit: Limit<DEFAULT_LIMIT, MAX_LIMIT>,
30 #[serde(default)]
31 cursor: Offset,
32 #[serde(default)]
33 order: Order,
34}
35
36struct Window {
37 offset: Offset,
38 limit: Limit<DEFAULT_LIMIT, MAX_LIMIT>,
39 descending: bool,
40}
41
42impl Paging {
43 fn window(self) -> Window {
44 Window {
45 offset: self.cursor,
46 limit: self.limit,
47 descending: self.order.descending(),
48 }
49 }
50}
51
52#[derive(Deserialize)]
53struct SubjectQuery {
54 subject: Option<String>,
55}
56
57fn subject_param(parts: &Parts) -> Result<String, XrpcError> {
58 Query::<SubjectQuery>::try_from_uri(&parts.uri)
59 .map_err(|rejection| XrpcError::invalid_request(rejection.body_text()))?
60 .0
61 .subject
62 .filter(|raw| !raw.is_empty())
63 .ok_or_else(|| XrpcError::invalid_request("missing subject parameter"))
64}
65
66pub(crate) struct MemberSubject;
67
68impl<S: Send + Sync> FromRequestParts<S> for MemberSubject {
69 type Rejection = XrpcError;
70
71 async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
72 let subject = subject_param(parts)?;
73 AccountDid::new(subject)
74 .map(|_| MemberSubject)
75 .map_err(|_| {
76 XrpcError::named(
77 StatusCode::BAD_REQUEST,
78 "InvalidSubject",
79 "subject must be an account DID",
80 )
81 })
82 }
83}
84
85pub(crate) struct CollaboratorRepo(RepoDid);
86
87impl<S: Send + Sync> FromRequestParts<S> for CollaboratorRepo {
88 type Rejection = XrpcError;
89
90 async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
91 let subject = subject_param(parts)?;
92 RepoDid::new(subject).map(CollaboratorRepo).map_err(|_| {
93 XrpcError::named(
94 StatusCode::BAD_REQUEST,
95 "InvalidRepo",
96 "subject must be a repo DID",
97 )
98 })
99 }
100}
101
102#[derive(Serialize)]
103struct ItemWire {
104 subject: AccountDid,
105 #[serde(rename = "addedBy")]
106 added_by: AccountDid,
107 #[serde(rename = "createdAt")]
108 created_at: String,
109}
110
111#[derive(Serialize)]
112struct PageWire {
113 items: Vec<ItemWire>,
114 #[serde(skip_serializing_if = "Option::is_none")]
115 cursor: Option<String>,
116}
117
118fn respond(mut entries: Vec<Grant>, window: Window) -> Response {
119 entries.sort_by(|a, b| {
120 let by_time = a.created_at.cmp(&b.created_at);
121 let by_time = match window.descending {
122 true => by_time.reverse(),
123 false => by_time,
124 };
125 by_time.then_with(|| a.subject.cmp(&b.subject))
126 });
127 let total = entries.len();
128 let items = entries
129 .into_iter()
130 .skip(window.offset.get())
131 .take(window.limit.get())
132 .map(|grant| ItemWire {
133 subject: grant.subject,
134 added_by: grant.added_by,
135 created_at: rfc3339(grant.created_at.get(), 0),
136 })
137 .collect();
138 let cursor = next_cursor(window.offset, window.limit, Total::new(total));
139 Json(PageWire { items, cursor }).into_response()
140}
141
142fn members_warming() -> XrpcError {
143 XrpcError::warming("members projection is still warming")
144}
145
146fn collaborators_warming() -> XrpcError {
147 XrpcError::warming("collaborators projection is still warming")
148}
149
150pub(crate) async fn list_members<H: HttpTransport, C: Clock>(
151 State(state): State<Arc<XrpcState<H, C>>>,
152 _subject: MemberSubject,
153 ValidatedQuery(paging): ValidatedQuery<Paging>,
154) -> Result<Response, XrpcError> {
155 let window = paging.window();
156 match state.index.member_entries() {
157 Resolved::Warming => Err(members_warming()),
158 Resolved::Ready(entries) => Ok(respond(entries, window)),
159 }
160}
161
162pub(crate) async fn list_collaborators<H: HttpTransport, C: Clock>(
163 State(state): State<Arc<XrpcState<H, C>>>,
164 CollaboratorRepo(repo): CollaboratorRepo,
165 ValidatedQuery(paging): ValidatedQuery<Paging>,
166) -> Result<Response, XrpcError> {
167 let window = paging.window();
168 match state.index.owner_of(&repo) {
169 Resolved::Warming => return Err(crate::reads::warming()),
170 Resolved::Ready(None) => return Ok(respond(Vec::new(), window)),
171 Resolved::Ready(Some(_)) => {}
172 }
173 let index = Arc::clone(&state.index);
174 let target = repo.clone();
175 let entries = crate::run_blocking(move || {
176 index
177 .ensure_collaborators(&target)
178 .map_err(|_| collaborators_warming())?;
179 match index.collaborator_entries(&target) {
180 Resolved::Warming => Err(collaborators_warming()),
181 Resolved::Ready(entries) => Ok(entries),
182 }
183 })
184 .await?;
185 Ok(respond(entries, window))
186}