1use std::{
2 collections::HashMap,
3 io::{self, Read, SeekFrom},
4};
5
6use futures::TryStreamExt;
7use reqwest::{
8 header::{CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, RANGE},
9 multipart::Part,
10 Method, StatusCode,
11};
12use serde::Deserialize;
13use tracing::{debug, trace};
14use url::Url;
15
16use crate::{
17 configuration::Endpoint, prelude::AttachmentIdentifier,
18 proto::AttachmentPointer, push_service::HttpAuthOverride,
19};
20
21use super::{response::ReqwestExt, PushService, ServiceError};
22
23#[derive(Debug, serde::Deserialize, Default)]
24#[serde(rename_all = "camelCase")]
25pub struct AttachmentV2UploadAttributes {
26 key: String,
27 credential: String,
28 acl: String,
29 algorithm: String,
30 date: String,
31 policy: String,
32 signature: String,
33}
34
35#[derive(Debug, Deserialize)]
36#[serde(rename_all = "camelCase")]
37pub struct AttachmentUploadForm {
38 pub cdn: u32,
39 pub key: String,
40 pub headers: HashMap<String, String>,
41 pub signed_upload_location: Url,
42}
43
44#[derive(Debug, Deserialize)]
45#[serde(rename_all = "camelCase")]
46pub struct AttachmentDigest {
47 pub digest: Vec<u8>,
48 pub incremental_digest: Option<Vec<u8>>,
49 pub incremental_mac_chunk_size: u64,
50}
51
52#[derive(Debug)]
53pub struct ResumeInfo {
54 pub content_range: Option<String>,
55 pub content_start: u64,
56}
57
58pub struct AttachmentDownload<R> {
59 pub stream: R,
60 pub content_length: Option<u64>,
61}
62
63impl PushService {
64 pub async fn get_attachment(
65 &mut self,
66 ptr: &AttachmentPointer,
67 ) -> Result<
68 AttachmentDownload<impl futures::io::AsyncRead + Send + Unpin>,
69 ServiceError,
70 > {
71 let path = match ptr.attachment_identifier.as_ref() {
72 Some(AttachmentIdentifier::CdnId(id)) => {
73 format!("attachments/{}", id)
74 },
75 Some(AttachmentIdentifier::CdnKey(key)) => {
76 format!("attachments/{}", key)
77 },
78 None => {
79 return Err(ServiceError::InvalidFrame {
80 reason: "no attachment identifier in pointer",
81 });
82 },
83 };
84 self.get_from_cdn(ptr.cdn_number(), &path).await
85 }
86
87 #[tracing::instrument(skip(self))]
88 pub(crate) async fn get_from_cdn(
89 &mut self,
90 cdn_id: u32,
91 path: &str,
92 ) -> Result<
93 AttachmentDownload<impl futures::io::AsyncRead + Send + Unpin>,
94 ServiceError,
95 > {
96 let response = self
97 .request(
98 Method::GET,
99 Endpoint::cdn(cdn_id, path),
100 HttpAuthOverride::Unidentified, )?
102 .send()
103 .await?
104 .error_for_status()?;
105 let content_length = match response.headers().get(CONTENT_LENGTH) {
106 Some(value) => {
107 match value.to_str().ok().and_then(|value| value.parse().ok()) {
108 Some(value) => Some(value),
109 None => {
110 tracing::warn!("invalid Content-Length header");
111 None
112 },
113 }
114 },
115 None => None,
116 };
117 let response_stream = response
118 .bytes_stream()
119 .map_err(io::Error::other)
120 .into_async_read();
121
122 Ok(AttachmentDownload {
123 stream: response_stream,
124 content_length,
125 })
126 }
127
128 pub(crate) async fn get_attachment_v4_upload_attributes(
129 &mut self,
130 ) -> Result<AttachmentUploadForm, ServiceError> {
131 self.request(
132 Method::GET,
133 Endpoint::service("/v4/attachments/form/upload"),
134 HttpAuthOverride::NoOverride,
135 )?
136 .send()
137 .await?
138 .service_error_for_status()
139 .await?
140 .json()
141 .await
142 .map_err(Into::into)
143 }
144
145 #[tracing::instrument(skip(self), level=tracing::Level::TRACE)]
146 pub(crate) async fn get_attachment_resumable_upload_url(
147 &mut self,
148 attachment_upload_form: &AttachmentUploadForm,
149 ) -> Result<Url, ServiceError> {
150 let mut request = self.request(
151 Method::POST,
152 Endpoint::Absolute(
153 attachment_upload_form.signed_upload_location.clone(),
154 ),
155 HttpAuthOverride::Unidentified,
156 )?;
157
158 for (key, value) in &attachment_upload_form.headers {
159 request = request.header(key, value);
160 }
161 request = request.header(CONTENT_LENGTH, "0");
162
163 if attachment_upload_form.cdn == 2 {
164 request = request.header(CONTENT_TYPE, "application/octet-stream");
165 } else if attachment_upload_form.cdn == 3 {
166 request = request
167 .header("Upload-Defer-Length", "1")
168 .header("Tus-Resumable", "1.0.0");
169 } else {
170 return Err(ServiceError::UnknownCdnVersion(
171 attachment_upload_form.cdn,
172 ));
173 };
174
175 Ok(request
176 .send()
177 .await?
178 .error_for_status()?
179 .headers()
180 .get("location")
181 .ok_or(ServiceError::InvalidFrame {
182 reason: "missing location header in HTTP response",
183 })?
184 .to_str()
185 .map_err(|_| ServiceError::InvalidFrame {
186 reason: "invalid location header bytes in HTTP response",
187 })?
188 .parse()?)
189 }
190
191 #[tracing::instrument(skip(self))]
192 async fn get_attachment_resume_info_cdn2(
193 &mut self,
194 resumable_url: &Url,
195 content_length: u64,
196 ) -> Result<ResumeInfo, ServiceError> {
197 let response = self
198 .request(
199 Method::PUT,
200 Endpoint::cdn_url(2, resumable_url),
201 HttpAuthOverride::Unidentified,
202 )?
203 .header(CONTENT_RANGE, format!("bytes */{content_length}"))
204 .send()
205 .await?
206 .error_for_status()?;
207
208 let status = response.status();
209
210 if status.is_success() {
211 Ok(ResumeInfo {
212 content_range: None,
213 content_start: content_length,
214 })
215 } else if status == StatusCode::PERMANENT_REDIRECT {
216 let offset =
217 match response.headers().get(RANGE) {
218 Some(range) => range
219 .to_str()
220 .map_err(|_| ServiceError::InvalidFrame {
221 reason: "invalid format for Range HTTP header",
222 })?
223 .split('-')
224 .nth(1)
225 .ok_or(ServiceError::InvalidFrame {
226 reason:
227 "invalid value format for Range HTTP header",
228 })?
229 .parse::<u64>()
230 .map_err(|_| ServiceError::InvalidFrame {
231 reason:
232 "invalid number format for Range HTTP header",
233 })?
234 + 1,
235 None => 0,
236 };
237
238 Ok(ResumeInfo {
239 content_range: Some(format!(
240 "bytes {}-{}/{}",
241 offset,
242 content_length - 1,
243 content_length
244 )),
245 content_start: offset,
246 })
247 } else {
248 Err(ServiceError::InvalidFrame {
249 reason: "failed to get resumable upload data from CDN2",
250 })
251 }
252 }
253
254 #[tracing::instrument(skip(self))]
255 async fn get_attachment_resume_info_cdn3(
256 &mut self,
257 resumable_url: &Url,
258 headers: &HashMap<String, String>,
259 ) -> Result<ResumeInfo, ServiceError> {
260 let mut request = self
261 .request(
262 Method::HEAD,
263 Endpoint::cdn_url(3, resumable_url),
264 HttpAuthOverride::Unidentified,
265 )?
266 .header("Tus-Resumable", "1.0.0");
267
268 for (key, value) in headers {
269 request = request.header(key, value);
270 }
271
272 let response = request.send().await?.error_for_status()?;
273
274 let upload_offset = response
275 .headers()
276 .get("upload-offset")
277 .ok_or(ServiceError::InvalidFrame {
278 reason: "no Upload-Offset header in response",
279 })?
280 .to_str()
281 .map_err(|_| ServiceError::InvalidFrame {
282 reason: "invalid upload-offset header bytes in HTTP response",
283 })?
284 .parse()
285 .map_err(|_| ServiceError::InvalidFrame {
286 reason: "invalid integer value for Upload-Offset header",
287 })?;
288
289 Ok(ResumeInfo {
290 content_range: None,
291 content_start: upload_offset,
292 })
293 }
294
295 #[tracing::instrument(skip(self, headers, content))]
299 pub(crate) async fn upload_attachment_v4(
300 &mut self,
301 cdn_id: u32,
302 resumable_url: &Url,
303 content_length: u64,
304 headers: HashMap<String, String>,
305 content: impl std::io::Read + std::io::Seek + Send,
306 ) -> Result<AttachmentDigest, ServiceError> {
307 if cdn_id == 2 {
308 self.upload_to_cdn2(resumable_url, content_length, content)
309 .await
310 } else {
311 self.upload_to_cdn3(
312 resumable_url,
313 &headers,
314 content_length,
315 content,
316 )
317 .await
318 }
319 }
320
321 #[tracing::instrument(skip(self, upload_attributes, reader))]
322 pub async fn upload_to_cdn0(
323 &mut self,
324 path: &str,
325 upload_attributes: AttachmentV2UploadAttributes,
326 filename: String,
327 mut reader: impl Read + Send,
328 ) -> Result<(), ServiceError> {
329 let mut buf = Vec::new();
330 reader
331 .read_to_end(&mut buf)
332 .expect("infallible Read instance");
333
334 let form = reqwest::multipart::Form::new()
337 .text("acl", upload_attributes.acl)
338 .text("key", upload_attributes.key)
339 .text("policy", upload_attributes.policy)
340 .text("Content-Type", "application/octet-stream")
341 .text("x-amz-algorithm", upload_attributes.algorithm)
342 .text("x-amz-credential", upload_attributes.credential)
343 .text("x-amz-date", upload_attributes.date)
344 .text("x-amz-signature", upload_attributes.signature)
345 .part(
346 "file",
347 Part::stream(buf)
348 .mime_str("application/octet-stream")?
349 .file_name(filename),
350 );
351
352 let response = self
353 .request(
354 Method::POST,
355 Endpoint::cdn(0, path),
356 HttpAuthOverride::NoOverride,
357 )?
358 .multipart(form)
359 .send()
360 .await?
361 .error_for_status()?;
362
363 debug!("HyperPushService::PUT response: {:?}", response);
364
365 Ok(())
366 }
367
368 #[tracing::instrument(skip(self, content))]
369 async fn upload_to_cdn2(
370 &mut self,
371 resumable_url: &Url,
372 content_length: u64,
373 mut content: impl std::io::Read + std::io::Seek + Send,
374 ) -> Result<AttachmentDigest, ServiceError> {
375 let resume_info = self
376 .get_attachment_resume_info_cdn2(resumable_url, content_length)
377 .await?;
378
379 let mut digester =
380 crate::digeststream::DigestingReader::new(&mut content);
381
382 let mut buf = Vec::new();
383 digester.read_to_end(&mut buf)?;
384
385 trace!("digested content");
386
387 let mut request = self.request(
388 Method::PUT,
389 Endpoint::cdn_url(2, resumable_url),
390 HttpAuthOverride::Unidentified,
391 )?;
392
393 if let Some(content_range) = resume_info.content_range {
394 request = request.header(CONTENT_RANGE, content_range);
395 }
396
397 request.body(buf).send().await?.error_for_status()?;
398
399 Ok(AttachmentDigest {
400 digest: digester.finalize(),
401 incremental_digest: None,
402 incremental_mac_chunk_size: 0,
403 })
404 }
405
406 #[tracing::instrument(skip(self, content))]
407 async fn upload_to_cdn3(
408 &mut self,
409 resumable_url: &Url,
410 headers: &HashMap<String, String>,
411 content_length: u64,
412 mut content: impl std::io::Read + std::io::Seek + Send,
413 ) -> Result<AttachmentDigest, ServiceError> {
414 let resume_info = self
415 .get_attachment_resume_info_cdn3(resumable_url, headers)
416 .await?;
417
418 trace!(?resume_info, "got resume info");
419
420 if resume_info.content_start == content_length {
421 let mut digester =
422 crate::digeststream::DigestingReader::new(&mut content);
423 let mut buf = Vec::new();
424 digester.read_to_end(&mut buf)?;
425 return Ok(AttachmentDigest {
426 digest: digester.finalize(),
427 incremental_digest: None,
428 incremental_mac_chunk_size: 0,
429 });
430 }
431
432 let mut digester =
433 crate::digeststream::DigestingReader::new(&mut content);
434 digester.seek(SeekFrom::Start(resume_info.content_start))?;
435
436 let mut buf = Vec::new();
437 digester.read_to_end(&mut buf)?;
438
439 trace!("digested content");
440
441 let mut request = self.request(
442 Method::PATCH,
443 Endpoint::cdn(3, resumable_url.path()),
444 HttpAuthOverride::Unidentified,
445 )?;
446
447 for (key, value) in headers {
448 request = request.header(key, value);
449 }
450
451 request
452 .header("Tus-Resumable", "1.0.0")
453 .header("Upload-Offset", resume_info.content_start)
454 .header("Upload-Length", buf.len())
455 .header(CONTENT_TYPE, "application/offset+octet-stream")
456 .body(buf)
457 .send()
458 .await?
459 .error_for_status()?;
460
461 trace!("attachment uploaded");
462
463 Ok(AttachmentDigest {
464 digest: digester.finalize(),
465 incremental_digest: None,
466 incremental_mac_chunk_size: 0,
467 })
468 }
469}