1use std::{collections::HashSet, marker::PhantomData, net::SocketAddr, str::FromStr, sync::Arc};
2
3use anyhow::Context;
4use buffers::ByteBufOwned;
5use dht::{DhtStats, Id20};
6use http::StatusCode;
7use librqbit_core::torrent_metainfo::{FileDetailsAttrs, ValidatedTorrentMetaV1Info};
8use serde::{Deserialize, Serialize};
9use tokio::sync::mpsc::UnboundedSender;
10
11use crate::{
12 WithStatus, WithStatusError,
13 api_error::ApiError,
14 session::{
15 AddTorrent, AddTorrentOptions, AddTorrentResponse, ListOnlyResponse, Session, TorrentId,
16 },
17 session_stats::snapshot::SessionStatsSnapshot,
18 torrent_state::{
19 FileStream, ManagedTorrentHandle,
20 peer::stats::snapshot::{PeerStatsFilter, PeerStatsSnapshot},
21 },
22 type_aliases::BF,
23};
24
25#[cfg(feature = "tracing-subscriber-utils")]
26use crate::tracing_subscriber_config_utils::LineBroadcast;
27#[cfg(feature = "tracing-subscriber-utils")]
28use futures::Stream;
29#[cfg(feature = "tracing-subscriber-utils")]
30use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError};
31
32pub use crate::torrent_state::stats::{LiveStats, TorrentStats};
33
34pub type Result<T> = std::result::Result<T, ApiError>;
35
36#[derive(Clone)]
39pub struct Api {
40 session: Arc<Session>,
41 rust_log_reload_tx: Option<UnboundedSender<String>>,
42 #[cfg(feature = "tracing-subscriber-utils")]
43 line_broadcast: Option<LineBroadcast>,
44}
45
46#[derive(Debug, Clone, Copy)]
47pub enum TorrentIdOrHash {
48 Id(TorrentId),
49 Hash(Id20),
50}
51
52impl Serialize for TorrentIdOrHash {
53 fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
54 where
55 S: serde::Serializer,
56 {
57 match self {
58 TorrentIdOrHash::Id(id) => id.serialize(serializer),
59 TorrentIdOrHash::Hash(h) => h.as_string().serialize(serializer),
60 }
61 }
62}
63
64impl<'de> Deserialize<'de> for TorrentIdOrHash {
65 fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
66 where
67 D: serde::Deserializer<'de>,
68 {
69 #[derive(Default)]
70 struct V<'de> {
71 p: PhantomData<&'de ()>,
72 }
73
74 macro_rules! visit_int {
75 ($v:expr) => {{
76 let tid: TorrentId = $v.try_into().map_err(|e| E::custom(format!("{e:#}")))?;
77 Ok(TorrentIdOrHash::from(tid))
78 }};
79 }
80
81 impl<'de> serde::de::Visitor<'de> for V<'de> {
82 type Value = TorrentIdOrHash;
83
84 fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
85 f.write_str("integer or 40 byte info hash")
86 }
87
88 fn visit_i64<E>(self, v: i64) -> std::result::Result<Self::Value, E>
89 where
90 E: serde::de::Error,
91 {
92 visit_int!(v)
93 }
94
95 fn visit_i128<E>(self, v: i128) -> std::result::Result<Self::Value, E>
96 where
97 E: serde::de::Error,
98 {
99 visit_int!(v)
100 }
101
102 fn visit_u128<E>(self, v: u128) -> std::result::Result<Self::Value, E>
103 where
104 E: serde::de::Error,
105 {
106 visit_int!(v)
107 }
108
109 fn visit_u64<E>(self, v: u64) -> std::result::Result<Self::Value, E>
110 where
111 E: serde::de::Error,
112 {
113 visit_int!(v)
114 }
115
116 fn visit_str<E>(self, v: &str) -> std::result::Result<Self::Value, E>
117 where
118 E: serde::de::Error,
119 {
120 TorrentIdOrHash::parse(v).map_err(|e| {
121 E::custom(format!(
122 "expected integer or 40 byte info hash, couldn't parse string: {e:#}"
123 ))
124 })
125 }
126 }
127
128 deserializer.deserialize_any(V::default())
129 }
130}
131
132impl std::fmt::Display for TorrentIdOrHash {
133 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
134 match self {
135 TorrentIdOrHash::Id(id) => write!(f, "{id}"),
136 TorrentIdOrHash::Hash(h) => write!(f, "{h:?}"),
137 }
138 }
139}
140
141impl From<TorrentId> for TorrentIdOrHash {
142 fn from(value: TorrentId) -> Self {
143 TorrentIdOrHash::Id(value)
144 }
145}
146
147impl From<Id20> for TorrentIdOrHash {
148 fn from(value: Id20) -> Self {
149 TorrentIdOrHash::Hash(value)
150 }
151}
152
153impl<'a> TryFrom<&'a str> for TorrentIdOrHash {
154 type Error = anyhow::Error;
155
156 fn try_from(value: &'a str) -> std::result::Result<Self, Self::Error> {
157 Self::parse(value)
158 }
159}
160
161impl TorrentIdOrHash {
162 pub fn parse(s: &str) -> anyhow::Result<Self> {
163 if s.len() == 40 {
164 let id = Id20::from_str(s)?;
165 return Ok(id.into());
166 }
167 let id: TorrentId = s.parse()?;
168 Ok(id.into())
169 }
170}
171
172#[derive(Deserialize, Default)]
173pub struct ApiTorrentListOpts {
174 #[serde(default)]
175 pub with_stats: bool,
176}
177
178impl Api {
179 pub fn new(
180 session: Arc<Session>,
181 rust_log_reload_tx: Option<UnboundedSender<String>>,
182 #[cfg(feature = "tracing-subscriber-utils")] line_broadcast: Option<LineBroadcast>,
183 ) -> Self {
184 Self {
185 session,
186 rust_log_reload_tx,
187 #[cfg(feature = "tracing-subscriber-utils")]
188 line_broadcast,
189 }
190 }
191
192 pub fn session(&self) -> &Arc<Session> {
193 &self.session
194 }
195
196 pub fn mgr_handle(&self, idx: TorrentIdOrHash) -> Result<ManagedTorrentHandle> {
197 self.session
198 .get(idx)
199 .ok_or(ApiError::torrent_not_found(idx))
200 }
201
202 pub fn api_torrent_list(&self) -> TorrentListResponse {
203 self.api_torrent_list_ext(ApiTorrentListOpts { with_stats: false })
204 }
205
206 pub fn api_torrent_list_ext(&self, opts: ApiTorrentListOpts) -> TorrentListResponse {
207 let items = self.session.with_torrents(|torrents| {
208 torrents
209 .map(|(id, mgr)| {
210 let total_pieces = mgr
211 .metadata
212 .load()
213 .as_ref()
214 .map(|m| m.info.lengths().total_pieces())
215 .unwrap_or(0);
216 let mut r = TorrentDetailsResponse {
217 id: Some(id),
218 info_hash: mgr.shared().info_hash.as_string(),
219 name: mgr.name(),
220 output_folder: mgr
221 .shared()
222 .options
223 .output_folder
224 .to_string_lossy()
225 .into_owned(),
226 total_pieces,
227
228 files: None,
230 stats: None,
231 };
232 if opts.with_stats {
233 r.stats = Some(mgr.stats());
234 }
235 r
236 })
237 .collect()
238 });
239 TorrentListResponse { torrents: items }
240 }
241
242 pub fn api_torrent_details(&self, idx: TorrentIdOrHash) -> Result<TorrentDetailsResponse> {
243 let handle = self.mgr_handle(idx)?;
244 let info_hash = handle.shared().info_hash;
245 let only_files = handle.only_files();
246 let output_folder = handle
247 .shared()
248 .options
249 .output_folder
250 .to_string_lossy()
251 .into_owned()
252 .to_string();
253 make_torrent_details(
254 Some(handle.id()),
255 &info_hash,
256 handle.metadata.load().as_ref().map(|r| &r.info),
257 handle.name().as_deref(),
258 only_files.as_deref(),
259 output_folder,
260 )
261 }
262
263 pub fn api_session_stats(&self) -> SessionStatsSnapshot {
264 self.session().stats_snapshot()
265 }
266
267 pub fn torrent_file_mime_type(
268 &self,
269 idx: TorrentIdOrHash,
270 file_idx: usize,
271 ) -> Result<&'static str> {
272 let handle = self.mgr_handle(idx)?;
273 handle.with_metadata(|r| torrent_file_mime_type(&r.info, file_idx))?
274 }
275
276 pub fn api_peer_stats(
277 &self,
278 idx: TorrentIdOrHash,
279 filter: PeerStatsFilter,
280 ) -> Result<PeerStatsSnapshot> {
281 let handle = self.mgr_handle(idx)?;
282 Ok(handle
283 .live()
284 .with_status_error(
285 StatusCode::PRECONDITION_FAILED,
286 crate::Error::TorrentIsNotLive,
287 )?
288 .per_peer_stats_snapshot(filter))
289 }
290
291 pub async fn api_torrent_action_pause(
292 &self,
293 idx: TorrentIdOrHash,
294 ) -> Result<EmptyJsonResponse> {
295 let handle = self.mgr_handle(idx)?;
296 self.session()
297 .pause(&handle)
298 .await
299 .with_status(StatusCode::BAD_REQUEST)?;
300 Ok(Default::default())
301 }
302
303 pub async fn api_torrent_action_start(
304 &self,
305 idx: TorrentIdOrHash,
306 ) -> Result<EmptyJsonResponse> {
307 let handle = self.mgr_handle(idx)?;
308 self.session
309 .unpause(&handle)
310 .await
311 .with_status(StatusCode::BAD_REQUEST)?;
312 Ok(Default::default())
313 }
314
315 pub async fn api_torrent_action_forget(
316 &self,
317 idx: TorrentIdOrHash,
318 ) -> Result<EmptyJsonResponse> {
319 self.session
320 .delete(idx, false)
321 .await
322 .context("error forgetting torrent")?;
323 Ok(Default::default())
324 }
325
326 pub async fn api_torrent_action_delete(
327 &self,
328 idx: TorrentIdOrHash,
329 ) -> Result<EmptyJsonResponse> {
330 self.session
331 .delete(idx, true)
332 .await
333 .context("error deleting torrent with files")?;
334 Ok(Default::default())
335 }
336
337 pub async fn api_torrent_action_update_only_files(
338 &self,
339 idx: TorrentIdOrHash,
340 only_files: &HashSet<usize>,
341 ) -> Result<EmptyJsonResponse> {
342 let handle = self.mgr_handle(idx)?;
343 self.session
344 .update_only_files(&handle, only_files)
345 .await
346 .context("error updating only_files")?;
347 Ok(Default::default())
348 }
349
350 pub fn api_set_rust_log(&self, new_value: String) -> Result<EmptyJsonResponse> {
351 let tx = self
352 .rust_log_reload_tx
353 .as_ref()
354 .context("rust_log_reload_tx was not set")?;
355 tx.send(new_value)
356 .context("noone is listening to RUST_LOG changes")?;
357 Ok(Default::default())
358 }
359
360 #[cfg(feature = "tracing-subscriber-utils")]
361 pub fn api_log_lines_stream(
362 &self,
363 ) -> Result<
364 impl Stream<Item = std::result::Result<bytes::Bytes, BroadcastStreamRecvError>>
365 + Send
366 + Sync
367 + 'static,
368 > {
369 Ok(self
370 .line_broadcast
371 .as_ref()
372 .map(|sender| BroadcastStream::new(sender.subscribe()))
373 .context("line_rx wasn't set")?)
374 }
375
376 pub async fn api_add_torrent(
377 &self,
378 add: AddTorrent<'_>,
379 opts: Option<AddTorrentOptions>,
380 ) -> Result<ApiAddTorrentResponse> {
381 let response = match self
382 .session
383 .add_torrent(add, opts)
384 .await
385 .context("error adding torrent")
386 .with_status(StatusCode::BAD_REQUEST)?
387 {
388 AddTorrentResponse::AlreadyManaged(id, handle) => {
389 let details = make_torrent_details(
390 Some(id),
391 &handle.info_hash(),
392 handle.metadata.load().as_ref().map(|r| &r.info),
393 handle.name().as_deref(),
394 handle.only_files().as_deref(),
395 handle
396 .shared()
397 .options
398 .output_folder
399 .to_string_lossy()
400 .into_owned(),
401 )
402 .context("error making torrent details")?;
403 ApiAddTorrentResponse {
404 id: Some(id),
405 details,
406 seen_peers: None,
407 output_folder: handle
408 .shared()
409 .options
410 .output_folder
411 .to_string_lossy()
412 .into_owned(),
413 }
414 }
415 AddTorrentResponse::ListOnly(ListOnlyResponse {
416 info_hash,
417 info,
418 only_files,
419 seen_peers,
420 output_folder,
421 ..
422 }) => ApiAddTorrentResponse {
423 id: None,
424 output_folder: output_folder.to_string_lossy().into_owned(),
425 seen_peers: Some(seen_peers),
426 details: make_torrent_details(
427 None,
428 &info_hash,
429 Some(&info),
430 None,
431 only_files.as_deref(),
432 output_folder.to_string_lossy().into_owned().to_string(),
433 )
434 .context("error making torrent details")?,
435 },
436 AddTorrentResponse::Added(id, handle) => {
437 let details = make_torrent_details(
438 Some(id),
439 &handle.info_hash(),
440 handle.metadata.load().as_ref().map(|r| &r.info),
441 handle.name().as_deref(),
442 handle.only_files().as_deref(),
443 handle
444 .shared()
445 .options
446 .output_folder
447 .to_string_lossy()
448 .into_owned(),
449 )
450 .context("error making torrent details")?;
451 ApiAddTorrentResponse {
452 id: Some(id),
453 details,
454 seen_peers: None,
455 output_folder: handle
456 .shared()
457 .options
458 .output_folder
459 .to_string_lossy()
460 .into_owned(),
461 }
462 }
463 };
464 Ok(response)
465 }
466
467 pub fn api_dht_stats(&self) -> Result<DhtStats> {
468 self.session
469 .get_dht()
470 .as_ref()
471 .map(|d| d.stats())
472 .ok_or(ApiError::dht_disabled())
473 }
474
475 pub fn api_dht_table(&self) -> Result<impl Serialize + use<>> {
476 let dht = self.session.get_dht().ok_or(ApiError::dht_disabled())?;
477 Ok(dht.with_routing_tables(|v4, v6| {
478 #[derive(Serialize)]
479 struct Tables<T> {
480 v4: T,
481 v6: T,
482 }
483 Tables {
484 v4: v4.clone(),
485 v6: v6.clone(),
486 }
487 }))
488 }
489
490 pub fn api_stats_v0(&self, idx: TorrentIdOrHash) -> Result<LiveStats> {
491 let mgr = self.mgr_handle(idx)?;
492 let live = mgr.live().context("torrent not live")?;
493 Ok(LiveStats::from(&*live))
494 }
495
496 pub fn api_stats_v1(&self, idx: TorrentIdOrHash) -> Result<TorrentStats> {
497 let mgr = self.mgr_handle(idx)?;
498 Ok(mgr.stats())
499 }
500
501 pub fn api_dump_haves(&self, idx: TorrentIdOrHash) -> Result<(BF, u32)> {
502 let mgr = self.mgr_handle(idx)?;
503 mgr.with_chunk_tracker(|chunks| {
504 let bf = BF::from_bitslice(chunks.get_have_pieces().as_slice());
505 let len = chunks.get_lengths().total_pieces();
506 (bf, len)
507 })
508 .with_status_error(
509 StatusCode::PRECONDITION_FAILED,
510 crate::Error::TorrentIsNotLive,
511 )
512 }
513
514 pub async fn api_stream(&self, idx: TorrentIdOrHash, file_id: usize) -> Result<FileStream> {
515 let mgr = self.mgr_handle(idx)?;
516 Ok(mgr.stream(file_id).await?)
517 }
518}
519
520#[derive(Serialize)]
521pub struct TorrentListResponse {
522 pub torrents: Vec<TorrentDetailsResponse>,
523}
524
525#[derive(Serialize, Deserialize)]
526pub struct TorrentDetailsResponseFile {
527 pub name: String,
528 pub components: Vec<String>,
529 pub length: u64,
530 pub included: bool,
531 pub attributes: FileDetailsAttrs,
532}
533
534#[derive(Default, Serialize)]
535pub struct EmptyJsonResponse {}
536
537#[derive(Serialize, Deserialize)]
538pub struct TorrentDetailsResponse {
539 #[serde(skip_serializing_if = "Option::is_none")]
540 pub id: Option<usize>,
541 pub info_hash: String,
542 pub name: Option<String>,
543 pub output_folder: String,
544
545 #[serde(default)]
546 pub total_pieces: u32,
547
548 #[serde(skip_serializing_if = "Option::is_none")]
549 pub files: Option<Vec<TorrentDetailsResponseFile>>,
550 #[serde(skip_serializing_if = "Option::is_none", skip_deserializing)]
551 pub stats: Option<TorrentStats>,
552}
553
554#[derive(Serialize, Deserialize)]
555pub struct ApiAddTorrentResponse {
556 pub id: Option<usize>,
557 pub details: TorrentDetailsResponse,
558 pub output_folder: String,
559 pub seen_peers: Option<Vec<SocketAddr>>,
560}
561
562fn make_torrent_details(
563 id: Option<TorrentId>,
564 info_hash: &Id20,
565 info: Option<&ValidatedTorrentMetaV1Info<ByteBufOwned>>,
566 name: Option<&str>,
567 only_files: Option<&[usize]>,
568 output_folder: String,
569) -> Result<TorrentDetailsResponse> {
570 let files = match info {
571 Some(info) => info
572 .iter_file_details()
573 .enumerate()
574 .map(|(idx, d)| {
575 let name = d.filename.to_string();
576 let components = d.filename.to_vec();
577 let included = only_files.map(|o| o.contains(&idx)).unwrap_or(true);
578 TorrentDetailsResponseFile {
579 name,
580 components,
581 length: d.len,
582 included,
583 attributes: d.attrs(),
584 }
585 })
586 .collect(),
587 None => Default::default(),
588 };
589 let total_pieces = info.map(|i| i.lengths().total_pieces()).unwrap_or(0);
590 Ok(TorrentDetailsResponse {
591 id,
592 info_hash: info_hash.as_string(),
593 name: name
594 .map(|s| s.to_owned())
595 .or_else(|| info.and_then(|i| i.name().map(|n| n.into_owned()))),
596 files: Some(files),
597 output_folder,
598 total_pieces,
599 stats: None,
600 })
601}
602
603fn torrent_file_mime_type(
604 info: &ValidatedTorrentMetaV1Info<ByteBufOwned>,
605 file_idx: usize,
606) -> Result<&'static str> {
607 Ok(info
608 .iter_file_details()
609 .nth(file_idx)
610 .and_then(|d| {
611 d.filename
612 .iter_components()
613 .last()
614 .and_then(|s| mime_guess::from_path(&*s).first_raw())
615 })
616 .ok_or((
617 StatusCode::INTERNAL_SERVER_ERROR,
618 "cannot determine mime type for file",
619 ))?)
620}