1use compact_str::ToCompactString;
2use serde::{Deserialize, Serialize};
3use sqlx::Row;
4use std::{collections::BTreeMap, fmt::Display, str::FromStr, sync::Arc};
5use tokio::sync::{RwLock, RwLockReadGuard};
6use utoipa::ToSchema;
7
8#[derive(ToSchema, Serialize, Deserialize, Clone)]
9pub struct VersionHistoryEntry {
10 version: compact_str::CompactString,
11 timestamp: chrono::DateTime<chrono::Utc>,
12}
13
14#[derive(ToSchema, Serialize, Deserialize, Clone)]
19pub struct ParsedVersionInformation {
20 #[schema(value_type = String)]
21 pub version: semver::Version,
22 pub commit: Option<compact_str::CompactString>,
23 pub branch: Option<compact_str::CompactString>,
24}
25
26impl FromStr for ParsedVersionInformation {
27 type Err = anyhow::Error;
28
29 fn from_str(s: &str) -> Result<Self, Self::Err> {
30 if let Ok(version) = semver::Version::parse(s) {
31 return Ok(Self {
32 version,
33 commit: None,
34 branch: None,
35 });
36 }
37
38 let (version, commit_branch) = if let Some((version, commit)) = s.split_once(':') {
39 (version, Some(commit))
40 } else {
41 (s, None)
42 };
43 let (commit, branch) =
44 if let Some((commit, branch)) = commit_branch.and_then(|cb| cb.split_once('@')) {
45 (
46 Some(commit.to_compact_string()),
47 Some(branch.to_compact_string()),
48 )
49 } else {
50 (commit_branch.map(|c| c.to_compact_string()), None)
51 };
52
53 Ok(Self {
54 version: semver::Version::parse(version)?,
55 commit,
56 branch,
57 })
58 }
59}
60
61impl Display for ParsedVersionInformation {
62 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63 if let Some(commit) = &self.commit {
64 if let Some(branch) = &self.branch {
65 write!(f, "{}:{}@{}", self.version, commit, branch)
66 } else {
67 write!(f, "{}:{}", self.version, commit)
68 }
69 } else {
70 write!(f, "{}", self.version)
71 }
72 }
73}
74
75#[derive(ToSchema, Serialize, Clone)]
76#[serde(tag = "type", rename_all = "snake_case")]
77pub enum ExtensionUpdateCheckResult {
78 NoUpdate,
79 UpdateAvailable {
80 #[schema(value_type = String)]
81 version: semver::Version,
82 #[schema(value_type = String)]
83 latest_version: semver::Version,
84 changes: Vec<compact_str::CompactString>,
85 },
86 Error {
87 error: compact_str::CompactString,
88 },
89}
90
91#[derive(ToSchema, Serialize, Clone)]
92pub struct UpdateInformation {
93 pub panel_version: compact_str::CompactString,
94 #[schema(value_type = String)]
95 pub latest_panel_version: semver::Version,
96 #[schema(value_type = String)]
97 pub latest_wings_version: semver::Version,
98 #[schema(value_type = String)]
99 pub latest_db_agent_version: semver::Version,
100
101 pub extensions: BTreeMap<&'static str, ExtensionUpdateCheckResult>,
102}
103
104type ChannelData = Result<Arc<UpdateInformation>, Arc<anyhow::Error>>;
105
106pub struct UpdateManager {
107 recheck_notifier: Arc<tokio::sync::Notify>,
108 recheck_finished_receiver: tokio::sync::broadcast::Receiver<ChannelData>,
109 recheck_finished_sender: tokio::sync::broadcast::Sender<ChannelData>,
110
111 latest_info: Arc<RwLock<Option<Arc<UpdateInformation>>>>,
112 panel_version_history: Arc<RwLock<Vec<VersionHistoryEntry>>>,
113 extension_version_history: Arc<RwLock<BTreeMap<&'static str, Vec<VersionHistoryEntry>>>>,
114}
115
116impl Default for UpdateManager {
117 fn default() -> Self {
118 let (recheck_finished_sender, recheck_finished_receiver) =
119 tokio::sync::broadcast::channel(1);
120
121 Self {
122 recheck_notifier: Arc::new(tokio::sync::Notify::new()),
123 recheck_finished_receiver,
124 recheck_finished_sender,
125 latest_info: Arc::new(RwLock::new(None)),
126 panel_version_history: Arc::new(RwLock::new(Vec::new())),
127 extension_version_history: Arc::new(RwLock::new(BTreeMap::new())),
128 }
129 }
130}
131
132impl UpdateManager {
133 pub fn init(&self, state: super::State) {
134 if !state.env.app_primary {
135 return;
136 }
137
138 let recheck_notifier = self.recheck_notifier.clone();
139 let recheck_finished_sender = self.recheck_finished_sender.clone();
140 let latest_info = self.latest_info.clone();
141
142 tokio::spawn({
143 let state = state.clone();
144
145 async move {
146 loop {
147 let run_inner = async || -> Result<(), anyhow::Error> {
148 let data: Response = state
149 .client
150 .get("https://calagopus.com/api/latest")
151 .send()
152 .await?
153 .json()
154 .await?;
155
156 #[derive(Deserialize)]
157 struct Response {
158 versions: ResponseVersions,
159 }
160
161 #[derive(Deserialize)]
162 struct ResponseVersions {
163 panel: semver::Version,
164 wings: semver::Version,
165 db_agent: semver::Version,
166 }
167
168 let mut update_info = UpdateInformation {
169 panel_version: state.version.to_compact_string(),
170 latest_panel_version: data.versions.panel,
171 latest_wings_version: data.versions.wings,
172 latest_db_agent_version: data.versions.db_agent,
173 extensions: BTreeMap::new(),
174 };
175
176 for extension in state.extensions.extensions().await.iter() {
177 if state.extensions.is_disabled(extension.package_name)
178 || state.extensions.is_failed(extension.package_name)
179 {
180 continue;
181 }
182
183 let update_information = match extension
184 .check_for_updates(state.clone(), &extension.version)
185 .await
186 {
187 Ok(info) => info,
188 Err(err) => {
189 tracing::error!(
190 "failed to check for updates for extension {}: {:#?}",
191 extension.package_name,
192 err
193 );
194
195 update_info.extensions.insert(
196 extension.package_name,
197 ExtensionUpdateCheckResult::Error {
198 error: err.to_compact_string(),
199 },
200 );
201
202 continue;
203 }
204 };
205
206 if let Some(info) = update_information {
207 update_info.extensions.insert(
208 extension.package_name,
209 ExtensionUpdateCheckResult::UpdateAvailable {
210 version: extension.version.clone(),
211 latest_version: info.version,
212 changes: info.changes,
213 },
214 );
215 } else {
216 update_info.extensions.insert(
217 extension.package_name,
218 ExtensionUpdateCheckResult::NoUpdate,
219 );
220 }
221 }
222
223 let update_info = Arc::new(update_info);
224 *latest_info.write().await = Some(update_info.clone());
225 let _ = recheck_finished_sender.send(Ok(update_info));
226
227 Ok(())
228 };
229
230 if let Err(err) = run_inner().await {
231 tracing::error!("failed to check for updates: {:#?}", err);
232 let _ = recheck_finished_sender.send(Err(Arc::new(err)));
233 }
234
235 tracing::info!("finished update check, waiting for 12h or recheck trigger");
236
237 tokio::select! {
238 _ = recheck_notifier.notified() => {}
239 _ = tokio::time::sleep(std::time::Duration::from_hours(12)) => {}
240 }
241 }
242 }
243 });
244
245 tokio::spawn(async move {
246 let run = async || -> Result<(), anyhow::Error> {
247 sqlx::query(
248 "INSERT INTO version_history (extension, version) VALUES ('', $1)
249 ON CONFLICT (extension, version) DO NOTHING",
250 )
251 .bind(&state.version)
252 .execute(state.database.write())
253 .await?;
254
255 for extension in state.extensions.extensions().await.iter() {
256 sqlx::query(
257 "INSERT INTO version_history (extension, version) VALUES ($1, $2)
258 ON CONFLICT (extension, version) DO NOTHING",
259 )
260 .bind(extension.package_name)
261 .bind(extension.version.to_string())
262 .execute(state.database.write())
263 .await?;
264 }
265
266 let rows = sqlx::query("SELECT extension, version, installed FROM version_history ORDER BY version_history.installed DESC")
267 .fetch_all(state.database.read())
268 .await?;
269
270 let mut panel_history = Vec::new();
271 let mut extension_history = BTreeMap::new();
272
273 for row in rows {
274 let extension: compact_str::CompactString = row.try_get("extension")?;
275
276 let entry = VersionHistoryEntry {
277 version: row.try_get("version")?,
278 timestamp: row
279 .try_get::<chrono::NaiveDateTime, _>("installed")?
280 .and_utc(),
281 };
282
283 if extension.is_empty() {
284 panel_history.push(entry);
285 } else if let Some(ext) = state
286 .extensions
287 .extensions()
288 .await
289 .iter()
290 .find(|ext| ext.package_name == extension)
291 {
292 extension_history
293 .entry(ext.package_name)
294 .or_insert_with(Vec::new)
295 .push(entry);
296 }
297 }
298
299 *state.updates.panel_version_history.write().await = panel_history;
300 *state.updates.extension_version_history.write().await = extension_history;
301
302 Ok(())
303 };
304
305 if let Err(err) = run().await {
306 tracing::error!("failed to track version history: {:#?}", err);
307 }
308 });
309 }
310
311 pub async fn get_panel_version_history(&self) -> RwLockReadGuard<'_, Vec<VersionHistoryEntry>> {
312 self.panel_version_history.read().await
313 }
314
315 pub async fn get_extension_version_history(
316 &self,
317 ) -> RwLockReadGuard<'_, BTreeMap<&'static str, Vec<VersionHistoryEntry>>> {
318 self.extension_version_history.read().await
319 }
320
321 pub async fn get_update_information(&self) -> Option<Arc<UpdateInformation>> {
322 self.latest_info.read().await.clone()
323 }
324
325 pub fn trigger_recheck(&self) {
326 self.recheck_notifier.notify_waiters();
327 }
328
329 pub async fn trigger_recheck_and_wait(&self) -> ChannelData {
330 self.trigger_recheck();
331 self.recheck_finished_receiver
332 .resubscribe()
333 .recv()
334 .await
335 .map_err(|err| {
336 Arc::new(anyhow::anyhow!(
337 "failed to receive update check result: {:#?}",
338 err
339 ))
340 })
341 .flatten()
342 }
343}