upstream_rs/services/builder/
worker.rs

1use std::fs;
2use std::path::{Path, PathBuf};
3use std::time::{SystemTime, UNIX_EPOCH};
4
5use anyhow::{Context, Result, anyhow};
6
7use crate::models::common::{enums::Channel, version::Version};
8use crate::providers::provider_manager::ProviderManager;
9use crate::services::builder::determine::determine_profile;
10use crate::services::builder::downloader::SourceDownloader;
11use crate::services::builder::profiles::handlers;
12use crate::services::builder::{BuildOutput, BuildRequest, scripts};
13use crate::utils::static_paths::UpstreamPaths;
14
15pub struct BuildWorker<'a> {
16    provider_manager: &'a ProviderManager,
17    paths: &'a UpstreamPaths,
18}
19
20impl<'a> BuildWorker<'a> {
21    pub fn new(provider_manager: &'a ProviderManager, paths: &'a UpstreamPaths) -> Self {
22        Self {
23            provider_manager,
24            paths,
25        }
26    }
27
28    pub async fn build<H>(
29        &self,
30        request: BuildRequest,
31        channel: Channel,
32        line_callback: &mut Option<H>,
33    ) -> Result<BuildOutput>
34    where
35        H: FnMut(&str),
36    {
37        Self::emit_status(line_callback, "Preparing source checkout ...");
38        let downloader = SourceDownloader::new(self.provider_manager, self.paths)?;
39        let source = {
40            let mut status_callback = line_callback
41                .as_mut()
42                .map(|callback| callback as &mut dyn FnMut(&str));
43            downloader
44                .fetch_source(
45                    &request.repo_slug,
46                    &request.provider,
47                    request.base_url.as_deref(),
48                    &channel,
49                    request.version_tag.as_deref(),
50                    request.branch.as_deref(),
51                    &mut status_callback,
52                )
53                .await?
54        };
55
56        Self::emit_status(line_callback, "Detecting build profile ...");
57        let profile_handlers = handlers();
58        let profile = determine_profile(
59            &source.workspace_path,
60            request.requested_profile,
61            &profile_handlers,
62        )
63        .map_err(|err| anyhow!("{} (workspace: '{}')", err, source.workspace_path.display()))?;
64
65        Self::emit_status(
66            line_callback,
67            format!("Building with {profile:?} profile ..."),
68        );
69        let (build_tx, mut build_rx) = tokio::sync::mpsc::unbounded_channel();
70        let workspace_path = source.workspace_path.clone();
71        let package_name = request.name.clone();
72        let mut build_handle = tokio::task::spawn_blocking(move || {
73            let handlers = handlers();
74            let selected = handlers
75                .iter()
76                .find(|handler| handler.profile() == profile)
77                .ok_or_else(|| anyhow!("Unsupported build profile"))?;
78            let mut sender_callback = |line: &str| {
79                let _ = build_tx.send(line.to_string());
80            };
81            let mut build_line_callback: Option<&mut dyn FnMut(&str)> = Some(&mut sender_callback);
82
83            selected.run_build(&workspace_path, &package_name, &mut build_line_callback)
84        });
85
86        let artifact = loop {
87            tokio::select! {
88                Some(line) = build_rx.recv() => {
89                    Self::emit_status(line_callback, line);
90                }
91                result = &mut build_handle => {
92                    while let Ok(line) = build_rx.try_recv() {
93                        Self::emit_status(line_callback, line);
94                    }
95                    break result.context("Build task failed")??;
96                }
97            }
98        };
99        if scripts::script_for(request.script_action, &source.workspace_path).is_some() {
100            Self::emit_status(line_callback, "Running build scripts ...");
101            let build_script_callback = line_callback
102                .as_mut()
103                .map(|callback| callback as &mut dyn FnMut(&str));
104            scripts::run_build_script(
105                request.script_action,
106                &source.workspace_path,
107                build_script_callback,
108            )?;
109        }
110        Self::emit_status(line_callback, "Staging built artifact ...");
111        let persisted_artifact = Self::persist_artifact(&artifact)?;
112
113        let version = if source.release.version == Version::new(0, 0, 0, false) {
114            Version::from_tag(&source.release.tag).unwrap_or_else(|_| Version::new(0, 0, 0, false))
115        } else {
116            source.release.version.clone()
117        };
118
119        Ok(BuildOutput {
120            artifact_path: persisted_artifact,
121            profile,
122            release: source.release,
123            version,
124            branch: source.branch,
125            commit: source.commit,
126        })
127    }
128
129    fn emit_status<H>(line_callback: &mut Option<H>, status: impl AsRef<str>)
130    where
131        H: FnMut(&str),
132    {
133        if let Some(callback) = line_callback.as_mut() {
134            callback(status.as_ref());
135        }
136    }
137
138    fn persist_artifact(artifact_path: &Path) -> Result<PathBuf> {
139        let file_name = artifact_path
140            .file_name()
141            .ok_or_else(|| anyhow!("Built artifact path has no filename"))?;
142        let nonce = SystemTime::now()
143            .duration_since(UNIX_EPOCH)
144            .map(|d| d.as_nanos())
145            .unwrap_or(0);
146        let persist_dir = std::env::temp_dir().join(format!("upstream-artifact-{nonce}"));
147        fs::create_dir_all(&persist_dir).context(format!(
148            "Failed to create artifact staging directory '{}'",
149            persist_dir.display()
150        ))?;
151
152        let persisted_path = persist_dir.join(file_name);
153        fs::copy(artifact_path, &persisted_path).context(format!(
154            "Failed to stage built artifact from '{}' to '{}'",
155            artifact_path.display(),
156            persisted_path.display()
157        ))?;
158
159        let perms = fs::metadata(artifact_path)
160            .context(format!(
161                "Failed to read built artifact metadata '{}'",
162                artifact_path.display()
163            ))?
164            .permissions();
165        fs::set_permissions(&persisted_path, perms).context(format!(
166            "Failed to preserve artifact permissions on '{}'",
167            persisted_path.display()
168        ))?;
169
170        Ok(persisted_path)
171    }
172}
173
174#[cfg(test)]
175mod tests {
176    use super::BuildWorker;
177    use std::io::Write;
178    use std::time::{SystemTime, UNIX_EPOCH};
179    use std::{fs, path::PathBuf};
180
181    fn temp_root(name: &str) -> PathBuf {
182        let nanos = SystemTime::now()
183            .duration_since(UNIX_EPOCH)
184            .map(|d| d.as_nanos())
185            .unwrap_or(0);
186        std::env::temp_dir().join(format!("upstream-worker-test-{name}-{nanos}"))
187    }
188
189    #[test]
190    fn persist_artifact_copies_file_to_stable_temp_path() {
191        let root = temp_root("persist-artifact");
192        fs::create_dir_all(&root).expect("create temp root");
193        let src = root.join("tool");
194        let mut f = fs::File::create(&src).expect("create source artifact");
195        f.write_all(b"binary-data").expect("write source artifact");
196
197        let persisted = BuildWorker::persist_artifact(&src).expect("persist artifact");
198        assert!(persisted.exists());
199        assert_eq!(
200            fs::read(&persisted).expect("read persisted"),
201            b"binary-data"
202        );
203        assert_ne!(persisted, src);
204
205        let _ = fs::remove_dir_all(&root);
206        if let Some(parent) = persisted.parent() {
207            let _ = fs::remove_dir_all(parent);
208        }
209    }
210}
upstream_rs/services/builder/worker.rs

upstream_rs/services/builder/
worker.rs
