1use crate::{
2 models::{
3 CreatableModel, CreateListenerList, InsertQueryBuilder, UpdatableModel, UpdateHandlerList,
4 UpdateQueryBuilder,
5 },
6 prelude::*,
7};
8use compact_str::ToCompactString;
9use garde::Validate;
10use rand::{RngExt, distr::SampleString};
11use serde::{Deserialize, Serialize};
12use sha2::Digest;
13use sqlx::{Row, postgres::PgRow};
14use std::{
15 collections::{BTreeMap, HashMap},
16 sync::{Arc, LazyLock},
17};
18use utoipa::ToSchema;
19
20mod events;
21pub use events::NodeEvent;
22
23pub type GetNode = crate::extract::ConsumingExtension<Node>;
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum NodeDeploymentBlocker {
27 NoNodes,
29 DeploymentDisabled,
31 InsufficientMemory,
33 InsufficientDisk,
35 InsufficientResources,
37 ResourcesSplitAcrossNodes,
40}
41
42impl NodeDeploymentBlocker {
43 pub fn message(self) -> &'static str {
44 match self {
45 Self::NoNodes => "no nodes exist in the selected location(s)",
46 Self::DeploymentDisabled => {
47 "deployment is disabled on every node in the selected location(s)"
48 }
49 Self::InsufficientMemory => {
50 "no node in the selected location(s) has enough unallocated memory"
51 }
52 Self::InsufficientDisk => {
53 "no node in the selected location(s) has enough unallocated disk"
54 }
55 Self::InsufficientResources => {
56 "no node in the selected location(s) has enough unallocated memory or disk"
57 }
58 Self::ResourcesSplitAcrossNodes => {
59 "no single node in the selected location(s) has enough unallocated memory and disk at the same time"
60 }
61 }
62 }
63}
64
65#[derive(Serialize, Deserialize, Clone)]
66pub struct Node {
67 pub uuid: uuid::Uuid,
68 pub location: super::location::Location,
69 pub backup_configuration: Option<Fetchable<super::backup_configuration::BackupConfiguration>>,
70
71 pub name: compact_str::CompactString,
72 pub description: Option<compact_str::CompactString>,
73
74 pub deployment_enabled: bool,
75 pub maintenance_enabled: bool,
76
77 pub public_url: Option<reqwest::Url>,
78 pub url: reqwest::Url,
79 pub sftp_host: Option<compact_str::CompactString>,
80 pub sftp_port: i32,
81
82 pub memory: i64,
83 pub disk: i64,
84
85 pub token_id: compact_str::CompactString,
86 pub token: Vec<u8>,
87
88 pub created: chrono::NaiveDateTime,
89
90 extension_data: super::ModelExtensionData,
91}
92
93impl BaseModel for Node {
94 const NAME: &'static str = "node";
95
96 fn get_extension_list() -> &'static super::ModelExtensionList {
97 static EXTENSIONS: LazyLock<super::ModelExtensionList> =
98 LazyLock::new(|| parking_lot::RwLock::new(Vec::new()));
99
100 &EXTENSIONS
101 }
102
103 fn get_extension_data(&self) -> &super::ModelExtensionData {
104 &self.extension_data
105 }
106
107 #[inline]
108 fn base_columns(prefix: Option<&str>) -> BTreeMap<&'static str, compact_str::CompactString> {
109 let prefix = prefix.unwrap_or_default();
110
111 let mut columns = BTreeMap::from([
112 ("nodes.uuid", compact_str::format_compact!("{prefix}uuid")),
113 (
114 "nodes.backup_configuration_uuid",
115 compact_str::format_compact!("{prefix}node_backup_configuration_uuid"),
116 ),
117 ("nodes.name", compact_str::format_compact!("{prefix}name")),
118 (
119 "nodes.description",
120 compact_str::format_compact!("{prefix}description"),
121 ),
122 (
123 "nodes.deployment_enabled",
124 compact_str::format_compact!("{prefix}deployment_enabled"),
125 ),
126 (
127 "nodes.maintenance_enabled",
128 compact_str::format_compact!("{prefix}maintenance_enabled"),
129 ),
130 (
131 "nodes.public_url",
132 compact_str::format_compact!("{prefix}public_url"),
133 ),
134 ("nodes.url", compact_str::format_compact!("{prefix}url")),
135 (
136 "nodes.sftp_host",
137 compact_str::format_compact!("{prefix}sftp_host"),
138 ),
139 (
140 "nodes.sftp_port",
141 compact_str::format_compact!("{prefix}sftp_port"),
142 ),
143 (
144 "nodes.memory",
145 compact_str::format_compact!("{prefix}memory"),
146 ),
147 ("nodes.disk", compact_str::format_compact!("{prefix}disk")),
148 (
149 "nodes.token_id",
150 compact_str::format_compact!("{prefix}token_id"),
151 ),
152 ("nodes.token", compact_str::format_compact!("{prefix}token")),
153 (
154 "nodes.created",
155 compact_str::format_compact!("{prefix}created"),
156 ),
157 ]);
158
159 columns.extend(super::location::Location::base_columns(Some("location_")));
160
161 columns
162 }
163
164 #[inline]
165 fn map(prefix: Option<&str>, row: &PgRow) -> Result<Self, crate::database::DatabaseError> {
166 let prefix = prefix.unwrap_or_default();
167
168 Ok(Self {
169 uuid: row.try_get(compact_str::format_compact!("{prefix}uuid").as_str())?,
170 location: super::location::Location::map(Some("location_"), row)?,
171 backup_configuration:
172 super::backup_configuration::BackupConfiguration::get_fetchable_from_row(
173 row,
174 compact_str::format_compact!("{prefix}node_backup_configuration_uuid"),
175 ),
176 name: row.try_get(compact_str::format_compact!("{prefix}name").as_str())?,
177 description: row
178 .try_get(compact_str::format_compact!("{prefix}description").as_str())?,
179 deployment_enabled: row
180 .try_get(compact_str::format_compact!("{prefix}deployment_enabled").as_str())?,
181 maintenance_enabled: row
182 .try_get(compact_str::format_compact!("{prefix}maintenance_enabled").as_str())?,
183 public_url: row
184 .try_get::<Option<String>, _>(
185 compact_str::format_compact!("{prefix}public_url").as_str(),
186 )?
187 .try_map(|url| url.parse())
188 .map_err(anyhow::Error::new)?,
189 url: row
190 .try_get::<String, _>(compact_str::format_compact!("{prefix}url").as_str())?
191 .parse()
192 .map_err(anyhow::Error::new)?,
193 sftp_host: row.try_get(compact_str::format_compact!("{prefix}sftp_host").as_str())?,
194 sftp_port: row.try_get(compact_str::format_compact!("{prefix}sftp_port").as_str())?,
195 memory: row.try_get(compact_str::format_compact!("{prefix}memory").as_str())?,
196 disk: row.try_get(compact_str::format_compact!("{prefix}disk").as_str())?,
197 token_id: row.try_get(compact_str::format_compact!("{prefix}token_id").as_str())?,
198 token: row.try_get(compact_str::format_compact!("{prefix}token").as_str())?,
199 created: row.try_get(compact_str::format_compact!("{prefix}created").as_str())?,
200 extension_data: Self::map_extensions(prefix, row)?,
201 })
202 }
203}
204
205impl Node {
206 pub const AIO_NODE_UUID: uuid::Uuid = uuid::uuid!("7dbbbb63-1734-48c4-e1de-d1a65f62cada");
207
208 pub async fn by_token_id_token_cached(
209 database: &crate::database::Database,
210 token_id: &str,
211 token: &str,
212 ) -> Result<Option<Self>, anyhow::Error> {
213 database
214 .cache
215 .cached(
216 &format!(
217 "node::token::{token_id}.{}",
218 hex::encode(sha2::Sha256::digest(token.as_bytes()))
219 ),
220 10,
221 || async {
222 let row = sqlx::query(sqlx::AssertSqlSafe(format!(
223 r#"
224 SELECT {}
225 FROM nodes
226 JOIN locations ON locations.uuid = nodes.location_uuid
227 WHERE nodes.token_id = $1
228 "#,
229 Self::columns_sql(None)
230 )))
231 .bind(token_id)
232 .fetch_optional(database.read())
233 .await?;
234
235 Ok::<_, anyhow::Error>(
236 if let Some(node) = row.try_map(|row| Self::map(None, &row))? {
237 if constant_time_eq::constant_time_eq(
238 database.decrypt(node.token.clone()).await?.as_bytes(),
239 token.as_bytes(),
240 ) {
241 Some(node)
242 } else {
243 None
244 }
245 } else {
246 None
247 },
248 )
249 },
250 )
251 .await
252 }
253
254 pub async fn by_location_uuid_with_pagination(
255 database: &crate::database::Database,
256 location_uuid: uuid::Uuid,
257 page: i64,
258 per_page: i64,
259 search: Option<&str>,
260 ) -> Result<super::Pagination<Self>, crate::database::DatabaseError> {
261 let offset = (page - 1) * per_page;
262
263 let rows = sqlx::query(sqlx::AssertSqlSafe(format!(
264 r#"
265 SELECT {}, COUNT(*) OVER() AS total_count
266 FROM nodes
267 JOIN locations ON locations.uuid = nodes.location_uuid
268 WHERE nodes.location_uuid = $1 AND ($2 IS NULL OR nodes.name ILIKE '%' || $2 || '%')
269 ORDER BY nodes.created
270 LIMIT $3 OFFSET $4
271 "#,
272 Self::columns_sql(None)
273 )))
274 .bind(location_uuid)
275 .bind(search)
276 .bind(per_page)
277 .bind(offset)
278 .fetch_all(database.read())
279 .await?;
280
281 Ok(super::Pagination {
282 total: rows
283 .first()
284 .map_or(Ok(0), |row| row.try_get("total_count"))?,
285 per_page,
286 page,
287 data: rows
288 .into_iter()
289 .map(|row| Self::map(None, &row))
290 .try_collect_vec()?,
291 })
292 }
293
294 pub async fn by_backup_configuration_uuid_with_pagination(
295 database: &crate::database::Database,
296 backup_configuration_uuid: uuid::Uuid,
297 page: i64,
298 per_page: i64,
299 search: Option<&str>,
300 ) -> Result<super::Pagination<Self>, crate::database::DatabaseError> {
301 let offset = (page - 1) * per_page;
302
303 let rows = sqlx::query(sqlx::AssertSqlSafe(format!(
304 r#"
305 SELECT {}, COUNT(*) OVER() AS total_count
306 FROM nodes
307 JOIN locations ON locations.uuid = nodes.location_uuid
308 WHERE nodes.backup_configuration_uuid = $1 AND ($2 IS NULL OR nodes.name ILIKE '%' || $2 || '%')
309 ORDER BY nodes.created
310 LIMIT $3 OFFSET $4
311 "#,
312 Self::columns_sql(None)
313 )))
314 .bind(backup_configuration_uuid)
315 .bind(search)
316 .bind(per_page)
317 .bind(offset)
318 .fetch_all(database.read())
319 .await?;
320
321 Ok(super::Pagination {
322 total: rows
323 .first()
324 .map_or(Ok(0), |row| row.try_get("total_count"))?,
325 per_page,
326 page,
327 data: rows
328 .into_iter()
329 .map(|row| Self::map(None, &row))
330 .try_collect_vec()?,
331 })
332 }
333
334 pub async fn all_with_pagination(
335 database: &crate::database::Database,
336 page: i64,
337 per_page: i64,
338 search: Option<&str>,
339 ) -> Result<super::Pagination<Self>, crate::database::DatabaseError> {
340 let offset = (page - 1) * per_page;
341
342 let rows = sqlx::query(sqlx::AssertSqlSafe(format!(
343 r#"
344 SELECT {}, COUNT(*) OVER() AS total_count
345 FROM nodes
346 JOIN locations ON locations.uuid = nodes.location_uuid
347 WHERE $1 IS NULL OR nodes.name ILIKE '%' || $1 || '%'
348 ORDER BY nodes.created
349 LIMIT $2 OFFSET $3
350 "#,
351 Self::columns_sql(None)
352 )))
353 .bind(search)
354 .bind(per_page)
355 .bind(offset)
356 .fetch_all(database.read())
357 .await?;
358
359 Ok(super::Pagination {
360 total: rows
361 .first()
362 .map_or(Ok(0), |row| row.try_get("total_count"))?,
363 per_page,
364 page,
365 data: rows
366 .into_iter()
367 .map(|row| Self::map(None, &row))
368 .try_collect_vec()?,
369 })
370 }
371
372 pub async fn by_location_uuids_most_eligible(
373 database: &crate::database::Database,
374 location_uuids: &[uuid::Uuid],
375 limits: super::server::AdminApiServerLimits,
376 allow_overallocation: bool,
377 suspension_penalty: f64,
378 randomness: f64,
379 ) -> Result<Vec<Self>, crate::database::DatabaseError> {
380 let rows = sqlx::query(sqlx::AssertSqlSafe(format!(
381 r#"
382 WITH server_usage AS (
383 SELECT
384 node_uuid,
385 COALESCE(SUM((memory + memory_overhead) * CASE WHEN suspended THEN $5 ELSE 1.0 END), 0)::BIGINT AS used_memory,
386 COALESCE(SUM(disk * CASE WHEN suspended THEN $5 ELSE 1.0 END), 0)::BIGINT AS used_disk
387 FROM servers
388 GROUP BY node_uuid
389 )
390 SELECT {}, COALESCE(u.used_memory, 0) AS used_memory, COALESCE(u.used_disk, 0) AS used_disk
391 FROM nodes
392 JOIN locations ON locations.uuid = nodes.location_uuid
393 LEFT JOIN server_usage u ON nodes.uuid = u.node_uuid
394 WHERE nodes.location_uuid = ANY($1)
395 AND nodes.deployment_enabled
396 AND (
397 $4 OR (
398 COALESCE(u.used_memory, 0) + $2 <= nodes.memory
399 AND COALESCE(u.used_disk, 0) + $3 <= nodes.disk
400 )
401 )
402 ORDER BY
403 (
404 GREATEST(COALESCE(u.used_memory, 0) + $2 - nodes.memory, 0) +
405 GREATEST(COALESCE(u.used_disk, 0) + $3 - nodes.disk, 0)
406 ),
407 GREATEST(
408 (COALESCE(u.used_memory, 0) + $2)::FLOAT / NULLIF(nodes.memory, 0),
409 (COALESCE(u.used_disk, 0) + $3)::FLOAT / NULLIF(nodes.disk, 0)
410 )
411 "#,
412 Self::columns_sql(None),
413 )))
414 .bind(location_uuids)
415 .bind(limits.memory)
416 .bind(limits.disk)
417 .bind(allow_overallocation)
418 .bind(suspension_penalty)
419 .fetch_all(database.read())
420 .await?;
421
422 let nodes = rows
423 .into_iter()
424 .map(|row| {
425 Ok((
426 Self::map(None, &row)?,
427 row.try_get::<i64, _>("used_memory")?,
428 row.try_get::<i64, _>("used_disk")?,
429 ))
430 })
431 .collect::<Result<Vec<_>, crate::database::DatabaseError>>()?;
432
433 if randomness > 0.0 {
434 let mut rng = rand::rng();
435
436 let mut keyed = nodes
437 .into_iter()
438 .map(|(node, used_memory, used_disk)| {
439 let free_ratio = f64::min(
440 1.0 - (used_memory + limits.memory) as f64 / node.memory.max(1) as f64,
441 1.0 - (used_disk + limits.disk) as f64 / node.disk.max(1) as f64,
442 )
443 .clamp(0.0001, 1.0);
444
445 let weight = free_ratio.powf(1.0 / randomness);
446 let key = rng.random::<f64>().powf(1.0 / weight);
447
448 (key, node)
449 })
450 .collect::<Vec<_>>();
451
452 keyed.sort_by(|(a, _), (b, _)| b.total_cmp(a));
453
454 return Ok(keyed.into_iter().map(|(_, node)| node).collect());
455 }
456
457 Ok(nodes.into_iter().map(|(node, _, _)| node).collect())
458 }
459
460 pub async fn by_name(
461 database: &crate::database::Database,
462 name: &str,
463 ) -> Result<Option<Self>, crate::database::DatabaseError> {
464 let row = sqlx::query(sqlx::AssertSqlSafe(format!(
465 r#"
466 SELECT {}
467 FROM nodes
468 JOIN locations ON locations.uuid = nodes.location_uuid
469 WHERE nodes.name = $1
470 "#,
471 Self::columns_sql(None)
472 )))
473 .bind(name)
474 .fetch_optional(database.read())
475 .await?;
476
477 row.try_map(|row| Self::map(None, &row))
478 }
479
480 pub async fn count_by_location_uuid(
481 database: &crate::database::Database,
482 location_uuid: uuid::Uuid,
483 ) -> Result<i64, sqlx::Error> {
484 sqlx::query_scalar(
485 r#"
486 SELECT COUNT(*)
487 FROM nodes
488 WHERE nodes.location_uuid = $1
489 "#,
490 )
491 .bind(location_uuid)
492 .fetch_one(database.read())
493 .await
494 }
495
496 pub async fn find_deployment_blocker(
497 database: &crate::database::Database,
498 location_uuids: &[uuid::Uuid],
499 limits: super::server::AdminApiServerLimits,
500 allow_overallocation: bool,
501 suspension_penalty: f64,
502 ) -> Result<Option<NodeDeploymentBlocker>, crate::database::DatabaseError> {
503 let row = sqlx::query(
504 r#"
505 WITH server_usage AS (
506 SELECT
507 node_uuid,
508 COALESCE(SUM((memory + memory_overhead) * CASE WHEN suspended THEN $5 ELSE 1.0 END), 0)::BIGINT AS used_memory,
509 COALESCE(SUM(disk * CASE WHEN suspended THEN $5 ELSE 1.0 END), 0)::BIGINT AS used_disk
510 FROM servers
511 GROUP BY node_uuid
512 )
513 SELECT
514 COUNT(*) AS total,
515 COUNT(*) FILTER (WHERE nodes.deployment_enabled) AS deployable,
516 COUNT(*) FILTER (
517 WHERE nodes.deployment_enabled
518 AND ($4 OR COALESCE(u.used_memory, 0) + $2 <= nodes.memory)
519 ) AS memory_ok,
520 COUNT(*) FILTER (
521 WHERE nodes.deployment_enabled
522 AND ($4 OR COALESCE(u.used_disk, 0) + $3 <= nodes.disk)
523 ) AS disk_ok,
524 COUNT(*) FILTER (
525 WHERE nodes.deployment_enabled
526 AND ($4 OR COALESCE(u.used_memory, 0) + $2 <= nodes.memory)
527 AND ($4 OR COALESCE(u.used_disk, 0) + $3 <= nodes.disk)
528 ) AS resource_ok
529 FROM nodes
530 LEFT JOIN server_usage u ON nodes.uuid = u.node_uuid
531 WHERE nodes.location_uuid = ANY($1)
532 "#,
533 )
534 .bind(location_uuids)
535 .bind(limits.memory)
536 .bind(limits.disk)
537 .bind(allow_overallocation)
538 .bind(suspension_penalty)
539 .fetch_one(database.read())
540 .await?;
541
542 let total: i64 = row.try_get("total")?;
543 let deployable: i64 = row.try_get("deployable")?;
544 let memory_ok: i64 = row.try_get("memory_ok")?;
545 let disk_ok: i64 = row.try_get("disk_ok")?;
546 let resource_ok: i64 = row.try_get("resource_ok")?;
547
548 Ok(Some(if total == 0 {
549 NodeDeploymentBlocker::NoNodes
550 } else if deployable == 0 {
551 NodeDeploymentBlocker::DeploymentDisabled
552 } else if resource_ok > 0 {
553 return Ok(None);
554 } else if memory_ok == 0 && disk_ok == 0 {
555 NodeDeploymentBlocker::InsufficientResources
556 } else if memory_ok == 0 {
557 NodeDeploymentBlocker::InsufficientMemory
558 } else if disk_ok == 0 {
559 NodeDeploymentBlocker::InsufficientDisk
560 } else {
561 NodeDeploymentBlocker::ResourcesSplitAcrossNodes
562 }))
563 }
564
565 pub async fn fetch_configuration(
569 &self,
570 database: &crate::database::Database,
571 ) -> Result<wings_api::Config, anyhow::Error> {
572 database
573 .cache
574 .cached(
575 &format!("node::{}::configuration", self.uuid),
576 120,
577 || async {
578 Ok::<_, anyhow::Error>(
579 self.api_client(database).await?.get_system_config().await?,
580 )
581 },
582 )
583 .await
584 }
585
586 pub async fn update_configuration(
590 &self,
591 database: &crate::database::Database,
592 config_patch: &serde_json::Value,
593 ) -> Result<bool, anyhow::Error> {
594 let response = self
595 .api_client(database)
596 .await?
597 .post_update(config_patch)
598 .await?;
599 if !response.applied {
600 return Ok(false);
601 }
602
603 database
604 .cache
605 .invalidate(&format!("node::{}::configuration", self.uuid))
606 .await?;
607
608 Ok(true)
609 }
610
611 pub async fn fetch_server_resources(
615 &self,
616 database: &crate::database::Database,
617 ) -> Result<HashMap<uuid::Uuid, wings_api::ResourceUsage>, anyhow::Error> {
618 database
619 .cache
620 .cached(
621 &format!("node::{}::server_resources", self.uuid),
622 15,
623 || async {
624 let resources = self
625 .api_client(database)
626 .await?
627 .get_servers_utilization()
628 .await?;
629
630 Ok::<_, anyhow::Error>(resources.into_iter().collect())
631 },
632 )
633 .await
634 }
635
636 #[inline]
637 pub fn generate_token() -> (String, String) {
638 let token_id = rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 16);
639 let token = rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 64);
640
641 (token_id, token)
642 }
643
644 pub async fn reset_token(
645 &self,
646 state: &crate::State,
647 ) -> Result<(String, String), anyhow::Error> {
648 let (token_id, token) = Self::generate_token();
649
650 sqlx::query(
651 r#"
652 UPDATE nodes
653 SET token_id = $2, token = $3
654 WHERE nodes.uuid = $1
655 "#,
656 )
657 .bind(self.uuid)
658 .bind(&token_id)
659 .bind(state.database.encrypt(token.clone()).await?)
660 .execute(state.database.write())
661 .await?;
662
663 Self::get_event_emitter().emit(
664 state.clone(),
665 NodeEvent::TokenReset {
666 node: Box::new(self.clone()),
667 token_id: token_id.clone(),
668 token: token.clone(),
669 },
670 );
671
672 Ok((token_id, token))
673 }
674
675 #[inline]
676 pub fn is_all_in_one_node(&self) -> bool {
677 self.uuid == Self::AIO_NODE_UUID
678 }
679
680 #[inline]
681 pub fn url(&self, path: &str) -> reqwest::Url {
682 let mut url = self.url.clone();
683 url.path_segments_mut()
684 .unwrap()
685 .extend(path.trim_start_matches('/').split('/'));
686 url
687 }
688
689 #[inline]
690 pub async fn public_url(
691 &self,
692 state: &crate::State,
693 path: &str,
694 ) -> Result<reqwest::Url, anyhow::Error> {
695 let mut url = if self.is_all_in_one_node() {
696 let mut url = state
697 .settings
698 .get_as(|s| reqwest::Url::parse(&s.app.url))
699 .await??;
700 url.path_segments_mut()
701 .unwrap()
702 .extend(&["wings-proxy", &self.uuid.to_compact_string()]);
703 url
704 } else {
705 self.public_url.clone().unwrap_or(self.url.clone())
706 };
707
708 url.path_segments_mut()
709 .unwrap()
710 .extend(path.trim_start_matches('/').split('/'));
711
712 Ok(url)
713 }
714
715 #[inline]
716 pub async fn api_client(
717 &self,
718 database: &crate::database::Database,
719 ) -> Result<wings_api::client::WingsClient, anyhow::Error> {
720 Ok(wings_api::client::WingsClient::new(
721 self.url.to_string(),
722 database.decrypt(self.token.to_vec()).await?.into(),
723 ))
724 }
725
726 #[inline]
727 pub fn create_jwt<T: Serialize>(
728 &self,
729 database: &crate::database::Database,
730 jwt: &crate::jwt::Jwt,
731 payload: &T,
732 ) -> Result<String, anyhow::Error> {
733 Ok(jwt.create_custom(database.blocking_decrypt(&self.token)?.as_bytes(), payload)?)
734 }
735}
736
737#[async_trait::async_trait]
738impl IntoAdminApiObject for Node {
739 type AdminApiObject = AdminApiNode;
740 type ExtraArgs<'a> = ();
741
742 async fn into_admin_api_object<'a>(
743 self,
744 state: &crate::State,
745 _args: Self::ExtraArgs<'a>,
746 ) -> Result<Self::AdminApiObject, crate::database::DatabaseError> {
747 let api_object = AdminApiNode::init_hooks(&self, state).await?;
748
749 let public_url = if self.is_all_in_one_node() {
750 Some(self.public_url(state, "/").await?.to_string())
751 } else {
752 self.public_url.map(|url| url.to_string())
753 };
754
755 let (location, backup_configuration) =
756 tokio::join!(self.location.into_admin_api_object(state, ()), async {
757 if let Some(backup_configuration) = self.backup_configuration {
758 if let Ok(backup_configuration) =
759 backup_configuration.fetch_cached(&state.database).await
760 {
761 backup_configuration
762 .into_admin_api_object(state, ())
763 .await
764 .ok()
765 } else {
766 None
767 }
768 } else {
769 None
770 }
771 });
772
773 let api_object = finish_extendible!(
774 AdminApiNode {
775 uuid: self.uuid,
776 location: location?,
777 backup_configuration,
778 name: self.name,
779 description: self.description,
780 deployment_enabled: self.deployment_enabled,
781 maintenance_enabled: self.maintenance_enabled,
782 public_url,
783 url: self.url.to_string(),
784 sftp_host: self.sftp_host,
785 sftp_port: self.sftp_port,
786 memory: self.memory,
787 disk: self.disk,
788 created: self.created.and_utc(),
789 },
790 api_object,
791 state
792 )?;
793
794 Ok(api_object)
795 }
796}
797
798#[async_trait::async_trait]
799impl ByUuid for Node {
800 async fn by_uuid(
801 database: &crate::database::Database,
802 uuid: uuid::Uuid,
803 ) -> Result<Self, crate::database::DatabaseError> {
804 let row = sqlx::query(sqlx::AssertSqlSafe(format!(
805 r#"
806 SELECT {}, {}
807 FROM nodes
808 JOIN locations ON locations.uuid = nodes.location_uuid
809 WHERE nodes.uuid = $1
810 "#,
811 Self::columns_sql(None),
812 super::location::Location::columns_sql(Some("location_")),
813 )))
814 .bind(uuid)
815 .fetch_one(database.read())
816 .await?;
817
818 Self::map(None, &row)
819 }
820
821 async fn by_uuid_with_transaction(
822 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
823 uuid: uuid::Uuid,
824 ) -> Result<Self, crate::database::DatabaseError> {
825 let row = sqlx::query(sqlx::AssertSqlSafe(format!(
826 r#"
827 SELECT {}, {}
828 FROM nodes
829 JOIN locations ON locations.uuid = nodes.location_uuid
830 WHERE nodes.uuid = $1
831 "#,
832 Self::columns_sql(None),
833 super::location::Location::columns_sql(Some("location_")),
834 )))
835 .bind(uuid)
836 .fetch_one(&mut **transaction)
837 .await?;
838
839 Self::map(None, &row)
840 }
841}
842
843#[derive(ToSchema, Deserialize, Validate)]
844pub struct CreateNodeOptions {
845 #[garde(skip)]
846 pub location_uuid: uuid::Uuid,
847 #[garde(skip)]
848 pub backup_configuration_uuid: Option<uuid::Uuid>,
849 #[garde(length(chars, min = 1, max = 255))]
850 #[schema(min_length = 1, max_length = 255)]
851 pub name: compact_str::CompactString,
852 #[garde(length(chars, min = 1, max = 1024))]
853 #[schema(min_length = 1, max_length = 1024)]
854 pub description: Option<compact_str::CompactString>,
855 #[garde(skip)]
856 pub deployment_enabled: bool,
857 #[garde(skip)]
858 pub maintenance_enabled: bool,
859 #[garde(length(chars, min = 3, max = 255), url)]
860 #[schema(min_length = 3, max_length = 255, format = "uri")]
861 pub public_url: Option<compact_str::CompactString>,
862 #[garde(length(chars, min = 3, max = 255), url)]
863 #[schema(min_length = 3, max_length = 255, format = "uri")]
864 pub url: compact_str::CompactString,
865 #[garde(length(chars, min = 3, max = 255))]
866 #[schema(min_length = 3, max_length = 255)]
867 pub sftp_host: Option<compact_str::CompactString>,
868 #[garde(range(min = 1))]
869 #[schema(minimum = 1)]
870 pub sftp_port: u16,
871 #[garde(range(min = 1))]
872 #[schema(minimum = 1)]
873 pub memory: i64,
874 #[garde(range(min = 1))]
875 #[schema(minimum = 1)]
876 pub disk: i64,
877}
878
879#[async_trait::async_trait]
880impl CreatableModel for Node {
881 type CreateOptions<'a> = CreateNodeOptions;
882 type CreateResult = Self;
883
884 fn get_create_handlers() -> &'static LazyLock<CreateListenerList<Self>> {
885 static CREATE_LISTENERS: LazyLock<CreateListenerList<Node>> =
886 LazyLock::new(|| Arc::new(ModelHandlerList::default()));
887
888 &CREATE_LISTENERS
889 }
890
891 async fn create_with_transaction(
892 state: &crate::State,
893 mut options: Self::CreateOptions<'_>,
894 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
895 ) -> Result<Self, crate::database::DatabaseError> {
896 options.validate()?;
897
898 if let Some(backup_configuration_uuid) = &options.backup_configuration_uuid {
899 super::backup_configuration::BackupConfiguration::by_uuid_optional(
900 &state.database,
901 *backup_configuration_uuid,
902 )
903 .await?
904 .ok_or(crate::database::InvalidRelationError(
905 "backup_configuration",
906 ))?;
907 }
908
909 let mut query_builder = InsertQueryBuilder::new("nodes");
910
911 Self::run_create_handlers(&mut options, &mut query_builder, state, transaction).await?;
912
913 let (token_id, token) = Self::generate_token();
914
915 query_builder
916 .set("location_uuid", options.location_uuid)
917 .set(
918 "backup_configuration_uuid",
919 options.backup_configuration_uuid,
920 )
921 .set("name", &options.name)
922 .set("description", &options.description)
923 .set("deployment_enabled", options.deployment_enabled)
924 .set("maintenance_enabled", options.maintenance_enabled)
925 .set("public_url", &options.public_url)
926 .set("url", &options.url)
927 .set("sftp_host", &options.sftp_host)
928 .set("sftp_port", options.sftp_port as i32)
929 .set("memory", options.memory)
930 .set("disk", options.disk)
931 .set("token_id", token_id.clone())
932 .set("token", state.database.encrypt(token.clone()).await?);
933
934 let row = query_builder
935 .returning("uuid")
936 .fetch_one(&mut **transaction)
937 .await?;
938 let uuid: uuid::Uuid = row.try_get("uuid")?;
939
940 let mut result = Self::by_uuid_with_transaction(transaction, uuid).await?;
941
942 Self::run_after_create_handlers(&mut result, &options, state, transaction).await?;
943
944 Ok(result)
945 }
946}
947
948#[derive(ToSchema, Serialize, Deserialize, Validate, Clone, Default)]
949pub struct UpdateNodeOptions {
950 #[garde(skip)]
951 pub location_uuid: Option<uuid::Uuid>,
952 #[serde(
953 default,
954 skip_serializing_if = "Option::is_none",
955 with = "::serde_with::rust::double_option"
956 )]
957 #[garde(skip)]
958 pub backup_configuration_uuid: Option<Option<uuid::Uuid>>,
959 #[garde(length(chars, min = 1, max = 255))]
960 #[schema(min_length = 1, max_length = 255)]
961 pub name: Option<compact_str::CompactString>,
962 #[garde(length(chars, min = 1, max = 1024))]
963 #[schema(min_length = 1, max_length = 1024)]
964 #[serde(
965 default,
966 skip_serializing_if = "Option::is_none",
967 with = "::serde_with::rust::double_option"
968 )]
969 pub description: Option<Option<compact_str::CompactString>>,
970 #[garde(skip)]
971 pub deployment_enabled: Option<bool>,
972 #[garde(skip)]
973 pub maintenance_enabled: Option<bool>,
974 #[garde(length(chars, min = 3, max = 255), url)]
975 #[schema(min_length = 3, max_length = 255, format = "uri")]
976 #[serde(
977 default,
978 skip_serializing_if = "Option::is_none",
979 with = "::serde_with::rust::double_option"
980 )]
981 pub public_url: Option<Option<compact_str::CompactString>>,
982 #[garde(length(chars, min = 3, max = 255), url)]
983 #[schema(min_length = 3, max_length = 255, format = "uri")]
984 pub url: Option<compact_str::CompactString>,
985 #[garde(length(chars, min = 3, max = 255))]
986 #[schema(min_length = 3, max_length = 255)]
987 #[serde(
988 default,
989 skip_serializing_if = "Option::is_none",
990 with = "::serde_with::rust::double_option"
991 )]
992 pub sftp_host: Option<Option<compact_str::CompactString>>,
993 #[garde(range(min = 1))]
994 #[schema(minimum = 1)]
995 pub sftp_port: Option<u16>,
996 #[garde(range(min = 1))]
997 #[schema(minimum = 1)]
998 pub memory: Option<i64>,
999 #[garde(range(min = 1))]
1000 #[schema(minimum = 1)]
1001 pub disk: Option<i64>,
1002}
1003
1004#[async_trait::async_trait]
1005impl UpdatableModel for Node {
1006 type UpdateOptions = UpdateNodeOptions;
1007
1008 fn get_update_handlers() -> &'static LazyLock<UpdateHandlerList<Self>> {
1009 static UPDATE_LISTENERS: LazyLock<UpdateHandlerList<Node>> =
1010 LazyLock::new(|| Arc::new(ModelHandlerList::default()));
1011
1012 &UPDATE_LISTENERS
1013 }
1014
1015 async fn update_with_transaction(
1016 &mut self,
1017 state: &crate::State,
1018 mut options: Self::UpdateOptions,
1019 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1020 ) -> Result<(), crate::database::DatabaseError> {
1021 options.validate()?;
1022
1023 let location = if let Some(location_uuid) = options.location_uuid {
1024 Some(
1025 super::location::Location::by_uuid_optional(&state.database, location_uuid)
1026 .await?
1027 .ok_or(crate::database::InvalidRelationError("location"))?,
1028 )
1029 } else {
1030 None
1031 };
1032
1033 let backup_configuration =
1034 if let Some(backup_configuration_uuid) = &options.backup_configuration_uuid {
1035 match backup_configuration_uuid {
1036 Some(uuid) => {
1037 super::backup_configuration::BackupConfiguration::by_uuid_optional(
1038 &state.database,
1039 *uuid,
1040 )
1041 .await?
1042 .ok_or(crate::database::InvalidRelationError(
1043 "backup_configuration",
1044 ))?;
1045
1046 Some(Some(
1047 super::backup_configuration::BackupConfiguration::get_fetchable(*uuid),
1048 ))
1049 }
1050 None => Some(None),
1051 }
1052 } else {
1053 None
1054 };
1055
1056 let mut query_builder = UpdateQueryBuilder::new("nodes");
1057
1058 self.run_update_handlers(&mut options, &mut query_builder, state, transaction)
1059 .await?;
1060
1061 query_builder
1062 .set("location_uuid", options.location_uuid.as_ref())
1063 .set(
1064 "backup_configuration_uuid",
1065 options
1066 .backup_configuration_uuid
1067 .as_ref()
1068 .map(|u| u.as_ref()),
1069 )
1070 .set("name", options.name.as_ref())
1071 .set(
1072 "description",
1073 options.description.as_ref().map(|d| d.as_ref()),
1074 )
1075 .set("deployment_enabled", options.deployment_enabled)
1076 .set("maintenance_enabled", options.maintenance_enabled)
1077 .set(
1078 "public_url",
1079 options.public_url.as_ref().map(|u| u.as_ref()),
1080 )
1081 .set("url", options.url.as_ref())
1082 .set("sftp_host", options.sftp_host.as_ref().map(|h| h.as_ref()))
1083 .set("sftp_port", options.sftp_port.as_ref().map(|p| *p as i32))
1084 .set("memory", options.memory.as_ref())
1085 .set("disk", options.disk.as_ref())
1086 .where_eq("uuid", self.uuid);
1087
1088 query_builder.execute(&mut **transaction).await?;
1089
1090 if let Some(location) = location {
1091 self.location = location;
1092 }
1093 if let Some(backup_configuration) = backup_configuration {
1094 self.backup_configuration = backup_configuration;
1095 }
1096 if let Some(name) = options.name {
1097 self.name = name;
1098 }
1099 if let Some(description) = options.description {
1100 self.description = description;
1101 }
1102 if let Some(deployment_enabled) = options.deployment_enabled {
1103 self.deployment_enabled = deployment_enabled;
1104 }
1105 if let Some(maintenance_enabled) = options.maintenance_enabled {
1106 self.maintenance_enabled = maintenance_enabled;
1107 }
1108 if let Some(public_url) = options.public_url {
1109 self.public_url = public_url
1110 .try_map(|url| url.parse())
1111 .map_err(anyhow::Error::new)?;
1112 }
1113 if let Some(url) = options.url {
1114 self.url = url.parse().map_err(anyhow::Error::new)?;
1115 }
1116 if let Some(sftp_host) = options.sftp_host {
1117 self.sftp_host = sftp_host;
1118 }
1119 if let Some(sftp_port) = options.sftp_port {
1120 self.sftp_port = sftp_port as i32;
1121 }
1122 if let Some(memory) = options.memory {
1123 self.memory = memory;
1124 }
1125 if let Some(disk) = options.disk {
1126 self.disk = disk;
1127 }
1128
1129 self.run_after_update_handlers(state, transaction).await?;
1130
1131 Ok(())
1132 }
1133}
1134
1135#[async_trait::async_trait]
1136impl DeletableModel for Node {
1137 type DeleteOptions = ();
1138
1139 fn get_delete_handlers() -> &'static LazyLock<DeleteHandlerList<Self>> {
1140 static DELETE_LISTENERS: LazyLock<DeleteHandlerList<Node>> =
1141 LazyLock::new(|| Arc::new(ModelHandlerList::default()));
1142
1143 &DELETE_LISTENERS
1144 }
1145
1146 async fn delete_with_transaction(
1147 &self,
1148 state: &crate::State,
1149 options: Self::DeleteOptions,
1150 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1151 ) -> Result<(), anyhow::Error> {
1152 if self.is_all_in_one_node() && state.container_type.is_all_in_one() {
1153 return Err(anyhow::anyhow!("The AIO node cannot be deleted"));
1154 }
1155
1156 self.run_delete_handlers(&options, state, transaction)
1157 .await?;
1158
1159 sqlx::query(
1160 r#"
1161 DELETE FROM nodes
1162 WHERE nodes.uuid = $1
1163 "#,
1164 )
1165 .bind(self.uuid)
1166 .execute(&mut **transaction)
1167 .await?;
1168
1169 self.run_after_delete_handlers(&options, state, transaction)
1170 .await?;
1171
1172 Ok(())
1173 }
1174}
1175
1176#[derive(Validate)]
1177pub struct DuplicateNodeOptions {
1178 #[garde(length(chars, min = 1, max = 255))]
1179 pub name: compact_str::CompactString,
1180}
1181
1182#[async_trait::async_trait]
1183impl DuplicableModel for Node {
1184 type DuplicateOptions<'a> = DuplicateNodeOptions;
1185
1186 fn get_duplicate_handlers() -> &'static LazyLock<DuplicateHandlerList<Self>> {
1187 static DUPLICATE_LISTENERS: LazyLock<DuplicateHandlerList<Node>> =
1188 LazyLock::new(|| Arc::new(ModelHandlerList::default()));
1189
1190 &DUPLICATE_LISTENERS
1191 }
1192
1193 async fn duplicate_with_transaction(
1194 &self,
1195 state: &crate::State,
1196 options: Self::DuplicateOptions<'_>,
1197 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1198 ) -> Result<Self, crate::database::DatabaseError> {
1199 options.validate()?;
1200
1201 self.run_duplicate_handlers(&options, state, transaction)
1202 .await?;
1203
1204 let mut query_builder = InsertQueryBuilder::new("nodes");
1205
1206 let (token_id, token) = Self::generate_token();
1207
1208 query_builder
1209 .set("location_uuid", self.location.uuid)
1210 .set(
1211 "backup_configuration_uuid",
1212 self.backup_configuration.as_ref().map(|c| c.uuid),
1213 )
1214 .set("name", &options.name)
1215 .set("description", &self.description)
1216 .set("deployment_enabled", self.deployment_enabled)
1217 .set("maintenance_enabled", self.maintenance_enabled)
1218 .set("public_url", self.public_url.as_ref().map(|u| u.as_str()))
1219 .set("url", self.url.as_str())
1220 .set("sftp_host", &self.sftp_host)
1221 .set("sftp_port", self.sftp_port)
1222 .set("memory", self.memory)
1223 .set("disk", self.disk)
1224 .set("token_id", token_id)
1225 .set("token", state.database.encrypt(token).await?);
1226
1227 let row = query_builder
1228 .returning("uuid")
1229 .fetch_one(&mut **transaction)
1230 .await?;
1231 let uuid: uuid::Uuid = row.try_get("uuid")?;
1232
1233 let mut node = Self::by_uuid_with_transaction(transaction, uuid).await?;
1234
1235 sqlx::query!(
1236 "INSERT INTO node_mounts (node_uuid, mount_uuid)
1237 SELECT $1, node_mounts.mount_uuid
1238 FROM node_mounts
1239 WHERE node_mounts.node_uuid = $2",
1240 node.uuid,
1241 self.uuid,
1242 )
1243 .execute(&mut **transaction)
1244 .await?;
1245
1246 self.run_after_duplicate_handlers(&mut node, &options, state, transaction)
1247 .await?;
1248
1249 Ok(node)
1250 }
1251}
1252
1253#[schema_extension_derive::extendible]
1254#[init_args(Node, crate::State)]
1255#[hook_args(crate::State)]
1256#[derive(ToSchema, Serialize)]
1257#[schema(title = "Node")]
1258pub struct AdminApiNode {
1259 pub uuid: uuid::Uuid,
1260 pub location: super::location::AdminApiLocation,
1261 pub backup_configuration: Option<super::backup_configuration::AdminApiBackupConfiguration>,
1262
1263 pub name: compact_str::CompactString,
1264 pub description: Option<compact_str::CompactString>,
1265
1266 pub deployment_enabled: bool,
1267 pub maintenance_enabled: bool,
1268
1269 #[schema(format = "uri")]
1270 pub public_url: Option<String>,
1271 #[schema(format = "uri")]
1272 pub url: String,
1273 pub sftp_host: Option<compact_str::CompactString>,
1274 pub sftp_port: i32,
1275
1276 pub memory: i64,
1277 pub disk: i64,
1278
1279 pub created: chrono::DateTime<chrono::Utc>,
1280}