7142165995

Committed 08 Dec 2023 01:12PM UTC coverage: 89.675% (+0.01%) from 89.665%

Build # 7142165995

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #895 from geo-engine/raster_colorizer_bands

add raster colorizer for bands

Run Details

722 of 783 new or added lines in 33 files covered. (92.21%)

13 existing lines in 10 files now uncovered.

113148 of 126176 relevant lines covered (89.67%)

59847.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.07

/services/src/pro/layers/postgres_layer_db.rs

use crate::error;
use crate::layers::external::TypedDataProviderDefinition;
use crate::layers::layer::Property;
use crate::layers::postgres_layer_db::{
    delete_layer_collection, delete_layer_collection_from_parent, delete_layer_from_collection,
    insert_collection_parent, insert_layer, insert_layer_collection_with_id,
};
use crate::pro::contexts::ProPostgresDb;
use crate::pro::datasets::TypedProDataProviderDefinition;
use crate::pro::permissions::postgres_permissiondb::TxPermissionDb;
use crate::pro::permissions::{Permission, RoleId};
use crate::{
    error::Result,
    layers::{
        external::{DataProvider, DataProviderDefinition},
        layer::{
            AddLayer, AddLayerCollection, CollectionItem, Layer, LayerCollection,
            LayerCollectionListOptions, LayerCollectionListing, LayerListing,
            ProviderLayerCollectionId, ProviderLayerId,
        },
        listing::{LayerCollectionId, LayerCollectionProvider},
        storage::{
            LayerDb, LayerProviderDb, LayerProviderListing, LayerProviderListingOptions,
            INTERNAL_LAYER_DB_ROOT_COLLECTION_ID, INTERNAL_PROVIDER_ID,
        },
        LayerDbError,
    },
};
use async_trait::async_trait;
use bb8_postgres::tokio_postgres::{
    tls::{MakeTlsConnect, TlsConnect},
    Socket,
};
use geoengine_datatypes::dataset::{DataProviderId, LayerId};
use geoengine_datatypes::util::HashMapTextTextDbType;
use snafu::{ensure, ResultExt};
use std::str::FromStr;
use uuid::Uuid;

#[async_trait]
impl<Tls> LayerDb for ProPostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    async fn add_layer(&self, layer: AddLayer, collection: &LayerCollectionId) -> Result<LayerId> {
        let layer_id = Uuid::new_v4();
        let layer_id = LayerId(layer_id.to_string());

        self.add_layer_with_id(&layer_id, layer, collection).await?;

        Ok(layer_id)
    }

    async fn add_layer_with_id(
        &self,
        id: &LayerId,
        layer: AddLayer,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        let mut conn = self.conn_pool.get().await?;
        let trans = conn.build_transaction().start().await?;

        ensure!(
            self.has_permission_in_tx(collection.clone(), Permission::Owner, &trans)
                .await?,
            error::PermissionDenied
        );

        let layer_id = insert_layer(&trans, id, layer, collection).await?;

        // TODO: `ON CONFLICT DO NOTHING` means, we do not get an error if the permission already exists.
        //       Do we want that, or should we report an error and let the caller decide whether to ignore it?
        //       We should decide that and adjust all places where `ON CONFLICT DO NOTHING` is used.
        let stmt = trans
            .prepare(
                "
            INSERT INTO permissions (role_id, permission, layer_id)
            VALUES ($1, $2, $3) ON CONFLICT DO NOTHING;",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &RoleId::from(self.session.user.id),
                    &Permission::Owner,
                    &layer_id,
                ],
            )
            .await?;

        trans.commit().await?;

        Ok(())
    }

    async fn add_layer_to_collection(
        &self,
        layer: &LayerId,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        let mut conn = self.conn_pool.get().await?;
        let tx = conn.build_transaction().start().await?;

        ensure!(
            self.has_permission_in_tx(collection.clone(), Permission::Owner, &tx)
                .await?,
            error::PermissionDenied
        );

        let layer_id =
            Uuid::from_str(&layer.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: layer.0.clone(),
            })?;

        let collection_id =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        let stmt = tx
            .prepare(
                "
            INSERT INTO collection_layers (collection, layer)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        tx.execute(&stmt, &[&collection_id, &layer_id]).await?;

        tx.commit().await?;

        Ok(())
    }

    async fn add_layer_collection(
        &self,
        collection: AddLayerCollection,
        parent: &LayerCollectionId,
    ) -> Result<LayerCollectionId> {
        let collection_id = Uuid::new_v4();
        let collection_id = LayerCollectionId(collection_id.to_string());

        self.add_layer_collection_with_id(&collection_id, collection, parent)
            .await?;

        Ok(collection_id)
    }

    async fn add_layer_collection_with_id(
        &self,
        id: &LayerCollectionId,
        collection: AddLayerCollection,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let mut conn = self.conn_pool.get().await?;
        let trans = conn.build_transaction().start().await?;

        ensure!(
            self.has_permission_in_tx(parent.clone(), Permission::Owner, &trans)
                .await?,
            error::PermissionDenied
        );

        let collection_id = insert_layer_collection_with_id(&trans, id, collection, parent).await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO permissions (role_id, permission, layer_collection_id)
            VALUES ($1, $2, $3) ON CONFLICT DO NOTHING;",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &RoleId::from(self.session.user.id),
                    &Permission::Owner,
                    &collection_id,
                ],
            )
            .await?;

        trans.commit().await?;

        Ok(())
    }

    async fn add_collection_to_parent(
        &self,
        collection: &LayerCollectionId,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let conn = self.conn_pool.get().await?;
        insert_collection_parent(&conn, collection, parent).await
    }

    async fn remove_layer_collection(&self, collection: &LayerCollectionId) -> Result<()> {
        let mut conn = self.conn_pool.get().await?;
        let transaction = conn.build_transaction().start().await?;

        ensure!(
            self.has_permission_in_tx(collection.clone(), Permission::Owner, &transaction)
                .await?,
            error::PermissionDenied
        );

        delete_layer_collection(&transaction, collection).await?;

        transaction.commit().await.map_err(Into::into)
    }

    async fn remove_layer_from_collection(
        &self,
        layer: &LayerId,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        let mut conn = self.conn_pool.get().await?;
        let transaction = conn.build_transaction().start().await?;

        ensure!(
            self.has_permission_in_tx(layer.clone(), Permission::Owner, &transaction)
                .await?,
            error::PermissionDenied
        );

        delete_layer_from_collection(&transaction, layer, collection).await?;

        transaction.commit().await.map_err(Into::into)
    }

    async fn remove_layer_collection_from_parent(
        &self,
        collection: &LayerCollectionId,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let mut conn = self.conn_pool.get().await?;
        let transaction = conn.build_transaction().start().await?;

        ensure!(
            self.has_permission_in_tx(collection.clone(), Permission::Owner, &transaction)
                .await?,
            error::PermissionDenied
        );

        delete_layer_collection_from_parent(&transaction, collection, parent).await?;

        transaction.commit().await.map_err(Into::into)
    }
}

#[async_trait]
impl<Tls> LayerCollectionProvider for ProPostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    #[allow(clippy::too_many_lines)]
    async fn load_layer_collection(
        &self,
        collection_id: &LayerCollectionId,
        options: LayerCollectionListOptions,
    ) -> Result<LayerCollection> {
        let mut conn = self.conn_pool.get().await?;
        let tx = conn.build_transaction().start().await?;

        ensure!(
            self.has_permission_in_tx(collection_id.clone(), Permission::Read, &tx)
                .await?,
            error::PermissionDenied
        );
        let collection = Uuid::from_str(&collection_id.0).map_err(|_| {
            crate::error::Error::IdStringMustBeUuid {
                found: collection_id.0.clone(),
            }
        })?;

        let stmt = tx
            .prepare(
                "
        SELECT DISTINCT name, description, properties
        FROM user_permitted_layer_collections p 
            JOIN layer_collections c ON (p.layer_collection_id = c.id) 
        WHERE p.user_id = $1 AND layer_collection_id = $2;",
            )
            .await?;

        let row = tx
            .query_one(&stmt, &[&self.session.user.id, &collection])
            .await?;

        let name: String = row.get(0);
        let description: String = row.get(1);
        let properties: Vec<Property> = row.get(2);

        let stmt = tx
            .prepare(
                "
        SELECT DISTINCT id, name, description, properties, is_layer
        FROM (
            SELECT 
                concat(id, '') AS id, 
                name, 
                description, 
                properties, 
                FALSE AS is_layer
            FROM user_permitted_layer_collections u 
                JOIN layer_collections lc ON (u.layer_collection_id = lc.id)
                JOIN collection_children cc ON (layer_collection_id = cc.child)
            WHERE u.user_id = $4 AND cc.parent = $1
        ) u UNION (
            SELECT 
                concat(id, '') AS id, 
                name, 
                description, 
                properties, 
                TRUE AS is_layer
            FROM user_permitted_layers ul
                JOIN layers uc ON (ul.layer_id = uc.id) 
                JOIN collection_layers cl ON (layer_id = cl.layer)
            WHERE ul.user_id = $4 AND cl.collection = $1
        )
        ORDER BY is_layer ASC, name ASC
        LIMIT $2 
        OFFSET $3;            
        ",
            )
            .await?;

        let rows = tx
            .query(
                &stmt,
                &[
                    &collection,
                    &i64::from(options.limit),
                    &i64::from(options.offset),
                    &self.session.user.id,
                ],
            )
            .await?;

        let items = rows
            .into_iter()
            .map(|row| {
                let is_layer: bool = row.get(4);

                if is_layer {
                    Ok(CollectionItem::Layer(LayerListing {
                        id: ProviderLayerId {
                            provider_id: INTERNAL_PROVIDER_ID,
                            layer_id: LayerId(row.get(0)),
                        },
                        name: row.get(1),
                        description: row.get(2),
                        properties: row.get(3),
                    }))
                } else {
                    Ok(CollectionItem::Collection(LayerCollectionListing {
                        id: ProviderLayerCollectionId {
                            provider_id: INTERNAL_PROVIDER_ID,
                            collection_id: LayerCollectionId(row.get(0)),
                        },
                        name: row.get(1),
                        description: row.get(2),
                        properties: row.get(3),
                    }))
                }
            })
            .collect::<Result<Vec<CollectionItem>>>()?;

        tx.commit().await?;

        Ok(LayerCollection {
            id: ProviderLayerCollectionId {
                provider_id: INTERNAL_PROVIDER_ID,
                collection_id: collection_id.clone(),
            },
            name,
            description,
            items,
            entry_label: None,
            properties,
        })
    }

    async fn get_root_layer_collection_id(&self) -> Result<LayerCollectionId> {
        Ok(LayerCollectionId(
            INTERNAL_LAYER_DB_ROOT_COLLECTION_ID.to_string(),
        ))
    }

    async fn load_layer(&self, id: &LayerId) -> Result<Layer> {
        let mut conn = self.conn_pool.get().await?;
        let tx = conn.build_transaction().start().await?;

        ensure!(
            self.has_permission_in_tx(id.clone(), Permission::Read, &tx)
                .await?,
            error::PermissionDenied
        );

        let layer_id =
            Uuid::from_str(&id.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: id.0.clone(),
            })?;

        let stmt = tx
            .prepare(
                "
            SELECT 
                l.name,
                l.description,
                w.workflow,
                l.symbology,
                l.properties,
                l.metadata
            FROM 
                layers l JOIN workflows w ON (l.workflow_id = w.id)
            WHERE 
                l.id = $1;",
            )
            .await?;

        let row = tx
            .query_one(&stmt, &[&layer_id])
            .await
            .map_err(|_error| LayerDbError::NoLayerForGivenId { id: id.clone() })?;

        tx.commit().await?;

        Ok(Layer {
            id: ProviderLayerId {
                provider_id: INTERNAL_PROVIDER_ID,
                layer_id: id.clone(),
            },
            name: row.get(0),
            description: row.get(1),
            workflow: serde_json::from_value(row.get(2)).context(crate::error::SerdeJson)?,
            symbology: row.get(3),
            properties: row.get(4),
            metadata: row.get::<_, HashMapTextTextDbType>(5).into(),
        })
    }
}

#[async_trait]
impl<Tls> LayerProviderDb for ProPostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    async fn add_layer_provider(
        &self,
        provider: TypedDataProviderDefinition,
    ) -> Result<DataProviderId> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
              INSERT INTO layer_providers (
                  id, 
                  type_name, 
                  name,
                  definition
              )
              VALUES ($1, $2, $3, $4)",
            )
            .await?;

        let id = provider.id();
        conn.execute(
            &stmt,
            &[&id, &provider.type_name(), &provider.name(), &provider],
        )
        .await?;
        Ok(id)
    }

    async fn list_layer_providers(
        &self,
        options: LayerProviderListingOptions,
    ) -> Result<Vec<LayerProviderListing>> {
        // TODO: permission
        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "(
                    SELECT 
                        id, 
                        name,
                        type_name
                    FROM 
                        layer_providers
                    UNION ALL
                    SELECT 
                        id, 
                        name,
                        type_name
                    FROM 
                        pro_layer_providers
                )
                ORDER BY name ASC
                LIMIT $1 
                OFFSET $2;",
            )
            .await?;

        let rows = conn
            .query(
                &stmt,
                &[&i64::from(options.limit), &i64::from(options.offset)],
            )
            .await?;

        Ok(rows
            .iter()
            .map(|row| LayerProviderListing {
                id: row.get(0),
                name: row.get(1),
                description: row.get(2),
            })
            .collect())
    }

    async fn load_layer_provider(&self, id: DataProviderId) -> Result<Box<dyn DataProvider>> {
        // TODO: permissions
        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "SELECT
                    definition, NULL AS pro_definition
                FROM
                    layer_providers
                WHERE
                    id = $1
                UNION ALL
                SELECT
                    NULL AS definition, definition AS pro_definition
                FROM
                    pro_layer_providers
                WHERE
                    id = $1",
            )
            .await?;

        let row = conn.query_one(&stmt, &[&id]).await?;

        if let Some(definition) = row.get::<_, Option<TypedDataProviderDefinition>>(0) {
            return Box::new(definition).initialize().await;
        }

        let pro_definition: TypedProDataProviderDefinition = row.get(1);
        Box::new(pro_definition).initialize().await
    }
}

#[async_trait]
pub trait ProLayerProviderDb: Send + Sync + 'static {
    async fn add_pro_layer_provider(
        &self,
        provider: TypedProDataProviderDefinition,
    ) -> Result<DataProviderId>;
}

#[async_trait]
impl<Tls> ProLayerProviderDb for ProPostgresDb<Tls>
where
    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
{
    async fn add_pro_layer_provider(
        &self,
        provider: TypedProDataProviderDefinition,
    ) -> Result<DataProviderId> {
        ensure!(self.session.is_admin(), error::PermissionDenied);

        let conn = self.conn_pool.get().await?;

        let stmt = conn
            .prepare(
                "
              INSERT INTO pro_layer_providers (
                  id, 
                  type_name, 
                  name,
                  definition
              )
              VALUES ($1, $2, $3, $4)",
            )
            .await?;

        let id = provider.id();
        conn.execute(
            &stmt,
            &[&id, &provider.type_name(), &provider.name(), &provider],
        )
        .await?;
        Ok(id)
    }
}

1	use crate::error;
2	use crate::layers::external::TypedDataProviderDefinition;
3	use crate::layers::layer::Property;
4	use crate::layers::postgres_layer_db::{
5	delete_layer_collection, delete_layer_collection_from_parent, delete_layer_from_collection,
6	insert_collection_parent, insert_layer, insert_layer_collection_with_id,
7	};
8	use crate::pro::contexts::ProPostgresDb;
9	use crate::pro::datasets::TypedProDataProviderDefinition;
10	use crate::pro::permissions::postgres_permissiondb::TxPermissionDb;
11	use crate::pro::permissions::{Permission, RoleId};
12	use crate::{
13	error::Result,
14	layers::{
15	external::{DataProvider, DataProviderDefinition},
16	layer::{
17	AddLayer, AddLayerCollection, CollectionItem, Layer, LayerCollection,
18	LayerCollectionListOptions, LayerCollectionListing, LayerListing,
19	ProviderLayerCollectionId, ProviderLayerId,
20	},
21	listing::{LayerCollectionId, LayerCollectionProvider},
22	storage::{
23	LayerDb, LayerProviderDb, LayerProviderListing, LayerProviderListingOptions,
24	INTERNAL_LAYER_DB_ROOT_COLLECTION_ID, INTERNAL_PROVIDER_ID,
25	},
26	LayerDbError,
27	},
28	};
29	use async_trait::async_trait;
30	use bb8_postgres::tokio_postgres::{
31	tls::{MakeTlsConnect, TlsConnect},
32	Socket,
33	};
34	use geoengine_datatypes::dataset::{DataProviderId, LayerId};
35	use geoengine_datatypes::util::HashMapTextTextDbType;
36	use snafu::{ensure, ResultExt};
37	use std::str::FromStr;
38	use uuid::Uuid;
39
40	#[async_trait]
41	impl<Tls> LayerDb for ProPostgresDb<Tls>
42	where
43	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
44	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
45	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
46	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
47	{
48	async fn add_layer(&self, layer: AddLayer, collection: &LayerCollectionId) -> Result<LayerId> {	6✔
49	let layer_id = Uuid::new_v4();	6✔
50	let layer_id = LayerId(layer_id.to_string());	6✔
51		6✔
52	self.add_layer_with_id(&layer_id, layer, collection).await?;	219✔
53
54	Ok(layer_id)	6✔
55	}	12✔
56
57	async fn add_layer_with_id(	6✔
58	&self,	6✔
59	id: &LayerId,	6✔
60	layer: AddLayer,	6✔
61	collection: &LayerCollectionId,	6✔
62	) -> Result<()> {	6✔
63	let mut conn = self.conn_pool.get().await?;	6✔
64	let trans = conn.build_transaction().start().await?;	6✔
65
66	ensure!(	6✔
67	self.has_permission_in_tx(collection.clone(), Permission::Owner, &trans)	6✔
68	.await?,	17✔
69	error::PermissionDenied	×
70	);
71
72	let layer_id = insert_layer(&trans, id, layer, collection).await?;	172✔
73
74	// TODO: `ON CONFLICT DO NOTHING` means, we do not get an error if the permission already exists.
75	// Do we want that, or should we report an error and let the caller decide whether to ignore it?
76	// We should decide that and adjust all places where `ON CONFLICT DO NOTHING` is used.
77	let stmt = trans	6✔
78	.prepare(	6✔
79	"	6✔
80	INSERT INTO permissions (role_id, permission, layer_id)	6✔
81	VALUES ($1, $2, $3) ON CONFLICT DO NOTHING;",	6✔
82	)	6✔
83	.await?;	6✔
84
85	trans	6✔
86	.execute(	6✔
87	&stmt,	6✔
88	&[	6✔
89	&RoleId::from(self.session.user.id),	6✔
90	&Permission::Owner,	6✔
91	&layer_id,	6✔
92	],	6✔
93	)	6✔
94	.await?;	6✔
95
96	trans.commit().await?;	6✔
97
98	Ok(())	6✔
99	}	12✔
100
101	async fn add_layer_to_collection(	2✔
102	&self,	2✔
103	layer: &LayerId,	2✔
104	collection: &LayerCollectionId,	2✔
105	) -> Result<()> {	2✔
106	let mut conn = self.conn_pool.get().await?;	2✔
107	let tx = conn.build_transaction().start().await?;	2✔
108
109	ensure!(	2✔
110	self.has_permission_in_tx(collection.clone(), Permission::Owner, &tx)	2✔
111	.await?,	4✔
112	error::PermissionDenied	×
113	);
114
115	let layer_id =	1✔
116	Uuid::from_str(&layer.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
117	found: layer.0.clone(),	1✔
118	})?;	2✔
119
120	let collection_id =	1✔
121	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	1✔
122	found: collection.0.clone(),	×
123	})?;	1✔
124
125	let stmt = tx	1✔
126	.prepare(	1✔
127	"	1✔
128	INSERT INTO collection_layers (collection, layer)	1✔
129	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	1✔
130	)	1✔
131	.await?;	1✔
132
133	tx.execute(&stmt, &[&collection_id, &layer_id]).await?;	1✔
134
135	tx.commit().await?;	1✔
136
137	Ok(())	1✔
138	}	4✔
139
140	async fn add_layer_collection(	12✔
141	&self,	12✔
142	collection: AddLayerCollection,	12✔
143	parent: &LayerCollectionId,	12✔
144	) -> Result<LayerCollectionId> {	12✔
145	let collection_id = Uuid::new_v4();	12✔
146	let collection_id = LayerCollectionId(collection_id.to_string());	12✔
147		12✔
148	self.add_layer_collection_with_id(&collection_id, collection, parent)	12✔
149	.await?;	156✔
150
151	Ok(collection_id)	10✔
152	}	24✔
153
154	async fn add_layer_collection_with_id(	12✔
155	&self,	12✔
156	id: &LayerCollectionId,	12✔
157	collection: AddLayerCollection,	12✔
158	parent: &LayerCollectionId,	12✔
159	) -> Result<()> {	12✔
160	let mut conn = self.conn_pool.get().await?;	12✔
161	let trans = conn.build_transaction().start().await?;	12✔
162
163	ensure!(	12✔
164	self.has_permission_in_tx(parent.clone(), Permission::Owner, &trans)	12✔
165	.await?,	46✔
166	error::PermissionDenied	2✔
167	);
168
169	let collection_id = insert_layer_collection_with_id(&trans, id, collection, parent).await?;	56✔
170
171	let stmt = trans	10✔
172	.prepare(	10✔
173	"	10✔
174	INSERT INTO permissions (role_id, permission, layer_collection_id)	10✔
175	VALUES ($1, $2, $3) ON CONFLICT DO NOTHING;",	10✔
176	)	10✔
177	.await?;	10✔
178
179	trans	10✔
180	.execute(	10✔
181	&stmt,	10✔
182	&[	10✔
183	&RoleId::from(self.session.user.id),	10✔
184	&Permission::Owner,	10✔
185	&collection_id,	10✔
186	],	10✔
187	)	10✔
188	.await?;	10✔
189
190	trans.commit().await?;	10✔
191
192	Ok(())	10✔
193	}	24✔
194
195	async fn add_collection_to_parent(	1✔
196	&self,	1✔
197	collection: &LayerCollectionId,	1✔
198	parent: &LayerCollectionId,	1✔
199	) -> Result<()> {	1✔
200	let conn = self.conn_pool.get().await?;	1✔
201	insert_collection_parent(&conn, collection, parent).await	2✔
202	}	2✔
203
204	async fn remove_layer_collection(&self, collection: &LayerCollectionId) -> Result<()> {	3✔
205	let mut conn = self.conn_pool.get().await?;	3✔
206	let transaction = conn.build_transaction().start().await?;	3✔
207
208	ensure!(	3✔
209	self.has_permission_in_tx(collection.clone(), Permission::Owner, &transaction)	3✔
210	.await?,	6✔
211	error::PermissionDenied	×
212	);
213
214	delete_layer_collection(&transaction, collection).await?;	12✔
215
216	transaction.commit().await.map_err(Into::into)	2✔
217	}	6✔
218
219	async fn remove_layer_from_collection(	2✔
220	&self,	2✔
221	layer: &LayerId,	2✔
222	collection: &LayerCollectionId,	2✔
223	) -> Result<()> {	2✔
224	let mut conn = self.conn_pool.get().await?;	2✔
225	let transaction = conn.build_transaction().start().await?;	2✔
226
227	ensure!(	2✔
228	self.has_permission_in_tx(layer.clone(), Permission::Owner, &transaction)	2✔
229	.await?,	4✔
230	error::PermissionDenied	×
231	);
232
233	delete_layer_from_collection(&transaction, layer, collection).await?;	8✔
234
235	transaction.commit().await.map_err(Into::into)	2✔
236	}	4✔
237
238	async fn remove_layer_collection_from_parent(	1✔
239	&self,	1✔
240	collection: &LayerCollectionId,	1✔
241	parent: &LayerCollectionId,	1✔
242	) -> Result<()> {	1✔
243	let mut conn = self.conn_pool.get().await?;	1✔
244	let transaction = conn.build_transaction().start().await?;	1✔
245
246	ensure!(	1✔
247	self.has_permission_in_tx(collection.clone(), Permission::Owner, &transaction)	1✔
248	.await?,	2✔
249	error::PermissionDenied	×
250	);
251
252	delete_layer_collection_from_parent(&transaction, collection, parent).await?;	8✔
253
254	transaction.commit().await.map_err(Into::into)	1✔
255	}	2✔
256	}
257
258	#[async_trait]
259	impl<Tls> LayerCollectionProvider for ProPostgresDb<Tls>
260	where
261	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
262	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
263	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
264	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
265	{
266	#[allow(clippy::too_many_lines)]
267	async fn load_layer_collection(	15✔
268	&self,	15✔
269	collection_id: &LayerCollectionId,	15✔
270	options: LayerCollectionListOptions,	15✔
271	) -> Result<LayerCollection> {	15✔
272	let mut conn = self.conn_pool.get().await?;	15✔
273	let tx = conn.build_transaction().start().await?;	15✔
274
275	ensure!(	15✔
276	self.has_permission_in_tx(collection_id.clone(), Permission::Read, &tx)	15✔
277	.await?,	30✔
278	error::PermissionDenied	3✔
279	);
280	let collection = Uuid::from_str(&collection_id.0).map_err(\|_\| {	12✔
281	crate::error::Error::IdStringMustBeUuid {	×
282	found: collection_id.0.clone(),	×
283	}	×
284	})?;	12✔
285
286	let stmt = tx	12✔
287	.prepare(	12✔
288	"	12✔
289	SELECT DISTINCT name, description, properties	12✔
290	FROM user_permitted_layer_collections p	12✔
291	JOIN layer_collections c ON (p.layer_collection_id = c.id)	12✔
292	WHERE p.user_id = $1 AND layer_collection_id = $2;",	12✔
293	)	12✔
294	.await?;	12✔
295
296	let row = tx	12✔
297	.query_one(&stmt, &[&self.session.user.id, &collection])	12✔
298	.await?;	12✔
299
300	let name: String = row.get(0);	12✔
301	let description: String = row.get(1);	12✔
302	let properties: Vec<Property> = row.get(2);	12✔
303
304	let stmt = tx	12✔
305	.prepare(	12✔
306	"	12✔
307	SELECT DISTINCT id, name, description, properties, is_layer	12✔
308	FROM (	12✔
309	SELECT	12✔
310	concat(id, '') AS id,	12✔
311	name,	12✔
312	description,	12✔
313	properties,	12✔
314	FALSE AS is_layer	12✔
315	FROM user_permitted_layer_collections u	12✔
316	JOIN layer_collections lc ON (u.layer_collection_id = lc.id)	12✔
317	JOIN collection_children cc ON (layer_collection_id = cc.child)	12✔
318	WHERE u.user_id = $4 AND cc.parent = $1	12✔
319	) u UNION (	12✔
320	SELECT	12✔
321	concat(id, '') AS id,	12✔
322	name,	12✔
323	description,	12✔
324	properties,	12✔
325	TRUE AS is_layer	12✔
326	FROM user_permitted_layers ul	12✔
327	JOIN layers uc ON (ul.layer_id = uc.id)	12✔
328	JOIN collection_layers cl ON (layer_id = cl.layer)	12✔
329	WHERE ul.user_id = $4 AND cl.collection = $1	12✔
330	)	12✔
331	ORDER BY is_layer ASC, name ASC	12✔
332	LIMIT $2	12✔
333	OFFSET $3;	12✔
334	",	12✔
335	)	12✔
336	.await?;	12✔
337
338	let rows = tx	12✔
339	.query(	12✔
340	&stmt,	12✔
341	&[	12✔
342	&collection,	12✔
343	&i64::from(options.limit),	12✔
344	&i64::from(options.offset),	12✔
345	&self.session.user.id,	12✔
346	],	12✔
347	)	12✔
348	.await?;	12✔
349
350	let items = rows	12✔
351	.into_iter()	12✔
352	.map(\|row\| {	15✔
353	let is_layer: bool = row.get(4);	15✔
354		15✔
355	if is_layer {	15✔
356	Ok(CollectionItem::Layer(LayerListing {	5✔
357	id: ProviderLayerId {	5✔
358	provider_id: INTERNAL_PROVIDER_ID,	5✔
359	layer_id: LayerId(row.get(0)),	5✔
360	},	5✔
361	name: row.get(1),	5✔
362	description: row.get(2),	5✔
363	properties: row.get(3),	5✔
364	}))	5✔
365	} else {
366	Ok(CollectionItem::Collection(LayerCollectionListing {	10✔
367	id: ProviderLayerCollectionId {	10✔
368	provider_id: INTERNAL_PROVIDER_ID,	10✔
369	collection_id: LayerCollectionId(row.get(0)),	10✔
370	},	10✔
371	name: row.get(1),	10✔
372	description: row.get(2),	10✔
373	properties: row.get(3),	10✔
374	}))	10✔
375	}
376	})	15✔
377	.collect::<Result<Vec<CollectionItem>>>()?;	12✔
378
379	tx.commit().await?;	12✔
380
381	Ok(LayerCollection {	12✔
382	id: ProviderLayerCollectionId {	12✔
383	provider_id: INTERNAL_PROVIDER_ID,	12✔
384	collection_id: collection_id.clone(),	12✔
385	},	12✔
386	name,	12✔
387	description,	12✔
388	items,	12✔
389	entry_label: None,	12✔
390	properties,	12✔
391	})	12✔
392	}	30✔
393
394	async fn get_root_layer_collection_id(&self) -> Result<LayerCollectionId> {	6✔
395	Ok(LayerCollectionId(	6✔
396	INTERNAL_LAYER_DB_ROOT_COLLECTION_ID.to_string(),	6✔
397	))	6✔
398	}	6✔
399
400	async fn load_layer(&self, id: &LayerId) -> Result<Layer> {	6✔
401	let mut conn = self.conn_pool.get().await?;	6✔
402	let tx = conn.build_transaction().start().await?;	6✔
403
404	ensure!(	6✔
405	self.has_permission_in_tx(id.clone(), Permission::Read, &tx)	6✔
406	.await?,	12✔
407	error::PermissionDenied	3✔
408	);
409
410	let layer_id =	3✔
411	Uuid::from_str(&id.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	3✔
412	found: id.0.clone(),	×
413	})?;	3✔
414
415	let stmt = tx	3✔
416	.prepare(	3✔
417	"	3✔
418	SELECT	3✔
419	l.name,	3✔
420	l.description,	3✔
421	w.workflow,	3✔
422	l.symbology,	3✔
423	l.properties,	3✔
424	l.metadata	3✔
425	FROM	3✔
426	layers l JOIN workflows w ON (l.workflow_id = w.id)	3✔
427	WHERE	3✔
428	l.id = $1;",	3✔
429	)	3✔
430	.await?;	3✔
431
432	let row = tx	3✔
433	.query_one(&stmt, &[&layer_id])	3✔
434	.await	3✔
435	.map_err(\|_error\| LayerDbError::NoLayerForGivenId { id: id.clone() })?;	3✔
436
437	tx.commit().await?;	3✔
438
439	Ok(Layer {
440	id: ProviderLayerId {	3✔
441	provider_id: INTERNAL_PROVIDER_ID,	3✔
442	layer_id: id.clone(),	3✔
443	},	3✔
444	name: row.get(0),	3✔
445	description: row.get(1),	3✔
446	workflow: serde_json::from_value(row.get(2)).context(crate::error::SerdeJson)?,	3✔
447	symbology: row.get(3),	3✔
448	properties: row.get(4),	3✔
449	metadata: row.get::<_, HashMapTextTextDbType>(5).into(),	3✔
450	})
451	}	12✔
452	}
453
454	#[async_trait]
455	impl<Tls> LayerProviderDb for ProPostgresDb<Tls>
456	where
457	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
458	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
459	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
460	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
461	{
462	async fn add_layer_provider(	1✔
463	&self,	1✔
464	provider: TypedDataProviderDefinition,	1✔
465	) -> Result<DataProviderId> {	1✔
466	ensure!(self.session.is_admin(), error::PermissionDenied);	1✔
467
468	let conn = self.conn_pool.get().await?;	1✔
469
470	let stmt = conn	1✔
471	.prepare(	1✔
472	"	1✔
473	INSERT INTO layer_providers (	1✔
474	id,	1✔
475	type_name,	1✔
476	name,	1✔
477	definition	1✔
478	)	1✔
479	VALUES ($1, $2, $3, $4)",	1✔
480	)	1✔
481	.await?;	2✔
482
483	let id = provider.id();	1✔
484	conn.execute(	1✔
485	&stmt,	1✔
486	&[&id, &provider.type_name(), &provider.name(), &provider],	1✔
487	)	1✔
488	.await?;	1✔
489	Ok(id)	1✔
490	}	2✔
491
492	async fn list_layer_providers(	2✔
493	&self,	2✔
494	options: LayerProviderListingOptions,	2✔
495	) -> Result<Vec<LayerProviderListing>> {	2✔
496	// TODO: permission
497	let conn = self.conn_pool.get().await?;	2✔
498
499	let stmt = conn	2✔
500	.prepare(	2✔
501	"(	2✔
502	SELECT	2✔
503	id,	2✔
504	name,	2✔
505	type_name	2✔
506	FROM	2✔
507	layer_providers	2✔
508	UNION ALL	2✔
509	SELECT	2✔
510	id,	2✔
511	name,	2✔
512	type_name	2✔
513	FROM	2✔
514	pro_layer_providers	2✔
515	)	2✔
516	ORDER BY name ASC	2✔
517	LIMIT $1	2✔
518	OFFSET $2;",	2✔
519	)	2✔
UNCOV 520	.await?;	×
521
522	let rows = conn	2✔
523	.query(	2✔
524	&stmt,	2✔
525	&[&i64::from(options.limit), &i64::from(options.offset)],	2✔
526	)	2✔
UNCOV 527	.await?;	×
528
529	Ok(rows	2✔
530	.iter()	2✔
531	.map(\|row\| LayerProviderListing {	2✔
532	id: row.get(0),	2✔
533	name: row.get(1),	2✔
534	description: row.get(2),	2✔
535	})	2✔
536	.collect())	2✔
537	}	4✔
538
539	async fn load_layer_provider(&self, id: DataProviderId) -> Result<Box<dyn DataProvider>> {	2✔
540	// TODO: permissions
541	let conn = self.conn_pool.get().await?;	2✔
542
543	let stmt = conn	2✔
544	.prepare(	2✔
545	"SELECT	2✔
546	definition, NULL AS pro_definition	2✔
547	FROM	2✔
548	layer_providers	2✔
549	WHERE	2✔
550	id = $1	2✔
551	UNION ALL	2✔
552	SELECT	2✔
553	NULL AS definition, definition AS pro_definition	2✔
554	FROM	2✔
555	pro_layer_providers	2✔
556	WHERE	2✔
557	id = $1",	2✔
558	)	2✔
UNCOV 559	.await?;	×
560
561	let row = conn.query_one(&stmt, &[&id]).await?;	2✔
562
563	if let Some(definition) = row.get::<_, Option<TypedDataProviderDefinition>>(0) {	2✔
564	return Box::new(definition).initialize().await;	1✔
565	}	1✔
566		1✔
567	let pro_definition: TypedProDataProviderDefinition = row.get(1);	1✔
568	Box::new(pro_definition).initialize().await	1✔
569	}	4✔
570	}
571
572	#[async_trait]
573	pub trait ProLayerProviderDb: Send + Sync + 'static {
574	async fn add_pro_layer_provider(
575	&self,
576	provider: TypedProDataProviderDefinition,
577	) -> Result<DataProviderId>;
578	}
579
580	#[async_trait]
581	impl<Tls> ProLayerProviderDb for ProPostgresDb<Tls>
582	where
583	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
584	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
585	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
586	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
587	{
588	async fn add_pro_layer_provider(	1✔
589	&self,	1✔
590	provider: TypedProDataProviderDefinition,	1✔
591	) -> Result<DataProviderId> {	1✔
592	ensure!(self.session.is_admin(), error::PermissionDenied);	1✔
593
594	let conn = self.conn_pool.get().await?;	1✔
595
596	let stmt = conn	1✔
597	.prepare(	1✔
598	"	1✔
599	INSERT INTO pro_layer_providers (	1✔
600	id,	1✔
601	type_name,	1✔
602	name,	1✔
603	definition	1✔
604	)	1✔
605	VALUES ($1, $2, $3, $4)",	1✔
606	)	1✔
607	.await?;	6✔
608
609	let id = provider.id();	1✔
610	conn.execute(	1✔
611	&stmt,	1✔
612	&[&id, &provider.type_name(), &provider.name(), &provider],	1✔
613	)	1✔
614	.await?;	1✔
615	Ok(id)	1✔
616	}	2✔
617	}

geo-engine / geoengine / 7142165995

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous