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 continue;
179 }
180
181 let update_information = match extension
182 .check_for_updates(state.clone(), &extension.version)
183 .await
184 {
185 Ok(info) => info,
186 Err(err) => {
187 tracing::error!(
188 "failed to check for updates for extension {}: {:#?}",
189 extension.package_name,
190 err
191 );
192
193 update_info.extensions.insert(
194 extension.package_name,
195 ExtensionUpdateCheckResult::Error {
196 error: err.to_compact_string(),
197 },
198 );
199
200 continue;
201 }
202 };
203
204 if let Some(info) = update_information {
205 update_info.extensions.insert(
206 extension.package_name,
207 ExtensionUpdateCheckResult::UpdateAvailable {
208 version: extension.version.clone(),
209 latest_version: info.version,
210 changes: info.changes,
211 },
212 );
213 } else {
214 update_info.extensions.insert(
215 extension.package_name,
216 ExtensionUpdateCheckResult::NoUpdate,
217 );
218 }
219 }
220
221 let update_info = Arc::new(update_info);
222 *latest_info.write().await = Some(update_info.clone());
223 let _ = recheck_finished_sender.send(Ok(update_info));
224
225 Ok(())
226 };
227
228 if let Err(err) = run_inner().await {
229 tracing::error!("failed to check for updates: {:#?}", err);
230 let _ = recheck_finished_sender.send(Err(Arc::new(err)));
231 }
232
233 tracing::info!("finished update check, waiting for 12h or recheck trigger");
234
235 tokio::select! {
236 _ = recheck_notifier.notified() => {}
237 _ = tokio::time::sleep(std::time::Duration::from_hours(12)) => {}
238 }
239 }
240 }
241 });
242
243 tokio::spawn(async move {
244 let run = async || -> Result<(), anyhow::Error> {
245 sqlx::query(
246 "INSERT INTO version_history (extension, version) VALUES ('', $1)
247 ON CONFLICT (extension, version) DO NOTHING",
248 )
249 .bind(&state.version)
250 .execute(state.database.write())
251 .await?;
252
253 for extension in state.extensions.extensions().await.iter() {
254 sqlx::query(
255 "INSERT INTO version_history (extension, version) VALUES ($1, $2)
256 ON CONFLICT (extension, version) DO NOTHING",
257 )
258 .bind(extension.package_name)
259 .bind(extension.version.to_string())
260 .execute(state.database.write())
261 .await?;
262 }
263
264 let rows = sqlx::query("SELECT extension, version, installed FROM version_history ORDER BY version_history.installed DESC")
265 .fetch_all(state.database.read())
266 .await?;
267
268 let mut panel_history = Vec::new();
269 let mut extension_history = BTreeMap::new();
270
271 for row in rows {
272 let extension: compact_str::CompactString = row.try_get("extension")?;
273
274 let entry = VersionHistoryEntry {
275 version: row.try_get("version")?,
276 timestamp: row
277 .try_get::<chrono::NaiveDateTime, _>("installed")?
278 .and_utc(),
279 };
280
281 if extension.is_empty() {
282 panel_history.push(entry);
283 } else if let Some(ext) = state
284 .extensions
285 .extensions()
286 .await
287 .iter()
288 .find(|ext| ext.package_name == extension)
289 {
290 extension_history
291 .entry(ext.package_name)
292 .or_insert_with(Vec::new)
293 .push(entry);
294 }
295 }
296
297 *state.updates.panel_version_history.write().await = panel_history;
298 *state.updates.extension_version_history.write().await = extension_history;
299
300 Ok(())
301 };
302
303 if let Err(err) = run().await {
304 tracing::error!("failed to track version history: {:#?}", err);
305 }
306 });
307 }
308
309 pub async fn get_panel_version_history(&self) -> RwLockReadGuard<'_, Vec<VersionHistoryEntry>> {
310 self.panel_version_history.read().await
311 }
312
313 pub async fn get_extension_version_history(
314 &self,
315 ) -> RwLockReadGuard<'_, BTreeMap<&'static str, Vec<VersionHistoryEntry>>> {
316 self.extension_version_history.read().await
317 }
318
319 pub async fn get_update_information(&self) -> Option<Arc<UpdateInformation>> {
320 self.latest_info.read().await.clone()
321 }
322
323 pub fn trigger_recheck(&self) {
324 self.recheck_notifier.notify_waiters();
325 }
326
327 pub async fn trigger_recheck_and_wait(&self) -> ChannelData {
328 self.trigger_recheck();
329 self.recheck_finished_receiver
330 .resubscribe()
331 .recv()
332 .await
333 .map_err(|err| {
334 Arc::new(anyhow::anyhow!(
335 "failed to receive update check result: {:#?}",
336 err
337 ))
338 })
339 .flatten()
340 }
341}