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