openleadr_vtn/data_source/postgres/
mod.rs

1#[cfg(feature = "internal-oauth")]
2use crate::data_source::{AuthSource, postgres::user::PgAuthSource};
3
4use super::{Migrate, VenObjectPrivacy};
5use crate::{
6    data_source::{
7        DataSource, EventCrud, ProgramCrud, ReportCrud, ResourceCrud, ResourceGroupCrud, VenCrud,
8        postgres::{
9            event::PgEventStorage, program::PgProgramStorage, report::PgReportStorage,
10            resource_group::PgResourceGroupStorage, subscription::PgSubscriptionStorage,
11            ven::PgVenStorage,
12        },
13    },
14    error::AppError,
15};
16use async_trait::async_trait;
17use dotenvy::dotenv;
18use openleadr_wire::{ClientId, target::Target};
19use resource::PgResourceStorage;
20use serde::Serialize;
21use sqlx::{PgPool, migrate::MigrateError, postgres::PgPoolOptions};
22use std::sync::Arc;
23use tracing::{error, info};
24
25mod event;
26mod program;
27mod report;
28mod resource;
29mod resource_group;
30mod subscription;
31#[cfg(feature = "internal-oauth")]
32mod user;
33mod ven;
34
35#[derive(Clone)]
36pub struct PostgresStorage {
37    db: PgPool,
38}
39
40impl DataSource for PostgresStorage {
41    fn programs(&self) -> Arc<dyn ProgramCrud> {
42        Arc::<PgProgramStorage>::new(self.db.clone().into())
43    }
44
45    fn reports(&self) -> Arc<dyn ReportCrud> {
46        Arc::<PgReportStorage>::new(self.db.clone().into())
47    }
48
49    fn events(&self) -> Arc<dyn EventCrud> {
50        Arc::<PgEventStorage>::new(self.db.clone().into())
51    }
52
53    fn vens(&self) -> Arc<dyn VenCrud> {
54        Arc::<PgVenStorage>::new(self.db.clone().into())
55    }
56
57    fn ven_object_privacy(&self) -> Arc<dyn VenObjectPrivacy> {
58        Arc::<PgVenStorage>::new(self.db.clone().into())
59    }
60
61    fn resources(&self) -> Arc<dyn ResourceCrud> {
62        Arc::<PgResourceStorage>::new(self.db.clone().into())
63    }
64
65    fn resource_groups(&self) -> Arc<dyn ResourceGroupCrud> {
66        Arc::<PgResourceGroupStorage>::new(self.db.clone().into())
67    }
68
69    fn subscriptions(&self) -> Arc<dyn super::SubscriptionCrud> {
70        Arc::<PgSubscriptionStorage>::new(self.db.clone().into())
71    }
72
73    #[cfg(feature = "internal-oauth")]
74    fn auth(&self) -> Arc<dyn AuthSource> {
75        Arc::<PgAuthSource>::new(self.db.clone().into())
76    }
77
78    /// Verify the connection pool is open and has at least one connection
79    fn connection_active(&self) -> bool {
80        !self.db.is_closed() && self.db.size() > 0
81    }
82}
83
84#[async_trait]
85impl Migrate for PostgresStorage {
86    async fn migrate(&self) -> Result<(), MigrateError> {
87        sqlx::migrate!("./migrations").run(&self.db).await
88    }
89}
90
91impl PostgresStorage {
92    pub fn new(db: PgPool) -> Result<Self, sqlx::Error> {
93        Ok(Self { db })
94    }
95
96    pub async fn from_env() -> Result<Self, sqlx::Error> {
97        dotenv().ok();
98        let db_url = std::env::var("DATABASE_URL")
99            .expect("Missing DATABASE_URL env var even though the 'postgres' feature is active");
100
101        let db = PgPoolOptions::new()
102            .min_connections(1)
103            .connect(&db_url)
104            .await?;
105
106        let connect_options = db.connect_options();
107        let safe_db_url = format!(
108            "{}:{}/{}",
109            connect_options.get_host(),
110            connect_options.get_port(),
111            connect_options.get_database().unwrap_or_default()
112        );
113
114        Self::new(db)
115            .inspect_err(|err| error!(?err, "could not connect to Postgres database"))
116            .inspect(|_| {
117                info!(
118                    "Successfully connected to Postgres backend at {}",
119                    safe_db_url
120                )
121            })
122    }
123}
124
125fn to_json_value<T: Serialize>(v: Option<T>) -> Result<Option<serde_json::Value>, AppError> {
126    v.map(|v| serde_json::to_value(v).map_err(AppError::SerdeJsonBadRequest))
127        .transpose()
128}
129
130/// Returns the targets of the VEN associated with the given `client_id` and it's resources.
131/// If the VEN does not exist, returns an empty vector.
132async fn get_ven_targets(db: PgPool, client_id: &ClientId) -> Result<Vec<Target>, AppError> {
133    let ven_store: PgVenStorage = db.into();
134    match ven_store.targets_by_client_id(client_id).await {
135        Ok(t) => Ok(t),
136        // Cite from OpenADR Spec 3.1.1 Definition.md, "VEN created object privacy":
137        //      4. If a VEN object is not found, return objects that do not have targets and do not proceed to step 5.
138        //      [...]
139        //      6. If the union of the targets of the VEN and its resources is empty, return objects that do not have targets and do not proceed to step 7.
140        Err(AppError::NotFound) => Ok(Vec::new()),
141        Err(err) => Err(err),
142    }
143}
144
145fn intersection<'a>(a: &'a [Target], b: &'a [Target]) -> Vec<&'a Target> {
146    a.iter().filter(|x| b.contains(x)).collect()
147}
148
149#[cfg(test)]
150mod test {
151    use openleadr_wire::target::Target;
152    use std::str::FromStr;
153
154    #[test]
155    fn intersection() {
156        let t1 = Target::from_str("t1").unwrap();
157        let t2 = Target::from_str("t2").unwrap();
158        let t3 = Target::from_str("t3").unwrap();
159        let t4 = Target::from_str("t4").unwrap();
160
161        let a = vec![t1, t2.clone(), t3.clone()];
162        let b = vec![t2.clone(), t3.clone(), t4];
163
164        let i = super::intersection(&a, &b);
165        assert_eq!(i, vec![&t2, &t3]);
166    }
167}
openleadr_vtn/data_source/postgres/mod.rs

openleadr_vtn/data_source/postgres/
mod.rs
