shared/
heavy.rs

1use crate::extensions::distr::{ExtensionDistrFile, MetadataToml};
2use serde::{Deserialize, Serialize};
3use std::{path::Path, time::Duration};
4use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt};
5use utoipa::ToSchema;
6
7pub static EXTENSION_DIR: &str = "/app/extensions";
8pub static SOCKET_PATH: &str = "/tmp/calagopus/supervisor.sock";
9
10const EXCHANGE_DEADLINE: Duration = Duration::from_secs(15);
11const MAX_RESPONSE: u64 = 1024 * 1024;
12
13#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
14#[serde(tag = "type", rename_all = "snake_case")]
15pub enum Request {
16    GetStatus,
17    RequestRebuild {
18        force: bool,
19    },
20    StreamLog {
21        build_id: Option<u64>,
22        from_offset: u64,
23    },
24    Cancel {
25        build_id: Option<u64>,
26    },
27    RequestRestart,
28}
29
30#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
31#[serde(tag = "type", rename_all = "snake_case")]
32pub enum Response {
33    Status(Status),
34    RebuildAccepted {
35        build_id: u64,
36    },
37    RebuildAlreadyRunning {
38        build_id: u64,
39    },
40    LogChunk {
41        offset: u64,
42        data: String,
43        eof: bool,
44    },
45    CancelAccepted {
46        build_id: u64,
47    },
48    CancelNotRunning,
49    RestartAccepted,
50    Error {
51        message: String,
52    },
53}
54
55#[derive(Debug, ToSchema, Serialize, Deserialize, Clone, Copy, PartialEq)]
56#[serde(tag = "type", rename_all = "snake_case")]
57pub enum BuildPhase {
58    Preparing,
59    Clearing,
60    Adding { done: u32, total: u32 },
61    Resync,
62    StagingTranslations,
63    Building,
64    Verifying,
65    Installing,
66    Restarting,
67}
68
69#[derive(Debug, ToSchema, Serialize, Deserialize, Clone, Copy, PartialEq)]
70#[serde(tag = "type", rename_all = "snake_case")]
71pub enum SupervisorState {
72    Idle,
73    Queued,
74    Building { phase: BuildPhase },
75    Succeeded,
76    Failed,
77}
78
79#[derive(Debug, ToSchema, Serialize, Deserialize, Clone, PartialEq)]
80pub struct Status {
81    pub state: SupervisorState,
82    pub panel_version: String,
83    pub cache_key: String,
84    pub bin_name: String,
85    pub build_id: Option<u64>,
86    pub started_at: Option<String>,
87    pub finished_at: Option<String>,
88    pub exit_code: Option<i32>,
89    pub failure_reason: Option<String>,
90    #[serde(default)]
91    pub failed_extensions: Vec<String>,
92    pub log_len: u64,
93}
94
95pub async fn ask(request: &Request) -> Result<Response, std::io::Error> {
96    ask_at(Path::new(SOCKET_PATH), request, EXCHANGE_DEADLINE).await
97}
98
99async fn ask_at(
100    socket: &Path,
101    request: &Request,
102    deadline: Duration,
103) -> Result<Response, std::io::Error> {
104    let exchange = async {
105        let stream = tokio::net::UnixStream::connect(socket).await?;
106        let (reader, mut writer) = stream.into_split();
107
108        let mut line = serde_json::to_vec(request).map_err(std::io::Error::other)?;
109        line.push(b'\n');
110        writer.write_all(&line).await?;
111        writer.flush().await?;
112
113        let mut answer = Vec::new();
114        tokio::io::BufReader::new(reader.take(MAX_RESPONSE))
115            .read_until(b'\n', &mut answer)
116            .await?;
117
118        serde_json::from_slice(&answer)
119            .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))
120    };
121
122    match tokio::time::timeout(deadline, exchange).await {
123        Ok(answer) => answer,
124        Err(_) => Err(std::io::Error::new(
125            std::io::ErrorKind::TimedOut,
126            "the supervisor did not answer",
127        )),
128    }
129}
130
131const STAGED_FILE_NAME: &str = "extension.c7s.zip";
132
133pub struct StagedExtension {
134    pub distr: ExtensionDistrFile,
135    tmp_dir: tempfile::TempDir,
136}
137
138impl StagedExtension {
139    pub async fn install(self) -> Result<ExtensionDistrFile, std::io::Error> {
140        tokio::fs::copy(
141            self.tmp_dir.path().join(STAGED_FILE_NAME),
142            Path::new(EXTENSION_DIR).join(format!(
143                "{}.c7s.zip",
144                self.distr.metadata_toml.get_package_identifier()
145            )),
146        )
147        .await?;
148
149        Ok(self.distr)
150    }
151}
152
153pub async fn stage_extension(
154    data: &mut (dyn tokio::io::AsyncRead + Unpin + Send),
155) -> Result<StagedExtension, anyhow::Error> {
156    let tmp_dir = tempfile::tempdir()?;
157    let tmp_path = tmp_dir.path().join(STAGED_FILE_NAME);
158
159    let mut tmp_file = tokio::fs::File::create_new(&tmp_path).await?;
160    tokio::io::copy(data, &mut tmp_file).await?;
161    let tmp_file = tmp_file.into_std().await;
162
163    let distr =
164        tokio::task::spawn_blocking(move || ExtensionDistrFile::parse_from_reader(tmp_file))
165            .await??;
166
167    let identifier = distr.metadata_toml.get_package_identifier();
168    if !MetadataToml::is_valid_package_identifier(&identifier) {
169        return Err(anyhow::anyhow!("invalid package identifier `{identifier}`"));
170    }
171
172    Ok(StagedExtension { distr, tmp_dir })
173}
174
175pub async fn build_target_version() -> semver::Version {
176    match ask(&Request::GetStatus).await {
177        Ok(Response::Status(status)) => match status.panel_version.parse() {
178            Ok(version) => return version,
179            Err(err) => tracing::error!(
180                "the extension supervisor reported an unparsable panel version {:?}: {err}",
181                status.panel_version
182            ),
183        },
184        Ok(answer) => {
185            tracing::error!("the extension supervisor answered a status request with {answer:?}")
186        }
187        Err(err) => tracing::error!("the extension supervisor could not be reached: {err}"),
188    }
189
190    crate::extensions::distr::running_panel_version()
191}
192
193pub async fn remove_extension(package_name: &str) -> Result<(), std::io::Error> {
194    let identifier = MetadataToml::convert_package_name_to_identifier(package_name);
195    if !MetadataToml::is_valid_package_identifier(&identifier) {
196        return Err(std::io::Error::new(
197            std::io::ErrorKind::InvalidInput,
198            format!("invalid package identifier `{identifier}`"),
199        ));
200    }
201
202    let path = Path::new(EXTENSION_DIR).join(format!("{}.c7s.zip", identifier));
203
204    tokio::fs::remove_file(path).await?;
205
206    Ok(())
207}
208
209pub async fn list_extensions() -> Result<Vec<ExtensionDistrFile>, anyhow::Error> {
210    let mut entries = tokio::fs::read_dir(EXTENSION_DIR).await?;
211    let mut extensions = Vec::new();
212
213    while let Some(entry) = entries.next_entry().await? {
214        if entry.file_type().await?.is_file() {
215            let path = entry.path();
216            if path.extension().and_then(|s| s.to_str()) == Some("zip") {
217                let file = tokio::fs::File::open(path).await?;
218                let file = file.into_std().await;
219                let distr = match tokio::task::spawn_blocking(move || {
220                    ExtensionDistrFile::parse_from_reader(file)
221                })
222                .await
223                {
224                    Ok(Ok(d)) => d,
225                    _ => continue,
226                };
227
228                extensions.push(distr);
229            }
230        }
231    }
232
233    Ok(extensions)
234}
shared/heavy.rs

shared/
heavy.rs
