1use crate::{
2 crypt::EncryptedString,
3 models::{
4 CreatableModel, CreateListenerList, InsertQueryBuilder, UpdatableModel, UpdateHandlerList,
5 UpdateQueryBuilder,
6 },
7 prelude::*,
8};
9use compact_str::ToCompactString;
10use futures_util::StreamExt;
11use garde::Validate;
12use rand::{RngExt, distr::SampleString};
13use serde::{Deserialize, Serialize};
14use sqlx::{Row, postgres::PgRow};
15use std::{
16 collections::{BTreeMap, HashMap},
17 sync::{Arc, LazyLock},
18};
19use utoipa::ToSchema;
20
21mod allowed_sources;
22mod enrollment;
23mod events;
24pub use allowed_sources::NodeAllowedSources;
25pub use enrollment::NodeEnrollment;
26pub use events::NodeEvent;
27
28pub type GetNode = crate::extract::ConsumingExtension<Node>;
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum NodeDeploymentBlocker {
32 NoNodes,
34 DeploymentDisabled,
36 MaintenanceEnabled,
38 InsufficientMemory,
40 InsufficientDisk,
42 InsufficientResources,
44 ResourcesSplitAcrossNodes,
47}
48
49impl NodeDeploymentBlocker {
50 pub fn message(self) -> &'static str {
51 match self {
52 Self::NoNodes => "no nodes exist in the selected location(s)",
53 Self::DeploymentDisabled => {
54 "deployment is disabled on every node in the selected location(s)"
55 }
56 Self::MaintenanceEnabled => {
57 "every deployable node in the selected location(s) is under maintenance"
58 }
59 Self::InsufficientMemory => {
60 "no node in the selected location(s) has enough unallocated memory"
61 }
62 Self::InsufficientDisk => {
63 "no node in the selected location(s) has enough unallocated disk"
64 }
65 Self::InsufficientResources => {
66 "no node in the selected location(s) has enough unallocated memory or disk"
67 }
68 Self::ResourcesSplitAcrossNodes => {
69 "no single node in the selected location(s) has enough unallocated memory and disk at the same time"
70 }
71 }
72 }
73}
74
75#[derive(Default)]
77pub struct PermissionsInvalidation {
78 pub users: Vec<uuid::Uuid>,
80 pub servers: Option<wings_api::ServerSelector>,
82 pub revoke_users: Vec<uuid::Uuid>,
84 pub revoke_credentials: Vec<uuid::Uuid>,
86}
87
88impl PermissionsInvalidation {
89 pub fn is_empty(&self) -> bool {
90 self.users.is_empty()
91 && self.servers.is_none()
92 && self.revoke_users.is_empty()
93 && self.revoke_credentials.is_empty()
94 }
95}
96
97#[derive(Serialize, Deserialize, Clone)]
98pub struct Node {
99 pub uuid: uuid::Uuid,
100 pub location: super::location::Location,
101 pub backup_configuration: Option<Fetchable<super::backup_configuration::BackupConfiguration>>,
102
103 pub name: compact_str::CompactString,
104 pub description: Option<compact_str::CompactString>,
105
106 pub deployment_enabled: bool,
107 pub maintenance_enabled: bool,
108
109 pub public_url: Option<reqwest::Url>,
110 pub url: reqwest::Url,
111 pub sftp_host: Option<compact_str::CompactString>,
112 pub sftp_port: i32,
113
114 pub memory: i64,
115 pub disk: i64,
116
117 pub token_id: compact_str::CompactString,
118 pub token: EncryptedString,
119
120 pub created: chrono::NaiveDateTime,
121
122 extension_data: super::ModelExtensionData,
123}
124
125impl BaseModel for Node {
126 const NAME: &'static str = "node";
127
128 fn get_extension_list() -> &'static super::ModelExtensionList {
129 static EXTENSIONS: LazyLock<super::ModelExtensionList> =
130 LazyLock::new(|| parking_lot::RwLock::new(Vec::new()));
131
132 &EXTENSIONS
133 }
134
135 fn get_extension_data(&self) -> &super::ModelExtensionData {
136 &self.extension_data
137 }
138
139 #[inline]
140 fn base_columns(prefix: Option<&str>) -> BTreeMap<&'static str, compact_str::CompactString> {
141 let prefix = prefix.unwrap_or_default();
142
143 let mut columns = BTreeMap::from([
144 ("nodes.uuid", compact_str::format_compact!("{prefix}uuid")),
145 (
146 "nodes.backup_configuration_uuid",
147 compact_str::format_compact!("{prefix}node_backup_configuration_uuid"),
148 ),
149 ("nodes.name", compact_str::format_compact!("{prefix}name")),
150 (
151 "nodes.description",
152 compact_str::format_compact!("{prefix}description"),
153 ),
154 (
155 "nodes.deployment_enabled",
156 compact_str::format_compact!("{prefix}deployment_enabled"),
157 ),
158 (
159 "nodes.maintenance_enabled",
160 compact_str::format_compact!("{prefix}maintenance_enabled"),
161 ),
162 (
163 "nodes.public_url",
164 compact_str::format_compact!("{prefix}public_url"),
165 ),
166 ("nodes.url", compact_str::format_compact!("{prefix}url")),
167 (
168 "nodes.sftp_host",
169 compact_str::format_compact!("{prefix}sftp_host"),
170 ),
171 (
172 "nodes.sftp_port",
173 compact_str::format_compact!("{prefix}sftp_port"),
174 ),
175 (
176 "nodes.memory",
177 compact_str::format_compact!("{prefix}memory"),
178 ),
179 ("nodes.disk", compact_str::format_compact!("{prefix}disk")),
180 (
181 "nodes.token_id",
182 compact_str::format_compact!("{prefix}token_id"),
183 ),
184 ("nodes.token", compact_str::format_compact!("{prefix}token")),
185 (
186 "nodes.created",
187 compact_str::format_compact!("{prefix}created"),
188 ),
189 ]);
190
191 columns.extend(super::location::Location::base_columns(Some("location_")));
192
193 columns
194 }
195
196 #[inline]
197 fn map(prefix: Option<&str>, row: &PgRow) -> Result<Self, crate::database::DatabaseError> {
198 let prefix = prefix.unwrap_or_default();
199
200 Ok(Self {
201 uuid: row.try_get(compact_str::format_compact!("{prefix}uuid").as_str())?,
202 location: super::location::Location::map(Some("location_"), row)?,
203 backup_configuration:
204 super::backup_configuration::BackupConfiguration::get_fetchable_from_row(
205 row,
206 compact_str::format_compact!("{prefix}node_backup_configuration_uuid"),
207 ),
208 name: row.try_get(compact_str::format_compact!("{prefix}name").as_str())?,
209 description: row
210 .try_get(compact_str::format_compact!("{prefix}description").as_str())?,
211 deployment_enabled: row
212 .try_get(compact_str::format_compact!("{prefix}deployment_enabled").as_str())?,
213 maintenance_enabled: row
214 .try_get(compact_str::format_compact!("{prefix}maintenance_enabled").as_str())?,
215 public_url: row
216 .try_get::<Option<String>, _>(
217 compact_str::format_compact!("{prefix}public_url").as_str(),
218 )?
219 .try_map(|url| url.parse())
220 .map_err(anyhow::Error::new)?,
221 url: row
222 .try_get::<String, _>(compact_str::format_compact!("{prefix}url").as_str())?
223 .parse()
224 .map_err(anyhow::Error::new)?,
225 sftp_host: row.try_get(compact_str::format_compact!("{prefix}sftp_host").as_str())?,
226 sftp_port: row.try_get(compact_str::format_compact!("{prefix}sftp_port").as_str())?,
227 memory: row.try_get(compact_str::format_compact!("{prefix}memory").as_str())?,
228 disk: row.try_get(compact_str::format_compact!("{prefix}disk").as_str())?,
229 token_id: row.try_get(compact_str::format_compact!("{prefix}token_id").as_str())?,
230 token: row.try_get(compact_str::format_compact!("{prefix}token").as_str())?,
231 created: row.try_get(compact_str::format_compact!("{prefix}created").as_str())?,
232 extension_data: Self::map_extensions(prefix, row)?,
233 })
234 }
235
236 fn cache_invalidation_keys(&self) -> Vec<compact_str::CompactString> {
237 vec![compact_str::format_compact!(
238 "{}::{}",
239 Self::NAME,
240 self.uuid
241 )]
242 }
243}
244
245#[async_trait::async_trait]
246impl ResolvableModel for Node {
247 type Fingerprint = EncryptedString;
248
249 fn uuid(&self) -> uuid::Uuid {
250 self.uuid
251 }
252
253 fn fingerprint(&self) -> Self::Fingerprint {
254 self.token.clone()
255 }
256
257 async fn resolve(
258 database: &crate::database::Database,
259 identifier: &str,
260 ) -> Result<Option<Self>, anyhow::Error> {
261 let Some((token_id, token)) = identifier.split_once('.') else {
262 return Ok(None);
263 };
264
265 let row = sqlx::query(sqlx::AssertSqlSafe(format!(
266 r#"
267 SELECT {}
268 FROM nodes
269 JOIN locations ON locations.uuid = nodes.location_uuid
270 WHERE nodes.token_id = $1
271 "#,
272 Self::columns_sql(None)
273 )))
274 .bind(token_id)
275 .fetch_optional(database.read())
276 .await?;
277
278 let Some(node) = row.try_map(|row| Self::map(None, &row))? else {
279 return Ok(None);
280 };
281
282 if constant_time_eq::constant_time_eq(
283 node.token.decrypt(database).await?.as_bytes(),
284 token.as_bytes(),
285 ) {
286 Ok(Some(node))
287 } else {
288 Ok(None)
289 }
290 }
291}
292
293impl Node {
294 pub const AIO_NODE_UUID: uuid::Uuid = uuid::uuid!("7dbbbb63-1734-48c4-e1de-d1a65f62cada");
295
296 pub async fn by_location_uuid_with_pagination(
297 database: &crate::database::Database,
298 location_uuid: uuid::Uuid,
299 page: i64,
300 per_page: i64,
301 search: Option<&str>,
302 ) -> Result<super::Pagination<Self>, crate::database::DatabaseError> {
303 super::PaginatedQueryBuilder::new(
304 Self::columns_sql(None),
305 format!(
306 r#"
307 FROM nodes
308 JOIN locations ON locations.uuid = nodes.location_uuid
309 WHERE nodes.location_uuid = $1 AND {search}
310 "#,
311 search = super::search_sql(2, &["nodes.name"], &["nodes.uuid"]),
312 ),
313 "nodes.created, nodes.uuid",
314 )
315 .bind(location_uuid)
316 .bind(search)
317 .fetch(database, page, per_page)
318 .await
319 }
320
321 pub async fn by_backup_configuration_uuid_with_pagination(
322 database: &crate::database::Database,
323 backup_configuration_uuid: uuid::Uuid,
324 page: i64,
325 per_page: i64,
326 search: Option<&str>,
327 ) -> Result<super::Pagination<Self>, crate::database::DatabaseError> {
328 super::PaginatedQueryBuilder::new(
329 Self::columns_sql(None),
330 format!(
331 r#"
332 FROM nodes
333 JOIN locations ON locations.uuid = nodes.location_uuid
334 WHERE nodes.backup_configuration_uuid = $1 AND {search}
335 "#,
336 search = super::search_sql(2, &["nodes.name"], &["nodes.uuid"]),
337 ),
338 "nodes.created, nodes.uuid",
339 )
340 .bind(backup_configuration_uuid)
341 .bind(search)
342 .fetch(database, page, per_page)
343 .await
344 }
345
346 pub async fn all_with_pagination(
347 database: &crate::database::Database,
348 page: i64,
349 per_page: i64,
350 search: Option<&str>,
351 ) -> Result<super::Pagination<Self>, crate::database::DatabaseError> {
352 super::PaginatedQueryBuilder::new(
353 Self::columns_sql(None),
354 format!(
355 r#"
356 FROM nodes
357 JOIN locations ON locations.uuid = nodes.location_uuid
358 WHERE {search}
359 "#,
360 search = super::search_sql(1, &["nodes.name"], &["nodes.uuid"]),
361 ),
362 "nodes.created, nodes.uuid",
363 )
364 .bind(search)
365 .fetch(database, page, per_page)
366 .await
367 }
368
369 pub async fn by_location_uuids_most_eligible(
370 database: &crate::database::Database,
371 location_uuids: &[uuid::Uuid],
372 limits: super::server::AdminApiServerLimits,
373 allow_overallocation: bool,
374 suspension_penalty: f64,
375 randomness: f64,
376 ) -> Result<Vec<Self>, crate::database::DatabaseError> {
377 let rows = sqlx::query(sqlx::AssertSqlSafe(format!(
378 r#"
379 WITH server_usage AS (
380 SELECT
381 node_uuid,
382 COALESCE(SUM((memory + memory_overhead) * CASE WHEN suspended THEN $5 ELSE 1.0 END), 0)::BIGINT AS used_memory,
383 COALESCE(SUM(disk * CASE WHEN suspended THEN $5 ELSE 1.0 END), 0)::BIGINT AS used_disk
384 FROM servers
385 GROUP BY node_uuid
386 )
387 SELECT {}, COALESCE(u.used_memory, 0) AS used_memory, COALESCE(u.used_disk, 0) AS used_disk
388 FROM nodes
389 JOIN locations ON locations.uuid = nodes.location_uuid
390 LEFT JOIN server_usage u ON nodes.uuid = u.node_uuid
391 WHERE nodes.location_uuid = ANY($1)
392 AND nodes.deployment_enabled
393 AND NOT nodes.maintenance_enabled
394 AND (
395 $4 OR (
396 (nodes.memory = 0 OR COALESCE(u.used_memory, 0) + $2 <= nodes.memory)
397 AND (nodes.disk = 0 OR COALESCE(u.used_disk, 0) + $3 <= nodes.disk)
398 )
399 )
400 ORDER BY
401 (
402 CASE WHEN nodes.memory = 0 THEN 0 ELSE GREATEST(COALESCE(u.used_memory, 0) + $2 - nodes.memory, 0) END +
403 CASE WHEN nodes.disk = 0 THEN 0 ELSE GREATEST(COALESCE(u.used_disk, 0) + $3 - nodes.disk, 0) END
404 ),
405 GREATEST(
406 CASE WHEN nodes.memory = 0 THEN 0 ELSE (COALESCE(u.used_memory, 0) + $2)::FLOAT / nodes.memory END,
407 CASE WHEN nodes.disk = 0 THEN 0 ELSE (COALESCE(u.used_disk, 0) + $3)::FLOAT / nodes.disk END
408 )
409 "#,
410 Self::columns_sql(None),
411 )))
412 .bind(location_uuids)
413 .bind(limits.memory)
414 .bind(limits.disk)
415 .bind(allow_overallocation)
416 .bind(suspension_penalty)
417 .fetch_all(database.read())
418 .await?;
419
420 let nodes = rows
421 .into_iter()
422 .map(|row| {
423 Ok((
424 Self::map(None, &row)?,
425 row.try_get::<i64, _>("used_memory")?,
426 row.try_get::<i64, _>("used_disk")?,
427 ))
428 })
429 .collect::<Result<Vec<_>, crate::database::DatabaseError>>()?;
430
431 if randomness > 0.0 {
432 let mut rng = rand::rng();
433
434 let mut keyed = nodes
435 .into_iter()
436 .map(|(node, used_memory, used_disk)| {
437 let memory_free = if node.memory == 0 {
438 1.0
439 } else {
440 1.0 - (used_memory + limits.memory) as f64 / node.memory as f64
441 };
442 let disk_free = if node.disk == 0 {
443 1.0
444 } else {
445 1.0 - (used_disk + limits.disk) as f64 / node.disk as f64
446 };
447
448 let free_ratio = f64::min(memory_free, disk_free).clamp(0.0001, 1.0);
449
450 let weight = free_ratio.powf(1.0 / randomness);
451 let key = rng.random::<f64>().powf(1.0 / weight);
452
453 (key, node)
454 })
455 .collect::<Vec<_>>();
456
457 keyed.sort_by(|(a, _), (b, _)| b.total_cmp(a));
458
459 return Ok(keyed.into_iter().map(|(_, node)| node).collect());
460 }
461
462 Ok(nodes.into_iter().map(|(node, _, _)| node).collect())
463 }
464
465 pub async fn by_name(
466 database: &crate::database::Database,
467 name: &str,
468 ) -> Result<Option<Self>, crate::database::DatabaseError> {
469 let row = sqlx::query(sqlx::AssertSqlSafe(format!(
470 r#"
471 SELECT {}
472 FROM nodes
473 JOIN locations ON locations.uuid = nodes.location_uuid
474 WHERE nodes.name = $1
475 "#,
476 Self::columns_sql(None)
477 )))
478 .bind(name)
479 .fetch_optional(database.read())
480 .await?;
481
482 row.try_map(|row| Self::map(None, &row))
483 }
484
485 pub async fn count_by_location_uuid(
486 database: &crate::database::Database,
487 location_uuid: uuid::Uuid,
488 ) -> Result<i64, sqlx::Error> {
489 sqlx::query_scalar(
490 r#"
491 SELECT COUNT(*)
492 FROM nodes
493 WHERE nodes.location_uuid = $1
494 "#,
495 )
496 .bind(location_uuid)
497 .fetch_one(database.read())
498 .await
499 }
500
501 pub async fn find_deployment_blocker(
502 database: &crate::database::Database,
503 location_uuids: &[uuid::Uuid],
504 limits: super::server::AdminApiServerLimits,
505 allow_overallocation: bool,
506 suspension_penalty: f64,
507 ) -> Result<Option<NodeDeploymentBlocker>, crate::database::DatabaseError> {
508 let row = sqlx::query(
509 r#"
510 WITH server_usage AS (
511 SELECT
512 node_uuid,
513 COALESCE(SUM((memory + memory_overhead) * CASE WHEN suspended THEN $5 ELSE 1.0 END), 0)::BIGINT AS used_memory,
514 COALESCE(SUM(disk * CASE WHEN suspended THEN $5 ELSE 1.0 END), 0)::BIGINT AS used_disk
515 FROM servers
516 GROUP BY node_uuid
517 )
518 SELECT
519 COUNT(*) AS total,
520 COUNT(*) FILTER (WHERE nodes.deployment_enabled) AS deployable,
521 COUNT(*) FILTER (WHERE nodes.deployment_enabled AND NOT nodes.maintenance_enabled) AS available,
522 COUNT(*) FILTER (
523 WHERE nodes.deployment_enabled
524 AND NOT nodes.maintenance_enabled
525 AND ($4 OR nodes.memory = 0 OR COALESCE(u.used_memory, 0) + $2 <= nodes.memory)
526 ) AS memory_ok,
527 COUNT(*) FILTER (
528 WHERE nodes.deployment_enabled
529 AND NOT nodes.maintenance_enabled
530 AND ($4 OR nodes.disk = 0 OR COALESCE(u.used_disk, 0) + $3 <= nodes.disk)
531 ) AS disk_ok,
532 COUNT(*) FILTER (
533 WHERE nodes.deployment_enabled
534 AND NOT nodes.maintenance_enabled
535 AND ($4 OR nodes.memory = 0 OR COALESCE(u.used_memory, 0) + $2 <= nodes.memory)
536 AND ($4 OR nodes.disk = 0 OR COALESCE(u.used_disk, 0) + $3 <= nodes.disk)
537 ) AS resource_ok
538 FROM nodes
539 LEFT JOIN server_usage u ON nodes.uuid = u.node_uuid
540 WHERE nodes.location_uuid = ANY($1)
541 "#,
542 )
543 .bind(location_uuids)
544 .bind(limits.memory)
545 .bind(limits.disk)
546 .bind(allow_overallocation)
547 .bind(suspension_penalty)
548 .fetch_one(database.read())
549 .await?;
550
551 let total: i64 = row.try_get("total")?;
552 let deployable: i64 = row.try_get("deployable")?;
553 let available: i64 = row.try_get("available")?;
554 let memory_ok: i64 = row.try_get("memory_ok")?;
555 let disk_ok: i64 = row.try_get("disk_ok")?;
556 let resource_ok: i64 = row.try_get("resource_ok")?;
557
558 Ok(Some(if total == 0 {
559 NodeDeploymentBlocker::NoNodes
560 } else if deployable == 0 {
561 NodeDeploymentBlocker::DeploymentDisabled
562 } else if available == 0 {
563 NodeDeploymentBlocker::MaintenanceEnabled
564 } else if resource_ok > 0 {
565 return Ok(None);
566 } else if memory_ok == 0 && disk_ok == 0 {
567 NodeDeploymentBlocker::InsufficientResources
568 } else if memory_ok == 0 {
569 NodeDeploymentBlocker::InsufficientMemory
570 } else if disk_ok == 0 {
571 NodeDeploymentBlocker::InsufficientDisk
572 } else {
573 NodeDeploymentBlocker::ResourcesSplitAcrossNodes
574 }))
575 }
576
577 pub async fn fetch_configuration(
581 &self,
582 database: &crate::database::Database,
583 ) -> Result<wings_api::Config, anyhow::Error> {
584 database
585 .cache
586 .cached(
587 &format!("node::{}::configuration::raw", self.uuid),
588 120,
589 || async {
590 Ok::<_, anyhow::Error>(wings_api::Config(
591 self.api_client(database).await?.get_system_config().await?,
592 ))
593 },
594 )
595 .await
596 }
597
598 pub async fn update_configuration(
602 &self,
603 database: &crate::database::Database,
604 config_patch: &serde_json::Value,
605 ) -> Result<bool, anyhow::Error> {
606 let response = self
607 .api_client(database)
608 .await?
609 .post_update(config_patch)
610 .await?;
611 if !response.applied {
612 return Ok(false);
613 }
614
615 database
616 .cache
617 .invalidate(&format!("node::{}::configuration::raw", self.uuid))
618 .await?;
619
620 Ok(true)
621 }
622
623 pub async fn fetch_server_resources(
627 &self,
628 database: &crate::database::Database,
629 ) -> Result<HashMap<uuid::Uuid, wings_api::ResourceUsage>, anyhow::Error> {
630 database
631 .cache
632 .cached(
633 &format!("node::{}::server_resources", self.uuid),
634 15,
635 || async {
636 let resources = self
637 .api_client(database)
638 .await?
639 .get_servers_utilization()
640 .await?;
641
642 Ok::<_, anyhow::Error>(resources.into_iter().collect())
643 },
644 )
645 .await
646 }
647
648 #[inline]
649 pub fn generate_token() -> (String, String) {
650 let token_id = rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 16);
651 let token = rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 64);
652
653 (token_id, token)
654 }
655
656 pub async fn reset_token(
657 &self,
658 state: &crate::State,
659 ) -> Result<(String, String), anyhow::Error> {
660 self.replace_token(state, None)
661 .await?
662 .ok_or_else(|| anyhow::anyhow!("node {} no longer exists", self.uuid))
663 }
664
665 pub async fn reset_token_if_unchanged(
668 &self,
669 state: &crate::State,
670 expected_token_id: &str,
671 ) -> Result<Option<(String, String)>, anyhow::Error> {
672 self.replace_token(state, Some(expected_token_id)).await
673 }
674
675 async fn replace_token(
676 &self,
677 state: &crate::State,
678 expected_token_id: Option<&str>,
679 ) -> Result<Option<(String, String)>, anyhow::Error> {
680 let (token_id, token) = Self::generate_token();
681 let (token, encrypted_token) =
682 EncryptedString::from_plaintext_with_input(token, &state.database).await?;
683
684 let result = sqlx::query(
685 r#"
686 UPDATE nodes
687 SET token_id = $2, token = $3
688 WHERE nodes.uuid = $1 AND ($4::text IS NULL OR nodes.token_id = $4)
689 "#,
690 )
691 .bind(self.uuid)
692 .bind(&token_id)
693 .bind(encrypted_token)
694 .bind(expected_token_id)
695 .execute(state.database.write())
696 .await?;
697
698 if result.rows_affected() == 0 {
699 return Ok(None);
700 }
701
702 Self::invalidate_cached(&state.database, self.uuid).await;
703
704 Self::get_event_emitter().emit(
705 state.clone(),
706 NodeEvent::TokenReset {
707 node: Box::new(self.clone()),
708 token_id: token_id.clone(),
709 token: token.clone(),
710 },
711 );
712
713 Ok(Some((token_id, token)))
714 }
715
716 #[inline]
717 pub fn is_all_in_one_node(&self) -> bool {
718 self.uuid == Self::AIO_NODE_UUID
719 }
720
721 #[inline]
722 pub fn url(&self, path: &str) -> reqwest::Url {
723 let mut url = self.url.clone();
724 url.path_segments_mut()
725 .unwrap()
726 .extend(path.trim_start_matches('/').split('/'));
727 url
728 }
729
730 #[inline]
731 pub async fn public_url(
732 &self,
733 state: &crate::State,
734 request_host: Option<&str>,
735 path: &str,
736 ) -> Result<reqwest::Url, anyhow::Error> {
737 let mut url = if self.is_all_in_one_node() {
738 let mut url = state
739 .settings
740 .get_as(|s| reqwest::Url::parse(s.app.url_for_host(request_host)))
741 .await??;
742 url.path_segments_mut()
743 .unwrap()
744 .extend(&["wings-proxy", &self.uuid.to_compact_string()]);
745 url
746 } else {
747 self.public_url.clone().unwrap_or(self.url.clone())
748 };
749
750 url.path_segments_mut()
751 .unwrap()
752 .extend(path.trim_start_matches('/').split('/'));
753
754 Ok(url)
755 }
756
757 #[inline]
758 pub async fn api_client(
759 &self,
760 database: &crate::database::Database,
761 ) -> Result<wings_api::client::WingsClient, anyhow::Error> {
762 Ok(wings_api::client::WingsClient::new(
763 self.url.to_string(),
764 self.token.decrypt(database).await?.into(),
765 ))
766 }
767
768 pub fn invalidate_permissions(
773 &self,
774 database: &Arc<crate::database::Database>,
775 invalidation: PermissionsInvalidation,
776 ) {
777 if invalidation.is_empty() {
778 return;
779 }
780
781 let as_of = chrono::Utc::now().timestamp_millis();
782 let node = self.clone();
783 let database = database.clone();
784
785 tokio::spawn(async move {
786 node.invalidate_permissions_as_of(&database, &invalidation, as_of)
787 .await;
788 });
789 }
790
791 pub fn invalidate_permissions_everywhere(
793 database: &Arc<crate::database::Database>,
794 invalidation: PermissionsInvalidation,
795 ) {
796 if invalidation.is_empty() {
797 return;
798 }
799
800 let as_of = chrono::Utc::now().timestamp_millis();
801 let database = database.clone();
802
803 tokio::spawn(async move {
804 let mut nodes = Vec::new();
805 let mut node_page = 1;
806 loop {
807 match Self::all_with_pagination(&database, node_page, 50, None).await {
808 Ok(page) if page.data.is_empty() => break,
809 Ok(page) => nodes.extend(page.data),
810 Err(err) => {
811 tracing::error!(
812 "failed to list nodes to invalidate permissions: {:?}",
813 err
814 );
815 break;
816 }
817 }
818
819 node_page += 1;
820 }
821
822 let mut futures = Vec::with_capacity(nodes.len());
823 for node in &nodes {
824 futures.push(node.invalidate_permissions_as_of(&database, &invalidation, as_of));
825 }
826
827 let mut results_stream = futures_util::stream::iter(futures).buffer_unordered(10);
828 while results_stream.next().await.is_some() {}
829 });
830 }
831
832 async fn invalidate_permissions_as_of(
833 &self,
834 database: &crate::database::Database,
835 invalidation: &PermissionsInvalidation,
836 as_of: i64,
837 ) {
838 const PERMISSIONS_INVALIDATE_TIMEOUT: std::time::Duration =
839 std::time::Duration::from_secs(5);
840
841 let result = async {
842 let api_client = self.api_client(database).await?;
843
844 let response = tokio::time::timeout(
845 PERMISSIONS_INVALIDATE_TIMEOUT,
846 api_client.post_permissions_invalidate(
847 &wings_api::permissions_invalidate::post::RequestBody {
848 as_of,
849 revoke_credentials: invalidation.revoke_credentials.clone(),
850 revoke_users: invalidation.revoke_users.clone(),
851 servers: invalidation.servers.clone(),
852 users: invalidation.users.clone(),
853 },
854 ),
855 )
856 .await?;
857
858 match response {
859 Ok(_) => Ok(()),
860 Err(wings_api::client::ApiHttpError::Http(reqwest::StatusCode::NOT_FOUND, _)) => {
861 for &user in invalidation.users.iter().chain(&invalidation.revoke_users) {
862 let deauthorized = async {
863 tokio::time::timeout(
864 PERMISSIONS_INVALIDATE_TIMEOUT,
865 api_client.post_deauthorize_user(
866 &wings_api::deauthorize_user::post::RequestBody {
867 servers: Vec::new(),
868 user,
869 },
870 ),
871 )
872 .await??;
873
874 Ok::<_, anyhow::Error>(())
875 }
876 .await;
877
878 if let Err(err) = deauthorized {
879 tracing::error!(node = %self.uuid, user = %user, "failed to deauthorize user in wings: {:?}", err);
880 }
881 }
882
883 Ok(())
884 }
885 Err(err) => Err(anyhow::Error::from(err)),
886 }
887 }
888 .await;
889
890 if let Err(err) = result {
891 tracing::error!(node = %self.uuid, "failed to invalidate permissions in wings: {:?}", err);
892 }
893 }
894
895 pub async fn fetch_firewall_support(
899 &self,
900 database: &crate::database::Database,
901 ) -> Option<bool> {
902 const FIREWALL_SUPPORT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
903
904 let config =
905 tokio::time::timeout(FIREWALL_SUPPORT_TIMEOUT, self.fetch_configuration(database))
906 .await
907 .ok()?
908 .ok()?;
909
910 firewall_supported(&config).ok()
911 }
912
913 pub async fn fetch_allowed_sources(
916 &self,
917 database: &crate::database::Database,
918 ) -> Option<NodeAllowedSources> {
919 const ALLOWED_SOURCES_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
920
921 let config =
922 tokio::time::timeout(ALLOWED_SOURCES_TIMEOUT, self.fetch_configuration(database))
923 .await
924 .ok()?
925 .ok()?;
926
927 NodeAllowedSources::from_config(&config).ok()
928 }
929
930 pub async fn bandwidth_status(
931 &self,
932 database: &crate::database::Database,
933 ) -> wings_api::BandwidthStatus {
934 let status = tokio::time::timeout(std::time::Duration::from_secs(3), async {
935 Ok::<_, anyhow::Error>(
936 self.api_client(database)
937 .await?
938 .get_system()
939 .await?
940 .bandwidth,
941 )
942 })
943 .await;
944
945 match status {
946 Ok(Ok(status)) => status,
947 Ok(Err(err)) => wings_api::BandwidthStatus {
948 enabled: false,
949 ready: false,
950 reason: Some(compact_str::format_compact!("unable to check wings: {err}")),
951 },
952 Err(_) => wings_api::BandwidthStatus {
953 enabled: false,
954 ready: false,
955 reason: Some("timed out checking wings".into()),
956 },
957 }
958 }
959
960 pub async fn ensure_bandwidth_ready(
961 &self,
962 database: &crate::database::Database,
963 ) -> Result<(), anyhow::Error> {
964 let status = self.bandwidth_status(database).await;
965 if status.enabled && status.ready {
966 return Ok(());
967 }
968
969 let reason = if status.ready {
970 "bandwidth limits are disabled on this node".into()
971 } else {
972 status
973 .reason
974 .unwrap_or_else(|| "node does not support bandwidth limits".into())
975 };
976
977 Err(crate::response::DisplayError::new(reason)
978 .with_status(axum::http::StatusCode::CONFLICT)
979 .into())
980 }
981
982 pub async fn fetch_tunnel_status(
985 &self,
986 database: &crate::database::Database,
987 ) -> Option<wings_api::TundraStatus> {
988 const TUNNEL_STATUS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
989
990 tokio::time::timeout(TUNNEL_STATUS_TIMEOUT, async {
991 self.api_client(database)
992 .await
993 .ok()?
994 .get_tundra()
995 .await
996 .ok()
997 })
998 .await
999 .ok()?
1000 }
1001
1002 pub async fn used_ports(
1003 &self,
1004 state: &crate::State,
1005 ips: &[std::net::IpAddr],
1006 ) -> Result<HashMap<std::net::IpAddr, Vec<u16>>, anyhow::Error> {
1007 let mut used = HashMap::new();
1008 let mut missing = Vec::new();
1009
1010 const USED_PORTS_TTL: u64 = 10;
1011 const USED_PORTS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
1012
1013 #[inline]
1014 fn used_ports_cache_key(
1015 node_uuid: uuid::Uuid,
1016 ip: std::net::IpAddr,
1017 ) -> compact_str::CompactString {
1018 compact_str::format_compact!("nodes::{node_uuid}::used_ports::{ip}")
1019 }
1020
1021 for ip in ips {
1022 match state
1023 .cache
1024 .get(&used_ports_cache_key(self.uuid, *ip))
1025 .await?
1026 {
1027 Some(ports) => {
1028 used.insert(*ip, ports);
1029 }
1030 None => missing.push(*ip),
1031 }
1032 }
1033
1034 if missing.is_empty() {
1035 return Ok(used);
1036 }
1037
1038 let client = self.api_client(&state.database).await?;
1039 let response = tokio::time::timeout(
1040 USED_PORTS_TIMEOUT,
1041 client.get_ports_used(&wings_api::ports_used::get::Query {
1042 ip: Some(
1043 missing
1044 .iter()
1045 .map(|ip| compact_str::format_compact!("{ip}"))
1046 .collect(),
1047 ),
1048 ..Default::default()
1049 }),
1050 )
1051 .await
1052 .map_err(|_| anyhow::anyhow!("timed out asking the node which ports are in use"))??;
1053
1054 for ip in missing {
1055 let ports: Vec<_> = response
1056 .used
1057 .get(compact_str::format_compact!("{ip}").as_str())
1058 .map(|ports| ports.iter().map(|port| port.port as u16).collect())
1059 .unwrap_or_default();
1060
1061 state
1062 .cache
1063 .set(&used_ports_cache_key(self.uuid, ip), USED_PORTS_TTL, &ports)
1064 .await?;
1065 used.insert(ip, ports);
1066 }
1067
1068 Ok(used)
1069 }
1070
1071 #[inline]
1072 pub async fn create_jwt<T: Serialize>(
1073 &self,
1074 database: &crate::database::Database,
1075 jwt: &crate::jwt::Jwt,
1076 payload: &T,
1077 ) -> Result<String, anyhow::Error> {
1078 Ok(jwt.create_custom(self.token.decrypt(database).await?.as_bytes(), payload)?)
1079 }
1080}
1081
1082fn firewall_supported(config: &wings_api::Config) -> Result<bool, serde_json::Error> {
1084 nestify::nest! {
1085 #[derive(Default, Deserialize)] #[serde(default)] struct FirewallConfig {
1086 system: #[derive(Default, Deserialize)] #[serde(default)] struct FirewallConfigSystem {
1087 user: #[derive(Default, Deserialize)] #[serde(default)] struct FirewallConfigSystemUser {
1088 rootless: #[derive(Default, Deserialize)] #[serde(default)] struct FirewallConfigSystemUserRootless {
1089 enabled: bool,
1090 },
1091 },
1092 },
1093 docker: #[derive(Default, Deserialize)] #[serde(default)] struct FirewallConfigDocker {
1094 firewall: Option<#[derive(Default, Deserialize)] #[serde(default)] struct FirewallConfigDockerFirewall {
1096 backend: compact_str::CompactString,
1097 }>,
1098 },
1099 }
1100 }
1101
1102 let config: FirewallConfig = config.view()?;
1103
1104 Ok(!config.system.user.rootless.enabled
1105 && config
1106 .docker
1107 .firewall
1108 .is_some_and(|firewall| firewall.backend != "disabled"))
1109}
1110
1111#[async_trait::async_trait]
1112impl IntoAdminApiObject for Node {
1113 type AdminApiObject = AdminApiNode;
1114 type ExtraArgs<'a> = ();
1115
1116 async fn into_admin_api_object<'a>(
1117 self,
1118 state: &crate::State,
1119 _args: Self::ExtraArgs<'a>,
1120 ) -> Result<Self::AdminApiObject, crate::database::DatabaseError> {
1121 let api_object = AdminApiNode::init_hooks(&self, state).await?;
1122
1123 let public_url = if self.is_all_in_one_node() {
1124 Some(self.public_url(state, None, "/").await?.to_string())
1125 } else {
1126 self.public_url.map(|url| url.to_string())
1127 };
1128
1129 let (location, backup_configuration) =
1130 tokio::join!(self.location.into_admin_api_object(state, ()), async {
1131 if let Some(backup_configuration) = self.backup_configuration {
1132 if let Ok(backup_configuration) =
1133 backup_configuration.fetch_cached(&state.database).await
1134 {
1135 backup_configuration
1136 .into_admin_api_object(state, ())
1137 .await
1138 .ok()
1139 } else {
1140 None
1141 }
1142 } else {
1143 None
1144 }
1145 });
1146
1147 let api_object = finish_extendible!(
1148 AdminApiNode {
1149 uuid: self.uuid,
1150 location: location?,
1151 backup_configuration,
1152 name: self.name,
1153 description: self.description,
1154 deployment_enabled: self.deployment_enabled,
1155 maintenance_enabled: self.maintenance_enabled,
1156 public_url,
1157 url: self.url.to_string(),
1158 sftp_host: self.sftp_host,
1159 sftp_port: self.sftp_port,
1160 memory: self.memory,
1161 disk: self.disk,
1162 created: self.created.and_utc(),
1163 },
1164 api_object,
1165 state
1166 )?;
1167
1168 Ok(api_object)
1169 }
1170}
1171
1172#[async_trait::async_trait]
1173impl ByUuid for Node {
1174 async fn by_uuid(
1175 database: &crate::database::Database,
1176 uuid: uuid::Uuid,
1177 ) -> Result<Self, crate::database::DatabaseError> {
1178 let row = sqlx::query(sqlx::AssertSqlSafe(format!(
1179 r#"
1180 SELECT {}, {}
1181 FROM nodes
1182 JOIN locations ON locations.uuid = nodes.location_uuid
1183 WHERE nodes.uuid = $1
1184 "#,
1185 Self::columns_sql(None),
1186 super::location::Location::columns_sql(Some("location_")),
1187 )))
1188 .bind(uuid)
1189 .fetch_one(database.read())
1190 .await?;
1191
1192 Self::map(None, &row)
1193 }
1194
1195 async fn by_uuid_with_transaction(
1196 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1197 uuid: uuid::Uuid,
1198 ) -> Result<Self, crate::database::DatabaseError> {
1199 let row = sqlx::query(sqlx::AssertSqlSafe(format!(
1200 r#"
1201 SELECT {}, {}
1202 FROM nodes
1203 JOIN locations ON locations.uuid = nodes.location_uuid
1204 WHERE nodes.uuid = $1
1205 "#,
1206 Self::columns_sql(None),
1207 super::location::Location::columns_sql(Some("location_")),
1208 )))
1209 .bind(uuid)
1210 .fetch_one(&mut **transaction)
1211 .await?;
1212
1213 Self::map(None, &row)
1214 }
1215}
1216
1217#[derive(ToSchema, Deserialize, Validate)]
1218pub struct CreateNodeOptions {
1219 #[garde(skip)]
1220 pub location_uuid: uuid::Uuid,
1221 #[garde(skip)]
1222 pub backup_configuration_uuid: Option<uuid::Uuid>,
1223 #[garde(length(chars, min = 1, max = 255))]
1224 #[schema(min_length = 1, max_length = 255)]
1225 pub name: compact_str::CompactString,
1226 #[garde(length(chars, min = 1, max = 1024))]
1227 #[schema(min_length = 1, max_length = 1024)]
1228 pub description: Option<compact_str::CompactString>,
1229 #[garde(skip)]
1230 pub deployment_enabled: bool,
1231 #[garde(skip)]
1232 pub maintenance_enabled: bool,
1233 #[garde(length(chars, min = 3, max = 255), url)]
1234 #[schema(min_length = 3, max_length = 255, format = "uri")]
1235 pub public_url: Option<compact_str::CompactString>,
1236 #[garde(length(chars, min = 3, max = 255), url)]
1237 #[schema(min_length = 3, max_length = 255, format = "uri")]
1238 pub url: compact_str::CompactString,
1239 #[garde(length(chars, min = 3, max = 255))]
1240 #[schema(min_length = 3, max_length = 255)]
1241 pub sftp_host: Option<compact_str::CompactString>,
1242 #[garde(range(min = 1))]
1243 #[schema(minimum = 1)]
1244 pub sftp_port: u16,
1245 #[garde(range(min = 0))]
1246 #[schema(minimum = 0)]
1247 pub memory: i64,
1248 #[garde(range(min = 0))]
1249 #[schema(minimum = 0)]
1250 pub disk: i64,
1251}
1252
1253#[async_trait::async_trait]
1254impl CreatableModel for Node {
1255 type CreateOptions<'a> = CreateNodeOptions;
1256 type CreateResult = Self;
1257
1258 fn get_create_handlers() -> &'static LazyLock<CreateListenerList<Self>> {
1259 static CREATE_LISTENERS: LazyLock<CreateListenerList<Node>> =
1260 LazyLock::new(|| Arc::new(ModelHandlerList::default()));
1261
1262 &CREATE_LISTENERS
1263 }
1264
1265 async fn create_with_transaction(
1266 state: &crate::State,
1267 mut options: Self::CreateOptions<'_>,
1268 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1269 ) -> Result<Self, crate::database::DatabaseError> {
1270 options.validate()?;
1271
1272 if let Some(backup_configuration_uuid) = &options.backup_configuration_uuid {
1273 super::backup_configuration::BackupConfiguration::by_uuid_optional(
1274 &state.database,
1275 *backup_configuration_uuid,
1276 )
1277 .await?
1278 .ok_or(crate::database::InvalidRelationError(
1279 "backup_configuration",
1280 ))?;
1281 }
1282
1283 let mut query_builder = InsertQueryBuilder::new("nodes");
1284
1285 Self::run_create_handlers(&mut options, &mut query_builder, state, transaction).await?;
1286
1287 let (token_id, token) = Self::generate_token();
1288
1289 query_builder
1290 .set("location_uuid", options.location_uuid)
1291 .set(
1292 "backup_configuration_uuid",
1293 options.backup_configuration_uuid,
1294 )
1295 .set("name", &options.name)
1296 .set("description", &options.description)
1297 .set("deployment_enabled", options.deployment_enabled)
1298 .set("maintenance_enabled", options.maintenance_enabled)
1299 .set("public_url", &options.public_url)
1300 .set("url", &options.url)
1301 .set("sftp_host", &options.sftp_host)
1302 .set("sftp_port", options.sftp_port as i32)
1303 .set("memory", options.memory)
1304 .set("disk", options.disk)
1305 .set("token_id", token_id.clone())
1306 .set(
1307 "token",
1308 EncryptedString::from_plaintext(token, &state.database).await?,
1309 );
1310
1311 let row = query_builder
1312 .returning("uuid")
1313 .fetch_one(&mut **transaction)
1314 .await?;
1315 let uuid: uuid::Uuid = row.try_get("uuid")?;
1316
1317 let mut result = Self::by_uuid_with_transaction(transaction, uuid).await?;
1318
1319 Self::run_after_create_handlers(&mut result, &options, state, transaction).await?;
1320
1321 Ok(result)
1322 }
1323}
1324
1325#[derive(ToSchema, Serialize, Deserialize, Validate, Clone, Default)]
1326pub struct UpdateNodeOptions {
1327 #[garde(skip)]
1328 pub location_uuid: Option<uuid::Uuid>,
1329 #[serde(
1330 default,
1331 skip_serializing_if = "Option::is_none",
1332 with = "::serde_with::rust::double_option"
1333 )]
1334 #[garde(skip)]
1335 pub backup_configuration_uuid: Option<Option<uuid::Uuid>>,
1336 #[garde(length(chars, min = 1, max = 255))]
1337 #[schema(min_length = 1, max_length = 255)]
1338 pub name: Option<compact_str::CompactString>,
1339 #[garde(length(chars, min = 1, max = 1024))]
1340 #[schema(min_length = 1, max_length = 1024)]
1341 #[serde(
1342 default,
1343 skip_serializing_if = "Option::is_none",
1344 with = "::serde_with::rust::double_option"
1345 )]
1346 pub description: Option<Option<compact_str::CompactString>>,
1347 #[garde(skip)]
1348 pub deployment_enabled: Option<bool>,
1349 #[garde(skip)]
1350 pub maintenance_enabled: Option<bool>,
1351 #[garde(length(chars, min = 3, max = 255), url)]
1352 #[schema(min_length = 3, max_length = 255, format = "uri")]
1353 #[serde(
1354 default,
1355 skip_serializing_if = "Option::is_none",
1356 with = "::serde_with::rust::double_option"
1357 )]
1358 pub public_url: Option<Option<compact_str::CompactString>>,
1359 #[garde(length(chars, min = 3, max = 255), url)]
1360 #[schema(min_length = 3, max_length = 255, format = "uri")]
1361 pub url: Option<compact_str::CompactString>,
1362 #[garde(length(chars, min = 3, max = 255))]
1363 #[schema(min_length = 3, max_length = 255)]
1364 #[serde(
1365 default,
1366 skip_serializing_if = "Option::is_none",
1367 with = "::serde_with::rust::double_option"
1368 )]
1369 pub sftp_host: Option<Option<compact_str::CompactString>>,
1370 #[garde(range(min = 1))]
1371 #[schema(minimum = 1)]
1372 pub sftp_port: Option<u16>,
1373 #[garde(range(min = 0))]
1374 #[schema(minimum = 0)]
1375 pub memory: Option<i64>,
1376 #[garde(range(min = 0))]
1377 #[schema(minimum = 0)]
1378 pub disk: Option<i64>,
1379}
1380
1381#[async_trait::async_trait]
1382impl UpdatableModel for Node {
1383 type UpdateOptions = UpdateNodeOptions;
1384
1385 fn get_update_handlers() -> &'static LazyLock<UpdateHandlerList<Self>> {
1386 static UPDATE_LISTENERS: LazyLock<UpdateHandlerList<Node>> =
1387 LazyLock::new(|| Arc::new(ModelHandlerList::default()));
1388
1389 &UPDATE_LISTENERS
1390 }
1391
1392 async fn update_with_transaction(
1393 &mut self,
1394 state: &crate::State,
1395 mut options: Self::UpdateOptions,
1396 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1397 ) -> Result<(), crate::database::DatabaseError> {
1398 options.validate()?;
1399
1400 let location = if let Some(location_uuid) = options.location_uuid {
1401 Some(
1402 super::location::Location::by_uuid_optional(&state.database, location_uuid)
1403 .await?
1404 .ok_or(crate::database::InvalidRelationError("location"))?,
1405 )
1406 } else {
1407 None
1408 };
1409
1410 let backup_configuration =
1411 if let Some(backup_configuration_uuid) = &options.backup_configuration_uuid {
1412 match backup_configuration_uuid {
1413 Some(uuid) => {
1414 super::backup_configuration::BackupConfiguration::by_uuid_optional(
1415 &state.database,
1416 *uuid,
1417 )
1418 .await?
1419 .ok_or(crate::database::InvalidRelationError(
1420 "backup_configuration",
1421 ))?;
1422
1423 Some(Some(
1424 super::backup_configuration::BackupConfiguration::get_fetchable(*uuid),
1425 ))
1426 }
1427 None => Some(None),
1428 }
1429 } else {
1430 None
1431 };
1432
1433 let mut query_builder = UpdateQueryBuilder::new("nodes");
1434
1435 self.run_update_handlers(&mut options, &mut query_builder, state, transaction)
1436 .await?;
1437
1438 query_builder
1439 .set("location_uuid", options.location_uuid.as_ref())
1440 .set(
1441 "backup_configuration_uuid",
1442 options
1443 .backup_configuration_uuid
1444 .as_ref()
1445 .map(|u| u.as_ref()),
1446 )
1447 .set("name", options.name.as_ref())
1448 .set(
1449 "description",
1450 options.description.as_ref().map(|d| d.as_ref()),
1451 )
1452 .set("deployment_enabled", options.deployment_enabled)
1453 .set("maintenance_enabled", options.maintenance_enabled)
1454 .set(
1455 "public_url",
1456 options.public_url.as_ref().map(|u| u.as_ref()),
1457 )
1458 .set("url", options.url.as_ref())
1459 .set("sftp_host", options.sftp_host.as_ref().map(|h| h.as_ref()))
1460 .set("sftp_port", options.sftp_port.as_ref().map(|p| *p as i32))
1461 .set("memory", options.memory.as_ref())
1462 .set("disk", options.disk.as_ref())
1463 .where_eq("uuid", self.uuid);
1464
1465 query_builder.execute(&mut **transaction).await?;
1466
1467 if let Some(location) = location {
1468 self.location = location;
1469 }
1470 if let Some(backup_configuration) = backup_configuration {
1471 self.backup_configuration = backup_configuration;
1472 }
1473 if let Some(name) = options.name {
1474 self.name = name;
1475 }
1476 if let Some(description) = options.description {
1477 self.description = description;
1478 }
1479 if let Some(deployment_enabled) = options.deployment_enabled {
1480 self.deployment_enabled = deployment_enabled;
1481 }
1482 if let Some(maintenance_enabled) = options.maintenance_enabled {
1483 self.maintenance_enabled = maintenance_enabled;
1484 }
1485 if let Some(public_url) = options.public_url {
1486 self.public_url = public_url
1487 .try_map(|url| url.parse())
1488 .map_err(anyhow::Error::new)?;
1489 }
1490 if let Some(url) = options.url {
1491 self.url = url.parse().map_err(anyhow::Error::new)?;
1492 }
1493 if let Some(sftp_host) = options.sftp_host {
1494 self.sftp_host = sftp_host;
1495 }
1496 if let Some(sftp_port) = options.sftp_port {
1497 self.sftp_port = sftp_port as i32;
1498 }
1499 if let Some(memory) = options.memory {
1500 self.memory = memory;
1501 }
1502 if let Some(disk) = options.disk {
1503 self.disk = disk;
1504 }
1505
1506 self.run_after_update_handlers(state, transaction).await?;
1507
1508 Ok(())
1509 }
1510}
1511
1512#[async_trait::async_trait]
1513impl DeletableModel for Node {
1514 type DeleteOptions = ();
1515
1516 fn get_delete_handlers() -> &'static LazyLock<DeleteHandlerList<Self>> {
1517 static DELETE_LISTENERS: LazyLock<DeleteHandlerList<Node>> =
1518 LazyLock::new(|| Arc::new(ModelHandlerList::default()));
1519
1520 &DELETE_LISTENERS
1521 }
1522
1523 async fn delete_with_transaction(
1524 &self,
1525 state: &crate::State,
1526 options: Self::DeleteOptions,
1527 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1528 ) -> Result<(), anyhow::Error> {
1529 if self.is_all_in_one_node() && state.container_type.is_all_in_one() {
1530 return Err(anyhow::anyhow!("The AIO node cannot be deleted"));
1531 }
1532
1533 self.run_delete_handlers(&options, state, transaction)
1534 .await?;
1535
1536 crate::tunnel::bump_epoch_if_node_on_mesh(transaction, self.uuid).await?;
1537
1538 sqlx::query(
1539 r#"
1540 DELETE FROM nodes
1541 WHERE nodes.uuid = $1
1542 "#,
1543 )
1544 .bind(self.uuid)
1545 .execute(&mut **transaction)
1546 .await?;
1547
1548 self.run_after_delete_handlers(&options, state, transaction)
1549 .await?;
1550
1551 Ok(())
1552 }
1553}
1554
1555#[derive(Validate)]
1556pub struct DuplicateNodeOptions {
1557 #[garde(length(chars, min = 1, max = 255))]
1558 pub name: compact_str::CompactString,
1559}
1560
1561#[async_trait::async_trait]
1562impl DuplicableModel for Node {
1563 type DuplicateOptions<'a> = DuplicateNodeOptions;
1564 type DuplicateResult = Self;
1565
1566 fn get_duplicate_handlers() -> &'static LazyLock<DuplicateHandlerList<Self>> {
1567 static DUPLICATE_LISTENERS: LazyLock<DuplicateHandlerList<Node>> =
1568 LazyLock::new(|| Arc::new(ModelHandlerList::default()));
1569
1570 &DUPLICATE_LISTENERS
1571 }
1572
1573 async fn duplicate_with_transaction(
1574 &self,
1575 state: &crate::State,
1576 options: Self::DuplicateOptions<'_>,
1577 transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1578 ) -> Result<Self, crate::database::DatabaseError> {
1579 options.validate()?;
1580
1581 self.run_duplicate_handlers(&options, state, transaction)
1582 .await?;
1583
1584 let mut query_builder = InsertQueryBuilder::new("nodes");
1585
1586 let (token_id, token) = Self::generate_token();
1587
1588 query_builder
1589 .set("location_uuid", self.location.uuid)
1590 .set(
1591 "backup_configuration_uuid",
1592 self.backup_configuration.as_ref().map(|c| c.uuid),
1593 )
1594 .set("name", &options.name)
1595 .set("description", &self.description)
1596 .set("deployment_enabled", self.deployment_enabled)
1597 .set("maintenance_enabled", self.maintenance_enabled)
1598 .set("public_url", self.public_url.as_ref().map(|u| u.as_str()))
1599 .set("url", self.url.as_str())
1600 .set("sftp_host", &self.sftp_host)
1601 .set("sftp_port", self.sftp_port)
1602 .set("memory", self.memory)
1603 .set("disk", self.disk)
1604 .set("token_id", token_id)
1605 .set(
1606 "token",
1607 EncryptedString::from_plaintext(token, &state.database).await?,
1608 );
1609
1610 let row = query_builder
1611 .returning("uuid")
1612 .fetch_one(&mut **transaction)
1613 .await?;
1614 let uuid: uuid::Uuid = row.try_get("uuid")?;
1615
1616 let mut node = Self::by_uuid_with_transaction(transaction, uuid).await?;
1617
1618 sqlx::query!(
1619 "INSERT INTO node_mounts (node_uuid, mount_uuid)
1620 SELECT $1, node_mounts.mount_uuid
1621 FROM node_mounts
1622 WHERE node_mounts.node_uuid = $2",
1623 node.uuid,
1624 self.uuid,
1625 )
1626 .execute(&mut **transaction)
1627 .await?;
1628
1629 sqlx::query(
1630 "INSERT INTO node_devices (node_uuid, device_uuid)
1631 SELECT $1, node_devices.device_uuid
1632 FROM node_devices
1633 WHERE node_devices.node_uuid = $2",
1634 )
1635 .bind(node.uuid)
1636 .bind(self.uuid)
1637 .execute(&mut **transaction)
1638 .await?;
1639
1640 sqlx::query!(
1641 "INSERT INTO node_database_hosts (node_uuid, database_host_uuid)
1642 SELECT $1, node_database_hosts.database_host_uuid
1643 FROM node_database_hosts
1644 WHERE node_database_hosts.node_uuid = $2",
1645 node.uuid,
1646 self.uuid,
1647 )
1648 .execute(&mut **transaction)
1649 .await?;
1650
1651 sqlx::query!(
1652 "INSERT INTO node_database_agent_hosts (node_uuid, database_agent_host_uuid)
1653 SELECT $1, node_database_agent_hosts.database_agent_host_uuid
1654 FROM node_database_agent_hosts
1655 WHERE node_database_agent_hosts.node_uuid = $2",
1656 node.uuid,
1657 self.uuid,
1658 )
1659 .execute(&mut **transaction)
1660 .await?;
1661
1662 self.run_after_duplicate_handlers(&mut node, &options, state, transaction)
1663 .await?;
1664
1665 Ok(node)
1666 }
1667}
1668
1669#[schema_extension_derive::extendible]
1670#[init_args(Node, crate::State)]
1671#[hook_args(crate::State)]
1672#[derive(ToSchema, Serialize)]
1673#[schema(title = "Node")]
1674pub struct AdminApiNode {
1675 pub uuid: uuid::Uuid,
1676 pub location: super::location::AdminApiLocation,
1677 pub backup_configuration: Option<super::backup_configuration::AdminApiBackupConfiguration>,
1678
1679 pub name: compact_str::CompactString,
1680 pub description: Option<compact_str::CompactString>,
1681
1682 pub deployment_enabled: bool,
1683 pub maintenance_enabled: bool,
1684
1685 #[schema(format = "uri")]
1686 pub public_url: Option<String>,
1687 #[schema(format = "uri")]
1688 pub url: String,
1689 pub sftp_host: Option<compact_str::CompactString>,
1690 pub sftp_port: i32,
1691
1692 pub memory: i64,
1693 pub disk: i64,
1694
1695 pub created: chrono::DateTime<chrono::Utc>,
1696}
1697
1698#[cfg(test)]
1699mod tests {
1700 use super::*;
1701 use serde_json::json;
1702
1703 #[test]
1705 fn firewall_is_unsupported_without_a_firewall_section() {
1706 let config = wings_api::Config(json!({ "docker": { "network": {} } }));
1707
1708 assert!(!firewall_supported(&config).unwrap());
1709 }
1710
1711 #[test]
1712 fn firewall_is_supported_when_the_section_has_no_backend() {
1713 let config = wings_api::Config(json!({ "docker": { "firewall": {} } }));
1714
1715 assert!(firewall_supported(&config).unwrap());
1716 }
1717
1718 #[test]
1719 fn firewall_is_unsupported_when_the_backend_is_disabled() {
1720 let config = wings_api::Config(json!({
1721 "docker": { "firewall": { "backend": "disabled" } },
1722 }));
1723
1724 assert!(!firewall_supported(&config).unwrap());
1725 }
1726
1727 #[test]
1728 fn unknown_firewall_backends_count_as_supported() {
1729 let config = wings_api::Config(json!({
1730 "docker": { "firewall": { "backend": "some_future_backend", "new_option": 1 } },
1731 }));
1732
1733 assert!(firewall_supported(&config).unwrap());
1734 }
1735
1736 #[test]
1737 fn firewall_is_unsupported_when_rootless() {
1738 let rootless = wings_api::Config(json!({
1739 "system": { "user": { "rootless": { "enabled": true } } },
1740 "docker": { "firewall": { "backend": "nftables" } },
1741 }));
1742 let rootful = wings_api::Config(json!({
1743 "system": { "user": { "rootless": { "enabled": false } } },
1744 "docker": { "firewall": { "backend": "nftables" } },
1745 }));
1746
1747 assert!(!firewall_supported(&rootless).unwrap());
1748 assert!(firewall_supported(&rootful).unwrap());
1749 }
1750}