1use crate::static_files::{StaticFile, collect_static_files};
2use anyhow::{Result, anyhow};
3use serde::{Deserialize, Serialize};
4use std::path::Path;
5
6#[derive(Serialize)]
7struct DeployInput<'a> {
8 project_id: &'a str,
9 code_version: u64,
10 bundle_size: u64,
11 files: Vec<DeployFile>,
12 jobs: &'a [CronJob],
13 cron_updated_at: &'a str,
14}
15
16#[derive(Serialize, Deserialize, Clone, Debug)]
17pub struct CronJob {
18 pub function: String,
19 pub every_minutes: u32,
20}
21
22#[derive(Serialize)]
23struct DeployFile {
24 path: String,
25 size: u64,
26}
27
28#[derive(Deserialize)]
29#[serde(tag = "t", rename_all_fields = "camelCase")]
30enum Deploy {
31 Ok {
32 presigned_put_url: String,
33 object_key: String,
34 static_uploads: Vec<StaticUpload>,
35 },
36 QuotaExceeded {
37 reason: String,
38 },
39 BadCodeVersion {
40 reason: String,
41 },
42 NotLoggedIn,
43 NotFound,
44 InternalError,
45}
46
47#[derive(Deserialize)]
48#[serde(rename_all = "camelCase")]
49struct StaticUpload {
50 path: String,
51 presigned_url: String,
52}
53
54#[derive(Serialize)]
55struct DeployStatusInput<'a> {
56 project_id: &'a str,
57 code_version: u64,
58}
59
60#[derive(Deserialize)]
61#[serde(tag = "t", rename_all_fields = "camelCase")]
62enum DeployStatus {
63 Done {
64 active_version: String,
65 pending_version: Option<String>,
66 pending_compiled: bool,
67 compiled_versions: Vec<String>,
68 },
69 Pending {
70 active_version: String,
71 pending_version: Option<String>,
72 pending_compiled: bool,
73 compiled_versions: Vec<String>,
74 },
75 NoActiveVersion,
76 NotLoggedIn,
77 NotFound,
78 InternalError,
79}
80
81#[allow(clippy::too_many_arguments)]
82pub async fn deploy_wasm(
83 control_url: &str,
84 token: &str,
85 project_id: &str,
86 code_version: u64,
87 bundle_tar_path: &Path,
88 jobs: &[CronJob],
89 cron_updated_at: &str,
90) -> Result<()> {
91 let client = reqwest::Client::new();
92 println!("project_id: {project_id}");
93
94 let bundle_size = bundle_size(bundle_tar_path)?;
95
96 let DeployOk {
97 presigned_put_url,
98 object_key,
99 static_uploads: _,
100 } = request_deploy(
101 &client,
102 control_url,
103 token,
104 project_id,
105 code_version,
106 Vec::new(),
107 bundle_size,
108 jobs,
109 cron_updated_at,
110 )
111 .await?;
112
113 println!("uploading bundle to {object_key} (code_version={code_version})...");
114 upload_bundle(&client, &presigned_put_url, bundle_tar_path).await?;
115
116 poll_deploy_status(&client, control_url, token, project_id, code_version).await?;
117 println!("Deploy complete!");
118 Ok(())
119}
120
121struct DeployOk {
122 presigned_put_url: String,
123 object_key: String,
124 static_uploads: Vec<StaticUpload>,
125}
126
127#[allow(clippy::too_many_arguments)]
128pub async fn deploy_forte(
129 control_url: &str,
130 token: &str,
131 project_id: &str,
132 code_version: u64,
133 fe_dist_dir: &Path,
134 bundle_tar_path: &Path,
135 jobs: &[CronJob],
136 cron_updated_at: &str,
137) -> Result<()> {
138 let client = reqwest::Client::new();
139 println!("project_id: {project_id}");
140
141 let static_files = collect_static_files(fe_dist_dir)?;
142 let deploy_files: Vec<DeployFile> = static_files
143 .iter()
144 .map(|f| DeployFile {
145 path: f.relative_path.clone(),
146 size: f.size,
147 })
148 .collect();
149 println!(
150 "Requesting deploy ({} static asset(s))...",
151 deploy_files.len()
152 );
153
154 let bundle_size = bundle_size(bundle_tar_path)?;
155
156 let DeployOk {
157 presigned_put_url,
158 object_key,
159 static_uploads,
160 } = request_deploy(
161 &client,
162 control_url,
163 token,
164 project_id,
165 code_version,
166 deploy_files,
167 bundle_size,
168 jobs,
169 cron_updated_at,
170 )
171 .await?;
172
173 if !static_files.is_empty() {
174 println!("Uploading {} static asset(s)...", static_files.len());
175 upload_static_assets(&client, &static_files, static_uploads).await?;
176 }
177
178 println!("uploading bundle to {object_key} (code_version={code_version})...");
179 upload_bundle(&client, &presigned_put_url, bundle_tar_path).await?;
180
181 poll_deploy_status(&client, control_url, token, project_id, code_version).await?;
182 println!("Deploy complete!");
183 Ok(())
184}
185
186#[allow(clippy::too_many_arguments)]
187async fn request_deploy(
188 client: &reqwest::Client,
189 control_url: &str,
190 token: &str,
191 project_id: &str,
192 code_version: u64,
193 files: Vec<DeployFile>,
194 bundle_size: u64,
195 jobs: &[CronJob],
196 cron_updated_at: &str,
197) -> Result<DeployOk> {
198 let deploy_url = format!(
199 "{}/__forte_action/deploy",
200 control_url.trim_end_matches('/')
201 );
202 let raw: Deploy = client
203 .post(&deploy_url)
204 .bearer_auth(token)
205 .json(&DeployInput {
206 project_id,
207 code_version,
208 bundle_size,
209 files,
210 jobs,
211 cron_updated_at,
212 })
213 .send()
214 .await?
215 .error_for_status()
216 .map_err(|e| anyhow!("deploy failed: {e}"))?
217 .json()
218 .await?;
219 match raw {
220 Deploy::Ok {
221 presigned_put_url,
222 object_key,
223 static_uploads,
224 } => Ok(DeployOk {
225 presigned_put_url,
226 object_key,
227 static_uploads,
228 }),
229 Deploy::QuotaExceeded { reason } => Err(anyhow!("deploy quota exceeded: {reason}")),
230 Deploy::BadCodeVersion { reason } => Err(anyhow!("deploy rejected code_version: {reason}")),
231 Deploy::NotLoggedIn => Err(anyhow!("control rejected token; run `fn0 login` again.")),
232 Deploy::NotFound => Err(anyhow!(
233 "project '{project_id}' not found or not owned by you."
234 )),
235 Deploy::InternalError => Err(anyhow!("deploy: server error; check fn0-control logs")),
236 }
237}
238
239fn bundle_size(bundle_tar_path: &Path) -> Result<u64> {
240 std::fs::metadata(bundle_tar_path)
241 .map(|metadata| metadata.len())
242 .map_err(|e| anyhow!("Failed to stat {}: {}", bundle_tar_path.display(), e))
243}
244
245async fn upload_bundle(
246 client: &reqwest::Client,
247 presigned_put_url: &str,
248 bundle_tar_path: &Path,
249) -> Result<()> {
250 let bundle_bytes = std::fs::read(bundle_tar_path)
251 .map_err(|e| anyhow!("Failed to read {}: {}", bundle_tar_path.display(), e))?;
252 client
253 .put(presigned_put_url)
254 .body(bundle_bytes)
255 .send()
256 .await?
257 .error_for_status()
258 .map_err(|e| anyhow!("bundle upload failed: {e}"))?;
259 Ok(())
260}
261
262async fn upload_static_assets(
263 client: &reqwest::Client,
264 files: &[StaticFile],
265 uploads: Vec<StaticUpload>,
266) -> Result<()> {
267 use futures::StreamExt;
268 use std::collections::HashMap;
269
270 let mut url_for_path: HashMap<String, String> = HashMap::new();
271 for u in uploads {
272 url_for_path.insert(u.path, u.presigned_url);
273 }
274
275 let mut tasks = futures::stream::FuturesUnordered::new();
276 for file in files {
277 let url = url_for_path.remove(&file.relative_path).ok_or_else(|| {
278 anyhow!(
279 "control did not return presigned URL for {}",
280 file.relative_path
281 )
282 })?;
283 let bytes = std::fs::read(&file.absolute_path)
284 .map_err(|e| anyhow!("read {}: {}", file.absolute_path.display(), e))?;
285 let client = client.clone();
286 let content_type = file.content_type;
287 let path = file.relative_path.clone();
288 tasks.push(async move {
289 let resp = client
290 .put(&url)
291 .header("content-type", content_type)
292 .body(bytes)
293 .send()
294 .await
295 .map_err(|e| anyhow!("R2 PUT {}: {}", path, e))?;
296 resp.error_for_status()
297 .map_err(|e| anyhow!("R2 PUT {} HTTP error: {}", path, e))?;
298 Ok::<_, anyhow::Error>(())
299 });
300 }
301 while let Some(result) = tasks.next().await {
302 result?;
303 }
304 Ok(())
305}
306
307async fn poll_deploy_status(
308 client: &reqwest::Client,
309 control_url: &str,
310 token: &str,
311 project_id: &str,
312 code_version: u64,
313) -> Result<()> {
314 let url = format!(
315 "{}/__forte_action/deploy_status",
316 control_url.trim_end_matches('/')
317 );
318 let timeout = std::time::Duration::from_secs(600);
319 let start = std::time::Instant::now();
320 let mut last_state: Option<String> = None;
321
322 loop {
323 let raw: DeployStatus = client
324 .post(&url)
325 .bearer_auth(token)
326 .json(&DeployStatusInput {
327 project_id,
328 code_version,
329 })
330 .send()
331 .await?
332 .error_for_status()
333 .map_err(|e| anyhow!("deploy_status failed: {e}"))?
334 .json()
335 .await?;
336
337 match raw {
338 DeployStatus::Done {
339 active_version,
340 pending_version,
341 pending_compiled,
342 compiled_versions,
343 } => {
344 log_status_line(
345 &active_version,
346 &compiled_versions,
347 &pending_version,
348 pending_compiled,
349 &mut last_state,
350 );
351 return Ok(());
352 }
353 DeployStatus::Pending {
354 active_version,
355 pending_version,
356 pending_compiled,
357 compiled_versions,
358 } => {
359 log_status_line(
360 &active_version,
361 &compiled_versions,
362 &pending_version,
363 pending_compiled,
364 &mut last_state,
365 );
366 if start.elapsed() > timeout {
367 return Err(anyhow!(
368 "deploy_status timed out after {}s",
369 timeout.as_secs()
370 ));
371 }
372 }
373 DeployStatus::NoActiveVersion => {
374 return Err(anyhow!("control has no active fn0-wasmtime version yet"));
375 }
376 DeployStatus::NotLoggedIn => {
377 return Err(anyhow!("control rejected token; run `fn0 login` again."));
378 }
379 DeployStatus::NotFound => {
380 return Err(anyhow!(
381 "project '{project_id}' not found or not owned by you."
382 ));
383 }
384 DeployStatus::InternalError => {
385 return Err(anyhow!(
386 "deploy_status: server error; check fn0-control logs"
387 ));
388 }
389 }
390 }
391}
392
393fn log_status_line(
394 active_version: &str,
395 compiled_versions: &[String],
396 pending_version: &Option<String>,
397 pending_compiled: bool,
398 last_state: &mut Option<String>,
399) {
400 let state = format!(
401 "active={active_version} compiled={compiled_versions:?} pending={pending_version:?} pending_compiled={pending_compiled}",
402 );
403 if last_state.as_deref() != Some(&state) {
404 println!(" {state}");
405 *last_state = Some(state);
406 }
407}
408
409#[cfg(test)]
410mod tests {
411 use super::*;
412 use std::io::Write;
413
414 #[test]
415 fn bundle_size_reads_file_length() {
416 let mut file = tempfile::NamedTempFile::new().unwrap();
417 file.write_all(&[0u8; 1234]).unwrap();
418 assert_eq!(bundle_size(file.path()).unwrap(), 1234);
419 }
420
421 #[test]
424 fn deploy_input_carries_bundle_size() {
425 let input = DeployInput {
426 project_id: "proj",
427 code_version: 42,
428 bundle_size: 999,
429 files: Vec::new(),
430 jobs: &[],
431 cron_updated_at: "2026-07-21T00:00:00Z",
432 };
433 let value = serde_json::to_value(&input).unwrap();
434 assert_eq!(value["bundle_size"], serde_json::json!(999));
435 }
436}