1use connexa::prelude::MultiaddrExt;
2use ipld_core::cid::Cid;
3use serde::{Deserialize, Serialize};
4use std::collections::{HashMap, HashSet};
5
6use crate::{Error, Ipfs};
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
9enum Origin {
10 #[allow(dead_code)]
11 All,
12 #[default]
13 Public,
14}
15
16#[derive(Clone)]
17pub struct RemotePinningService {
18 ipfs: Ipfs,
19 client: reqwest::Client,
20 endpoint: String,
21 token: String,
22 origin: Origin,
23}
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
26#[serde(rename_all = "lowercase")]
27pub enum Status {
28 Queued,
29 Pinning,
30 Pinned,
31 Failed,
32}
33
34#[derive(Debug, Clone, Serialize, Deserialize)]
35pub struct PinObject {
36 pub cid: String,
37 #[serde(default, skip_serializing_if = "Option::is_none")]
38 pub name: Option<String>,
39 #[serde(default, skip_serializing_if = "Vec::is_empty")]
40 pub origins: Vec<String>,
41 #[serde(default, skip_serializing_if = "HashMap::is_empty")]
42 pub meta: HashMap<String, String>,
43}
44
45#[derive(Debug, Clone, Deserialize)]
46pub struct PinStatus {
47 pub requestid: String,
48 pub status: Status,
49 pub created: String,
50 pub pin: PinObject,
51 #[serde(default)]
52 pub delegates: Vec<String>,
53 #[serde(default)]
54 pub info: HashMap<String, String>,
55}
56
57#[derive(Debug, Clone, Deserialize)]
58pub struct PinResults {
59 pub count: u64,
60 pub results: Vec<PinStatus>,
61}
62
63#[derive(Debug, Clone, Default)]
64pub struct AddOptions {
65 pub name: Option<String>,
66 pub origins: Vec<String>,
67 pub meta: HashMap<String, String>,
68}
69
70#[derive(Debug, Clone, Default)]
71pub struct ListQuery {
72 pub cids: Vec<Cid>,
73 pub name: Option<String>,
74 pub status: Vec<Status>,
75 pub limit: Option<u32>,
76}
77
78impl RemotePinningService {
79 pub(crate) fn new(ipfs: Ipfs, endpoint: impl Into<String>, token: impl Into<String>) -> Self {
80 let endpoint = endpoint.into().trim_end_matches('/').to_string();
81 Self {
82 ipfs,
83 client: reqwest::Client::new(),
84 endpoint,
85 token: token.into(),
86 origin: Origin::default(),
87 }
88 }
89
90 pub async fn add(&self, cid: Cid, options: AddOptions) -> Result<PinStatus, Error> {
91 let body = self.pin_object(cid, options).await;
92 let response = self
93 .client
94 .post(format!("{}/pins", self.endpoint))
95 .bearer_auth(&self.token)
96 .json(&body)
97 .send()
98 .await?;
99 parse(response).await
100 }
101
102 pub async fn status(&self, requestid: &str) -> Result<PinStatus, Error> {
103 let response = self
104 .client
105 .get(format!("{}/pins/{requestid}", self.endpoint))
106 .bearer_auth(&self.token)
107 .send()
108 .await?;
109 parse(response).await
110 }
111
112 pub async fn list(&self, query: ListQuery) -> Result<PinResults, Error> {
113 let mut request = self
114 .client
115 .get(format!("{}/pins", self.endpoint))
116 .bearer_auth(&self.token);
117
118 if !query.cids.is_empty() {
119 let cids = query
120 .cids
121 .iter()
122 .map(Cid::to_string)
123 .collect::<Vec<_>>()
124 .join(",");
125 request = request.query(&[("cid", cids)]);
126 }
127 if let Some(name) = query.name {
128 request = request.query(&[("name", name)]);
129 }
130 if !query.status.is_empty() {
131 let statuses = query
132 .status
133 .iter()
134 .map(status_str)
135 .collect::<Vec<_>>()
136 .join(",");
137 request = request.query(&[("status", statuses)]);
138 }
139 if let Some(limit) = query.limit {
140 request = request.query(&[("limit", limit.to_string())]);
141 }
142
143 let response = request.send().await?;
144 parse(response).await
145 }
146
147 pub async fn replace(
148 &self,
149 requestid: &str,
150 cid: Cid,
151 options: AddOptions,
152 ) -> Result<PinStatus, Error> {
153 let body = self.pin_object(cid, options).await;
154 let response = self
155 .client
156 .post(format!("{}/pins/{requestid}", self.endpoint))
157 .bearer_auth(&self.token)
158 .json(&body)
159 .send()
160 .await?;
161 parse(response).await
162 }
163
164 pub async fn remove(&self, requestid: &str) -> Result<(), Error> {
165 let response = self
166 .client
167 .delete(format!("{}/pins/{requestid}", self.endpoint))
168 .bearer_auth(&self.token)
169 .send()
170 .await?;
171 success(response).await.map(|_| ())
172 }
173
174 async fn pin_object(&self, cid: Cid, options: AddOptions) -> PinObject {
175 let origins = if options.origins.is_empty() {
176 self.local_origins().await
177 } else {
178 options.origins
179 };
180 PinObject {
181 cid: cid.to_string(),
182 name: options.name,
183 origins,
184 meta: options.meta,
185 }
186 }
187
188 async fn local_origins(&self) -> Vec<String> {
189 let peer_id = self.ipfs.keypair().public().to_peer_id();
190 let addresses = match self.origin {
191 Origin::Public => self
192 .ipfs
193 .external_addresses()
194 .await
195 .unwrap_or_default()
196 .into_iter()
197 .filter(|addr| addr.is_public())
198 .map(|addr| addr.clone().with_p2p(peer_id).unwrap_or(addr))
199 .collect::<HashSet<_>>(),
200 Origin::All => {
201 let listening = self
202 .ipfs
203 .listening_addresses()
204 .await
205 .unwrap_or_default()
206 .into_iter()
207 .map(|addr| addr.clone().with_p2p(peer_id).unwrap_or(addr))
208 .collect::<HashSet<_>>();
209
210 let external = self
211 .ipfs
212 .external_addresses()
213 .await
214 .unwrap_or_default()
215 .into_iter()
216 .map(|addr| addr.clone().with_p2p(peer_id).unwrap_or(addr))
217 .collect::<HashSet<_>>();
218
219 listening
220 .into_iter()
221 .chain(external)
222 .collect::<HashSet<_>>()
223 }
224 };
225
226 Vec::from_iter(addresses.into_iter().map(|addr| addr.to_string()))
227 }
228}
229
230fn status_str(status: &Status) -> &'static str {
231 match status {
232 Status::Queued => "queued",
233 Status::Pinning => "pinning",
234 Status::Pinned => "pinned",
235 Status::Failed => "failed",
236 }
237}
238
239async fn parse<T>(response: reqwest::Response) -> Result<T, Error>
240where
241 T: for<'de> Deserialize<'de>,
242{
243 let response = success(response).await?;
244 response.json::<T>().await.map_err(Error::from)
245}
246
247async fn success(response: reqwest::Response) -> Result<reqwest::Response, Error> {
248 if response.status().is_success() {
249 return Ok(response);
250 }
251 let status = response.status();
252 match response.json::<Failure>().await {
253 Ok(failure) => Err(anyhow::anyhow!(
254 "pinning service error {status}: {}",
255 failure.error.reason
256 )),
257 Err(_) => Err(anyhow::anyhow!("pinning service error {status}")),
258 }
259}
260
261#[derive(Deserialize)]
262struct Failure {
263 error: FailureReason,
264}
265
266#[derive(Deserialize)]
267struct FailureReason {
268 reason: String,
269}
270
271#[cfg(test)]
272mod tests {
273 use super::*;
274
275 #[test]
276 fn deserialize_pin_status() {
277 let json = r#"{
278 "requestid": "UniqueIdOfPinRequest",
279 "status": "pinned",
280 "created": "2020-07-27T17:32:28.276Z",
281 "pin": { "cid": "QmCIDToBePinned", "name": "my-pin" },
282 "delegates": ["/dnsaddr/pin-service.example.com"]
283 }"#;
284 let status: PinStatus = serde_json::from_str(json).unwrap();
285 assert_eq!(status.requestid, "UniqueIdOfPinRequest");
286 assert_eq!(status.status, Status::Pinned);
287 assert_eq!(status.pin.cid, "QmCIDToBePinned");
288 assert_eq!(status.pin.name.as_deref(), Some("my-pin"));
289 assert_eq!(status.delegates, vec!["/dnsaddr/pin-service.example.com"]);
290 }
291
292 #[test]
293 fn serialize_pin_object_omits_empty() {
294 let pin = PinObject {
295 cid: "QmFoo".into(),
296 name: None,
297 origins: vec![],
298 meta: HashMap::new(),
299 };
300 assert_eq!(
301 serde_json::to_value(&pin).unwrap(),
302 serde_json::json!({ "cid": "QmFoo" })
303 );
304 }
305}