juiceback/
juicehost.rs

1//! Communication with the `juicehost` storage backend.
2//!
3//! Provides streaming and buffered push operations, file rename, and file
4//! deletion. Every function authenticates via `JUICEHOST_API_KEY` and
5//! identifies the origin via `JUICEBACK_ORIGIN`.
6
7use std::sync::Arc;
8
9use crate::state::AppState;
10
11/// Build the authentication and origin headers sent to juicehost.
12pub(crate) fn juicehost_headers(state: &Arc<AppState>) -> reqwest::header::HeaderMap {
13    let mut headers = reqwest::header::HeaderMap::new();
14    if !state.config.juicehost_api_key.is_empty() {
15        headers.insert(
16            "x-juicehost-api-key",
17            state.config.juicehost_api_key.parse().unwrap(),
18        );
19    }
20    if !state.config.juiceback_origin.is_empty() {
21        headers.insert(
22            "x-juiceback-origin",
23            state.config.juiceback_origin.parse().unwrap(),
24        );
25    }
26    headers
27}
28
29/// Stream a file to juicehost by piping chunks from a channel straight into the HTTP body.
30/// Never keeps the whole file in memory. Close the sender to signal EOF.
31///
32/// Pass mode="quic" to send over QUIC instead of regular TCP.
33#[tracing::instrument(skip_all)]
34pub async fn push_file_streaming(
35    state: &Arc<AppState>,
36    id: &str,
37    filename: &str,
38    mime_type: &str,
39    chunk_rx: tokio::sync::mpsc::Receiver<Result<bytes::Bytes, String>>,
40    host: Option<String>,
41    mode: &str,
42) -> Result<(), String> {
43    if mode == "quic" {
44        return crate::quic::push_file_streaming_quic(
45            state, id, filename, mime_type, chunk_rx, host,
46        ).await;
47    }
48
49    let juicehost_url = host.as_deref().unwrap_or(&state.config.juicehost_url);
50    if juicehost_url.is_empty() {
51        return Err("JUICEHOST_URL not set".into());
52    }
53
54    let url = format!("{}/internal/file/stream/{}/{}", juicehost_url, id, filename);
55    let stream = tokio_stream::wrappers::ReceiverStream::new(chunk_rx);
56    let body = reqwest::Body::wrap_stream(stream);
57
58    let req_start = std::time::Instant::now();
59    let resp = state
60        .http
61        .post(&url)
62        .header("x-mime-type", mime_type)
63        .headers(juicehost_headers(state))
64        .body(body)
65        .send()
66        .await
67        .map_err(|e| format!("juicehost streaming push error: {}", e))?;
68    let send_time = req_start.elapsed();
69
70    if resp.status().is_success() {
71        tracing::debug!(
72            "push_stream: id={} url={} send={:?}",
73            id, url, send_time,
74        );
75        tracing::info!("juicehost streaming push: id={} ok", id);
76        Ok(())
77    } else {
78        let status = resp.status();
79        let resp_start = std::time::Instant::now();
80        let body_text = resp.text().await.unwrap_or_default();
81        let resp_time = resp_start.elapsed();
82        let error_code = serde_json::from_str::<serde_json::Value>(&body_text)
83            .ok()
84            .and_then(|v| v.get("error").and_then(|e| e.as_str().map(|s| s.to_string())))
85            .unwrap_or_default();
86        let detail = serde_json::from_str::<serde_json::Value>(&body_text)
87            .ok()
88            .and_then(|v| v.get("message").and_then(|m| m.as_str().map(|s| s.to_string())))
89            .unwrap_or_else(|| body_text.clone());
90        tracing::warn!(
91            "push_stream error: id={} send={:?} resp_read={:?} status={}",
92            id, send_time, resp_time, status.as_u16(),
93        );
94        Err(format!("[{}] {} (status={})", error_code, detail, status.as_u16()))
95    }
96}
97
98/// Send raw bytes to juicehost. The file lands on juicehost's disk, not juiceback's.
99/// Pass a host override or None to use the default URL.
100#[tracing::instrument(skip_all)]
101pub async fn push_file_to_juicehost(
102    state: &Arc<AppState>,
103    id: &str,
104    filename: &str,
105    mime_type: &str,
106    data: &[u8],
107    host: Option<&str>,
108) -> Result<(), String> {
109    let juicehost_url = host.unwrap_or(&state.config.juicehost_url);
110    if juicehost_url.is_empty() {
111        return Err("JUICEHOST_URL not set".into());
112    }
113
114    let build_start = std::time::Instant::now();
115    let file_part = reqwest::multipart::Part::bytes(data.to_vec())
116        .file_name(filename.to_string())
117        .mime_str(mime_type)
118        .map_err(|e| format!("mime error: {}", e))?;
119
120    let form = reqwest::multipart::Form::new()
121        .text("id", id.to_string())
122        .text("filename", filename.to_string())
123        .part("file", file_part);
124    let build_time = build_start.elapsed();
125
126    let send_start = std::time::Instant::now();
127    let resp = state
128        .http
129        .post(&format!("{}/internal/file", juicehost_url))
130        .headers(juicehost_headers(state))
131        .multipart(form)
132        .send()
133        .await
134        .map_err(|e| format!("juicehost push error: {}", e))?;
135    let send_time = send_start.elapsed();
136
137    tracing::debug!("juicehost push: id={} build={:?} send={:?} status={}", id, build_time, send_time, resp.status());
138
139    if resp.status().is_success() {
140        tracing::info!("juicehost push: id={} ok", id);
141        Ok(())
142    } else {
143        let body_start = std::time::Instant::now();
144        let status = resp.status();
145        let body = resp.text().await.unwrap_or_default();
146        let body_time = body_start.elapsed();
147        // Parse JSON response to extract error code + message for clean display
148        let error_code = serde_json::from_str::<serde_json::Value>(&body)
149            .ok()
150            .and_then(|v| v.get("error").and_then(|e| e.as_str().map(|s| s.to_string())))
151            .unwrap_or_default();
152        let detail = serde_json::from_str::<serde_json::Value>(&body)
153            .ok()
154            .and_then(|v| v.get("message").and_then(|m| m.as_str().map(|s| s.to_string())))
155            .unwrap_or_else(|| body.clone());
156        tracing::debug!("juicehost push error: body_read={:?}", body_time);
157        Err(format!("[{}] {} (status={})", error_code, detail, status.as_u16()))
158    }
159}
160
161/// Push a file from a local path to juicehost. Used as a staging step for
162/// TUS uploads that need temporary local storage. The file is deleted from
163/// juiceback's disk after the push.
164pub async fn push_file_path_to_juicehost(
165    state: &Arc<AppState>,
166    id: &str,
167    file_path: &std::path::Path,
168    filename: &str,
169    mime_type: &str,
170    host: Option<&str>,
171) -> Result<(), String> {
172    let data = tokio::fs::read(file_path)
173        .await
174        .map_err(|e| format!("failed to read {}: {}", file_path.display(), e))?;
175
176    let result = push_file_to_juicehost(state, id, filename, mime_type, &data, host).await;
177
178    let _ = tokio::fs::remove_file(file_path).await;
179
180    result
181}
182
183pub async fn rename_file_on_juicehost(state: &Arc<AppState>, old_id: &str, new_id: &str, host: Option<&str>) {
184    let juicehost_url = host.unwrap_or(&state.config.juicehost_url);
185    if juicehost_url.is_empty() {
186        return;
187    }
188
189    let body = serde_json::json!({ "new_id": new_id });
190
191    match state
192        .http
193        .post(&format!("{}/internal/file/{}/rename", juicehost_url, old_id))
194        .headers(juicehost_headers(state))
195        .json(&body)
196        .send()
197        .await
198    {
199        Ok(resp) if resp.status().is_success() => {
200            tracing::info!("juicehost rename: {} -> {} ok", old_id, new_id);
201        }
202        Ok(resp) if resp.status() == reqwest::StatusCode::NOT_FOUND => {
203            tracing::warn!("juicehost rename: {} not found", old_id);
204        }
205        Ok(resp) => {
206            let status = resp.status();
207            let body_text = resp.text().await.unwrap_or_default();
208            tracing::error!("juicehost rename: {} -> {} status={} body={}", old_id, new_id, status, body_text);
209        }
210        Err(e) => {
211            tracing::error!("juicehost rename: {} -> {} error={}", old_id, new_id, e);
212        }
213    }
214}
215
216pub async fn delete_file_on_juicehost(state: &Arc<AppState>, id: &str, host: Option<&str>) {
217    let juicehost_url = host.unwrap_or(&state.config.juicehost_url);
218    if juicehost_url.is_empty() {
219        return;
220    }
221
222    match state
223        .http
224        .delete(&format!("{}/internal/file/{}", juicehost_url, id))
225        .headers(juicehost_headers(state))
226        .send()
227        .await
228    {
229        Ok(resp) if resp.status().is_success() => {
230            tracing::info!("juicehost delete: id={} ok", id);
231        }
232        Ok(resp) if resp.status() == reqwest::StatusCode::NOT_FOUND => {
233            tracing::warn!("juicehost delete: id={} not found (already gone)", id);
234        }
235        Ok(resp) => {
236            let status = resp.status();
237            let body = resp.text().await.unwrap_or_default();
238            tracing::error!("juicehost delete: id={} status={} body={}", id, status, body);
239        }
240        Err(e) => {
241            tracing::error!("juicehost delete: id={} error={}", id, e);
242        }
243    }
244}
juiceback/juicehost.rs

juiceback/
juicehost.rs
