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    pub log_len: u64,
91}
92
93pub async fn ask(request: &Request) -> Result<Response, std::io::Error> {
94    ask_at(Path::new(SOCKET_PATH), request, EXCHANGE_DEADLINE).await
95}
96
97async fn ask_at(
98    socket: &Path,
99    request: &Request,
100    deadline: Duration,
101) -> Result<Response, std::io::Error> {
102    let exchange = async {
103        let stream = tokio::net::UnixStream::connect(socket).await?;
104        let (reader, mut writer) = stream.into_split();
105
106        let mut line = serde_json::to_vec(request).map_err(std::io::Error::other)?;
107        line.push(b'\n');
108        writer.write_all(&line).await?;
109        writer.flush().await?;
110
111        let mut answer = Vec::new();
112        tokio::io::BufReader::new(reader.take(MAX_RESPONSE))
113            .read_until(b'\n', &mut answer)
114            .await?;
115
116        serde_json::from_slice(&answer)
117            .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))
118    };
119
120    match tokio::time::timeout(deadline, exchange).await {
121        Ok(answer) => answer,
122        Err(_) => Err(std::io::Error::new(
123            std::io::ErrorKind::TimedOut,
124            "the supervisor did not answer",
125        )),
126    }
127}
128
129pub async fn write_extension(
130    data: &mut (dyn tokio::io::AsyncRead + Unpin + Send),
131) -> Result<ExtensionDistrFile, anyhow::Error> {
132    let tmp_dir = tempfile::tempdir()?;
133    let tmp_path = tmp_dir.path().join("extension.c7s.zip");
134
135    let mut tmp_file = tokio::fs::File::create_new(&tmp_path).await?;
136    tokio::io::copy(data, &mut tmp_file).await?;
137    let tmp_file = tmp_file.into_std().await;
138
139    let distr =
140        tokio::task::spawn_blocking(move || ExtensionDistrFile::parse_from_reader(tmp_file))
141            .await??;
142
143    let identifier = distr.metadata_toml.get_package_identifier();
144    if !MetadataToml::is_valid_package_identifier(&identifier) {
145        return Err(anyhow::anyhow!("invalid package identifier `{identifier}`"));
146    }
147
148    tokio::fs::copy(
149        tmp_path,
150        Path::new(EXTENSION_DIR).join(format!("{}.c7s.zip", identifier)),
151    )
152    .await?;
153
154    Ok(distr)
155}
156
157pub async fn remove_extension(package_name: &str) -> Result<(), std::io::Error> {
158    let identifier = MetadataToml::convert_package_name_to_identifier(package_name);
159    if !MetadataToml::is_valid_package_identifier(&identifier) {
160        return Err(std::io::Error::new(
161            std::io::ErrorKind::InvalidInput,
162            format!("invalid package identifier `{identifier}`"),
163        ));
164    }
165
166    let path = Path::new(EXTENSION_DIR).join(format!("{}.c7s.zip", identifier));
167
168    tokio::fs::remove_file(path).await?;
169
170    Ok(())
171}
172
173pub async fn list_extensions() -> Result<Vec<ExtensionDistrFile>, anyhow::Error> {
174    let mut entries = tokio::fs::read_dir(EXTENSION_DIR).await?;
175    let mut extensions = Vec::new();
176
177    while let Some(entry) = entries.next_entry().await? {
178        if entry.file_type().await?.is_file() {
179            let path = entry.path();
180            if path.extension().and_then(|s| s.to_str()) == Some("zip") {
181                let file = tokio::fs::File::open(path).await?;
182                let file = file.into_std().await;
183                let distr = match tokio::task::spawn_blocking(move || {
184                    ExtensionDistrFile::parse_from_reader(file)
185                })
186                .await
187                {
188                    Ok(Ok(d)) => d,
189                    _ => continue,
190                };
191
192                extensions.push(distr);
193            }
194        }
195    }
196
197    Ok(extensions)
198}
shared/heavy.rs

shared/
heavy.rs
