1use std::sync::Arc;
18
19use serde::Deserialize;
20use simploxide_api_types::{
21 client_api::ClientApi,
22 commands::ReceiveFile,
23 events::{Event, EventKind, RcvFileComplete, RcvFileError, RcvFileSndCancelled},
24 responses::{CancelFileResponse, RcvFileAcceptedSndCancelledResponse, ReceiveFileResponse},
25};
26
27use crate::{Hook, id::FileId};
28
29type FxDashMap<K, V> = dashmap::DashMap<K, V, rustc_hash::FxBuildHasher>;
30type XftpDownloadResponder = tokio::sync::oneshot::Sender<XftpManagerDownloadResponse>;
31
32pub trait XftpExt: ClientApi {
35 fn download_file<FID: Into<FileId>>(&self, file_id: FID) -> DownloadFileBuilder<'_, Self>;
45}
46
47#[derive(Clone)]
51pub struct XftpClient<C> {
52 client: C,
53 xftp: Arc<XftpManager>,
54}
55
56impl<C> XftpClient<C> {
57 pub(crate) fn manager(&self) -> Arc<XftpManager> {
58 self.xftp.clone()
59 }
60}
61
62#[cfg(feature = "websocket")]
63impl XftpClient<crate::ws::Client> {
64 pub fn version(
65 &self,
66 ) -> impl Future<Output = Result<crate::ws::SimplexVersion, crate::ws::VersionError>> {
67 self.client.version()
68 }
69
70 pub fn disconnect(self) -> impl Future<Output = ()> {
71 self.client.disconnect()
72 }
73}
74
75#[cfg(feature = "ffi")]
76impl XftpClient<crate::ffi::Client> {
77 pub fn version(
78 &self,
79 ) -> impl Future<Output = Result<crate::ffi::SimplexVersion, crate::ffi::VersionError>> {
80 self.client.version()
81 }
82
83 pub fn disconnect(self) -> impl Future<Output = ()> {
84 self.client.disconnect()
85 }
86}
87
88#[cfg(feature = "farm")]
89impl<C> XftpClient<C> {
90 pub(crate) fn new(client: C, xftp: Arc<XftpManager>) -> Self {
91 Self { client, xftp }
92 }
93}
94
95impl<C: ClientApi> ClientApi for XftpClient<C> {
96 type ResponseShape<'de, T: 'de + Deserialize<'de>> = C::ResponseShape<'de, T>;
97 type Error = C::Error;
98
99 fn send_raw(
100 &self,
101 command: String,
102 ) -> impl Future<Output = Result<String, Self::Error>> + Send {
103 self.client.send_raw(command)
104 }
105
106 fn cancel_file(
107 &self,
108 file_id: i64,
109 ) -> impl Future<Output = Result<CancelFileResponse, Self::Error>> + Send {
110 self.xftp.downloads.remove(&file_id);
111 self.client.cancel_file(file_id)
112 }
113}
114
115impl<C: ClientApi> From<C> for XftpClient<C> {
116 fn from(client: C) -> Self {
117 Self {
118 client,
119 xftp: Arc::new(XftpManager::default()),
120 }
121 }
122}
123
124impl<C: ClientApi> XftpExt for XftpClient<C> {
125 fn download_file<FID: Into<FileId>>(&self, file_id: FID) -> DownloadFileBuilder<'_, Self> {
126 DownloadFileBuilder {
127 client: self,
128 cmd: ReceiveFile::new(file_id.into().raw()),
129 }
130 }
131}
132
133impl Hook for XftpManager {
134 fn should_intercept(&self, kind: EventKind) -> bool {
135 const EVENT_KINDS: [EventKind; 3] = [
136 EventKind::RcvFileSndCancelled,
137 EventKind::RcvFileComplete,
138 EventKind::RcvFileError,
139 ];
140
141 EVENT_KINDS.contains(&kind)
142 }
143
144 fn intercept_event(&self, event: Event) {
145 match event {
146 Event::RcvFileComplete(ev) => {
147 if let Some((_, responder)) = self
148 .downloads
149 .remove(&ev.chat_item.chat_item.file.as_ref().unwrap().file_id)
150 {
151 let _ = responder.send(XftpManagerDownloadResponse::Complete(ev));
152 }
153 }
154 Event::RcvFileSndCancelled(ev) => {
155 if let Some((_, responder)) = self.downloads.remove(&ev.rcv_file_transfer.file_id) {
156 let _ = responder.send(XftpManagerDownloadResponse::Cancelled(ev));
157 }
158 }
159 Event::RcvFileError(ev) => {
160 if let Some((_, responder)) = self.downloads.remove(&ev.rcv_file_transfer.file_id) {
161 let _ = responder.send(XftpManagerDownloadResponse::Error(ev));
162 }
163 }
164 _ => (),
165 }
166 }
167}
168
169pub struct DownloadFileBuilder<'a, C: 'a + ?Sized> {
170 client: &'a C,
171 cmd: ReceiveFile,
172}
173
174impl<'a, C: 'a + ?Sized> DownloadFileBuilder<'a, C> {
175 pub fn via_user_approved_relays(mut self) -> Self {
177 self.cmd.user_approved_relays = true;
178 self
179 }
180
181 pub fn store_encrypted(mut self) -> Self {
183 self.cmd.store_encrypted = Some(true);
184 self
185 }
186
187 pub fn inline(mut self) -> Self {
189 self.cmd.file_inline = Some(true);
190 self
191 }
192
193 pub fn file_path<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
195 self.cmd.file_path = Some(path.as_ref().display().to_string());
196 self
197 }
198}
199
200impl<'a, C: 'a + ClientApi> IntoFuture for DownloadFileBuilder<'a, XftpClient<C>>
201where
202 <XftpClient<C> as ClientApi>::Error: 'static + Send,
203{
204 type Output = Result<Arc<RcvFileComplete>, DownloadError<<XftpClient<C> as ClientApi>::Error>>;
205 type IntoFuture = std::pin::Pin<Box<dyn 'a + Send + Future<Output = Self::Output>>>;
206
207 fn into_future(self) -> Self::IntoFuture {
208 Box::pin(async move {
209 let file_id = self.cmd.file_id;
210
211 let (responder, response) = tokio::sync::oneshot::channel();
212 self.client.xftp.downloads.insert(file_id, responder);
213
214 match self.client.receive_file(self.cmd).await {
215 Ok(ReceiveFileResponse::RcvFileAccepted(_)) => {
216 match response.await.expect("XFTP responses are always delivered") {
217 XftpManagerDownloadResponse::Complete(success) => Ok(success),
218 XftpManagerDownloadResponse::Cancelled(err) => {
219 Err(DownloadError::SendCancelled(err))
220 }
221 XftpManagerDownloadResponse::Error(err) => Err(DownloadError::Receive(err)),
222 }
223 }
224 Ok(ReceiveFileResponse::RcvFileAcceptedSndCancelled(err)) => {
225 self.client.xftp.downloads.remove(&file_id);
226 Err(DownloadError::AcceptCancelled(err))
227 }
228 Err(e) => {
229 self.client.xftp.downloads.remove(&file_id);
230 Err(DownloadError::Api(e))
231 }
232 }
233 })
234 }
235}
236
237pub enum DownloadError<E> {
239 SendCancelled(Arc<RcvFileSndCancelled>),
241 AcceptCancelled(Arc<RcvFileAcceptedSndCancelledResponse>),
243 Receive(Arc<RcvFileError>),
245 Api(E),
247}
248
249impl<E> std::fmt::Debug for DownloadError<E>
250where
251 E: std::fmt::Debug,
252{
253 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
254 match self {
255 Self::SendCancelled(arg) => f
256 .debug_tuple("SendCancelled")
257 .field(&arg.rcv_file_transfer.file_id)
258 .finish(),
259 Self::AcceptCancelled(arg) => f
260 .debug_tuple("AcceptCancelled")
261 .field(&arg.rcv_file_transfer.file_id)
262 .finish(),
263 Self::Receive(arg) => f.debug_tuple("Receive").field(&arg.agent_error).finish(),
264 Self::Api(e) => f.debug_tuple("Api").field(e).finish(),
265 }
266 }
267}
268
269impl<E> std::fmt::Display for DownloadError<E>
270where
271 E: std::fmt::Display,
272{
273 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
274 match self {
275 Self::SendCancelled(err) => write!(
276 f,
277 "File(ID={}) was cancelled by the user",
278 err.rcv_file_transfer.file_id
279 ),
280 Self::AcceptCancelled(err) => write!(
281 f,
282 "File(ID={}) is no longer available",
283 err.rcv_file_transfer.file_id
284 ),
285 Self::Receive(err) => write!(
286 f,
287 "File(ID={}) receive error: {:?}",
288 err.rcv_file_transfer.file_id, err.agent_error
289 ),
290 Self::Api(err) => write!(f, "{err}"),
291 }
292 }
293}
294
295impl<E> std::error::Error for DownloadError<E>
296where
297 E: 'static + std::error::Error,
298{
299 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
300 match self {
301 Self::SendCancelled(_) => None,
302 Self::AcceptCancelled(_) => None,
303 Self::Receive(_) => None,
304 Self::Api(error) => Some(error),
305 }
306 }
307}
308
309#[derive(Default)]
310pub(crate) struct XftpManager {
311 downloads: FxDashMap<i64, XftpDownloadResponder>,
312}
313
314enum XftpManagerDownloadResponse {
315 Complete(Arc<RcvFileComplete>),
316 Error(Arc<RcvFileError>),
317 Cancelled(Arc<RcvFileSndCancelled>),
318}