Skip to main content

shared/models/node/
mod.rs

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    /// No nodes exist in the selected location(s).
28    NoNodes,
29    /// Nodes exist but deployment is disabled on all of them.
30    DeploymentDisabled,
31    /// No node has enough unallocated memory.
32    InsufficientMemory,
33    /// No node has enough unallocated disk.
34    InsufficientDisk,
35    /// No node has enough unallocated memory or disk.
36    InsufficientResources,
37    /// Memory and disk are each available somewhere, but no single node has
38    /// enough of both at once.
39    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    /// Fetch the current configuration of this node
566    ///
567    /// Cached for 120 seconds.
568    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    /// Update the configuration of this node
587    ///
588    /// Invalidates the cached configuration.
589    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    /// Fetch the current resource usages of all servers on this node.
612    ///
613    /// Cached for 15 seconds.
614    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}