plane 0.5.5

Session backend orchestrator for ambitious browser-based apps.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

use self::{
    acme::AcmeDatabase,
    backend::BackendDatabase,
    backend_actions::BackendActionDatabase,
    backend_key::KeysDatabase,
    cluster::ClusterDatabase,
    connect::ConnectError,
    controller::ControllerDatabase,
    drone::DroneDatabase,
    node::NodeDatabase,
    subscribe::{EventSubscriptionManager, Notification, NotificationPayload, Subscription},
};
use plane_common::{
    types::{ClusterName, ConnectRequest, ConnectResponse, RevokeRequest},
    PlaneClient,
};
use serde_json::Value;
use sqlx::{postgres::PgPoolOptions, PgPool};
use std::sync::{Arc, OnceLock};
use tokio::sync::broadcast::Receiver;

pub mod acme;
pub mod backend;
pub mod backend_actions;
pub mod backend_key;
pub mod cluster;
pub mod connect;
pub mod controller;
pub mod drone;
pub mod node;
pub mod subscribe;
pub mod util;

pub async fn connect_and_migrate(db: &str) -> sqlx::Result<PlaneDatabase> {
    let db_pool = PgPoolOptions::new().connect(db).await?;
    sqlx::migrate!("schema/migrations").run(&db_pool).await?;
    Ok(PlaneDatabase::new(db_pool))
}

pub async fn connect(db: &str) -> sqlx::Result<PlaneDatabase> {
    let db_pool = PgPoolOptions::new().connect(db).await?;
    Ok(PlaneDatabase::new(db_pool))
}

#[derive(Clone)]
pub struct PlaneDatabase {
    pub pool: PgPool,
    subscription_manager: Arc<OnceLock<EventSubscriptionManager>>,
}

impl PlaneDatabase {
    pub fn new(pool: PgPool) -> Self {
        Self {
            pool,
            subscription_manager: Arc::default(),
        }
    }

    pub fn acme(&self) -> acme::AcmeDatabase {
        AcmeDatabase::new(&self.pool)
    }

    pub fn drone(&self) -> DroneDatabase {
        DroneDatabase::new(&self.pool)
    }

    pub fn cluster(&self) -> ClusterDatabase {
        ClusterDatabase::new(&self.pool)
    }

    pub fn node(&self) -> NodeDatabase {
        NodeDatabase::new(&self.pool)
    }

    pub fn backend(&self) -> BackendDatabase {
        BackendDatabase::new(self)
    }

    pub fn backend_actions(&self) -> BackendActionDatabase {
        BackendActionDatabase::new(&self.pool)
    }

    pub fn keys(&self) -> backend_key::KeysDatabase {
        KeysDatabase::new(&self.pool)
    }

    pub fn controller(&self) -> controller::ControllerDatabase {
        ControllerDatabase::new(&self.pool)
    }

    pub async fn health_check(&self) -> Result<(), sqlx::Error> {
        sqlx::query_scalar!("select 1")
            .fetch_one(&self.pool)
            .await?;
        Ok(())
    }

    pub async fn connect(
        &self,
        default_cluster: Option<&ClusterName>,
        request: &ConnectRequest,
        client: &PlaneClient,
    ) -> Result<ConnectResponse, ConnectError> {
        connect::connect(&self.pool, default_cluster, request, client).await
    }
    pub async fn revoke(&self, request: &RevokeRequest) -> Result<(), ConnectError> {
        connect::revoke(&self.pool, request).await
    }

    pub async fn clean_up_tokens(&self) -> Result<(), sqlx::Error> {
        connect::clean_up_tokens(&self.pool).await
    }

    /// this limits the number of events returned, so it may be necessary
    /// to call this function multiple times to get all events since a given id
    pub async fn get_events_since(
        &self,
        since: i32,
    ) -> Result<Vec<Notification<Value>>, sqlx::Error> {
        EventSubscriptionManager::get_events_since(&self.pool, since).await
    }

    fn subscription_manager(&self) -> &EventSubscriptionManager {
        self.subscription_manager
            .get_or_init(|| EventSubscriptionManager::new(&self.pool))
    }

    pub fn subscribe<T: NotificationPayload>(&self) -> Subscription<T> {
        self.subscription_manager().subscribe(None)
    }

    pub fn subscribe_with_key<T: NotificationPayload>(&self, key: &str) -> Subscription<T> {
        self.subscription_manager().subscribe(Some(key))
    }

    pub fn subscribe_all_events(&self) -> Receiver<Notification<Value>> {
        self.subscription_manager().subscribe_all_events()
    }
}