5624216179

Committed 21 Jul 2023 03:05PM UTC coverage: 89.186% (+0.03%) from 89.156%

Build # 5624216179

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #831 from geo-engine/free-postgres

Free-postgres

Run Details

4197 of 4197 new or added lines in 44 files covered. (100.0%)

105664 of 118476 relevant lines covered (89.19%)

61107.62 hits per line

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

80.2

/services/src/layers/postgres_layer_db.rs

use crate::api::model::datatypes::{DataProviderId, LayerId};

use crate::contexts::PostgresDb;
use crate::layers::layer::Property;

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 snafu::ResultExt;
use std::str::FromStr;
use uuid::Uuid;

/// delete all collections without parent collection
async fn _remove_collections_without_parent_collection(
    transaction: &tokio_postgres::Transaction<'_>,
) -> Result<()> {
    // HINT: a recursive delete statement seems reasonable, but hard to implement in postgres
    //       because you have a graph with potential loops

    let remove_layer_collections_without_parents_stmt = transaction
        .prepare(
            "DELETE FROM layer_collections
                 WHERE  id <> $1 -- do not delete root collection
                 AND    id NOT IN (
                    SELECT child FROM collection_children
                 );",
        )
        .await?;
    while 0 < transaction
        .execute(
            &remove_layer_collections_without_parents_stmt,
            &[&INTERNAL_LAYER_DB_ROOT_COLLECTION_ID],
        )
        .await?
    {
        // whenever one collection is deleted, we have to check again if there are more
        // collections without parents
    }

    Ok(())
}

/// delete all layers without parent collection
async fn _remove_layers_without_parent_collection(
    transaction: &tokio_postgres::Transaction<'_>,
) -> Result<()> {
    let remove_layers_without_parents_stmt = transaction
        .prepare(
            "DELETE FROM layers
                 WHERE id NOT IN (
                    SELECT layer FROM collection_layers
                 );",
        )
        .await?;
    transaction
        .execute(&remove_layers_without_parents_stmt, &[])
        .await?;

    Ok(())
}

#[async_trait]
impl<Tls> LayerDb for PostgresDb<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 collection_id =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

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

        let layer = layer;

        let layer_id = Uuid::new_v4();
        let symbology = serde_json::to_value(&layer.symbology).context(crate::error::SerdeJson)?;

        let metadata = serde_json::to_value(&layer.metadata).context(crate::error::SerdeJson)?;

        let trans = conn.build_transaction().start().await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO layers (id, name, description, workflow, symbology, properties, metadata)
            VALUES ($1, $2, $3, $4, $5, $6, $7);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &layer_id,
                    &layer.name,
                    &layer.description,
                    &serde_json::to_value(&layer.workflow).context(crate::error::SerdeJson)?,
                    &symbology,
                    &layer.properties,
                    &metadata,
                ],
            )
            .await?;

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

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

        trans.commit().await?;

        Ok(LayerId(layer_id.to_string()))
    }

    async fn add_layer_with_id(
        &self,
        id: &LayerId,
        layer: AddLayer,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        let layer_id =
            Uuid::from_str(&id.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

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

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

        let layer = layer;

        let symbology = serde_json::to_value(&layer.symbology).context(crate::error::SerdeJson)?;

        let metadata = serde_json::to_value(&layer.metadata).context(crate::error::SerdeJson)?;

        let trans = conn.build_transaction().start().await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO layers (id, name, description, workflow, symbology, properties, metadata)
            VALUES ($1, $2, $3, $4, $5, $6, $7);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &layer_id,
                    &layer.name,
                    &layer.description,
                    &serde_json::to_value(&layer.workflow).context(crate::error::SerdeJson)?,
                    &symbology,
                    &layer.properties,
                    &metadata,
                ],
            )
            .await?;

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

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

        trans.commit().await?;

        Ok(())
    }

    async fn add_layer_to_collection(
        &self,
        layer: &LayerId,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        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 conn = self.conn_pool.get().await?;

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

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

        Ok(())
    }

    async fn add_layer_collection(
        &self,
        collection: AddLayerCollection,
        parent: &LayerCollectionId,
    ) -> Result<LayerCollectionId> {
        let parent =
            Uuid::from_str(&parent.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: parent.0.clone(),
            })?;

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

        let collection = collection;

        let collection_id = Uuid::new_v4();

        let trans = conn.build_transaction().start().await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO layer_collections (id, name, description, properties)
            VALUES ($1, $2, $3, $4);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &collection_id,
                    &collection.name,
                    &collection.description,
                    &collection.properties,
                ],
            )
            .await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO collection_children (parent, child)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        trans.execute(&stmt, &[&parent, &collection_id]).await?;

        trans.commit().await?;

        Ok(LayerCollectionId(collection_id.to_string()))
    }

    async fn add_layer_collection_with_id(
        &self,
        id: &LayerCollectionId,
        collection: AddLayerCollection,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let collection_id =
            Uuid::from_str(&id.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: id.0.clone(),
            })?;

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

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

        let trans = conn.build_transaction().start().await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO layer_collections (id, name, description, properties)
            VALUES ($1, $2, $3, $4);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &collection_id,
                    &collection.name,
                    &collection.description,
                    &collection.properties,
                ],
            )
            .await?;

        let stmt = trans
            .prepare(
                "
            INSERT INTO collection_children (parent, child)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        trans.execute(&stmt, &[&parent, &collection_id]).await?;

        trans.commit().await?;

        Ok(())
    }

    async fn add_collection_to_parent(
        &self,
        collection: &LayerCollectionId,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let collection =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

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

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

        let stmt = conn
            .prepare(
                "
            INSERT INTO collection_children (parent, child)
            VALUES ($1, $2) ON CONFLICT DO NOTHING;",
            )
            .await?;

        conn.execute(&stmt, &[&parent, &collection]).await?;

        Ok(())
    }

    async fn remove_layer_collection(&self, collection: &LayerCollectionId) -> Result<()> {
        let collection =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

        if collection == INTERNAL_LAYER_DB_ROOT_COLLECTION_ID {
            return Err(LayerDbError::CannotRemoveRootCollection.into());
        }

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

        // delete the collection!
        // on delete cascade removes all entries from `collection_children` and `collection_layers`

        let remove_layer_collection_stmt = transaction
            .prepare(
                "DELETE FROM layer_collections
                 WHERE id = $1;",
            )
            .await?;
        transaction
            .execute(&remove_layer_collection_stmt, &[&collection])
            .await?;

        _remove_collections_without_parent_collection(&transaction).await?;

        _remove_layers_without_parent_collection(&transaction).await?;

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

    async fn remove_layer_from_collection(
        &self,
        layer: &LayerId,
        collection: &LayerCollectionId,
    ) -> Result<()> {
        let collection_uuid =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

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

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

        let remove_layer_collection_stmt = transaction
            .prepare(
                "DELETE FROM collection_layers
                 WHERE collection = $1
                 AND layer = $2;",
            )
            .await?;
        let num_results = transaction
            .execute(
                &remove_layer_collection_stmt,
                &[&collection_uuid, &layer_uuid],
            )
            .await?;

        if num_results == 0 {
            return Err(LayerDbError::NoLayerForGivenIdInCollection {
                layer: layer.clone(),
                collection: collection.clone(),
            }
            .into());
        }

        _remove_layers_without_parent_collection(&transaction).await?;

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

    async fn remove_layer_collection_from_parent(
        &self,
        collection: &LayerCollectionId,
        parent: &LayerCollectionId,
    ) -> Result<()> {
        let collection_uuid =
            Uuid::from_str(&collection.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: collection.0.clone(),
            })?;

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

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

        let remove_layer_collection_stmt = transaction
            .prepare(
                "DELETE FROM collection_children
                 WHERE child = $1
                 AND parent = $2;",
            )
            .await?;
        let num_results = transaction
            .execute(
                &remove_layer_collection_stmt,
                &[&collection_uuid, &parent_collection_uuid],
            )
            .await?;

        if num_results == 0 {
            return Err(LayerDbError::NoCollectionForGivenIdInCollection {
                collection: collection.clone(),
                parent: parent.clone(),
            }
            .into());
        }

        _remove_collections_without_parent_collection(&transaction).await?;

        _remove_layers_without_parent_collection(&transaction).await?;

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

#[async_trait]
impl<Tls> LayerCollectionProvider for PostgresDb<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 collection = Uuid::from_str(&collection_id.0).map_err(|_| {
            crate::error::Error::IdStringMustBeUuid {
                found: collection_id.0.clone(),
            }
        })?;

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

        let stmt = conn
            .prepare(
                "
        SELECT DISTINCT name, description, properties
        FROM layer_collections
        WHERE id = $1;",
            )
            .await?;

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

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

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

        let rows = conn
            .query(
                &stmt,
                &[
                    &collection,
                    &i64::from(options.limit),
                    &i64::from(options.offset),
                ],
            )
            .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>>>()?;

        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 layer_id =
            Uuid::from_str(&id.0).map_err(|_| crate::error::Error::IdStringMustBeUuid {
                found: id.0.clone(),
            })?;

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

        let stmt = conn
            .prepare(
                "
            SELECT 
                name,
                description,
                workflow,
                symbology,
                properties,
                metadata
            FROM layers
            WHERE id = $1;",
            )
            .await?;

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

        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: serde_json::from_value(row.get(3)).context(crate::error::SerdeJson)?,
            properties: row.get(4),
            metadata: serde_json::from_value(row.get(5)).context(crate::error::SerdeJson)?,
        })
    }
}

#[async_trait]
impl<Tls> LayerProviderDb for PostgresDb<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: Box<dyn DataProviderDefinition>,
    ) -> Result<DataProviderId> {
        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(),
                &serde_json::to_value(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
            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
               FROM 
                   layer_providers
               WHERE
                   id = $1",
            )
            .await?;

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

        let definition = serde_json::from_value::<Box<dyn DataProviderDefinition>>(row.get(0))?;

        definition.initialize().await
    }
}

1	use crate::api::model::datatypes::{DataProviderId, LayerId};
2
3	use crate::contexts::PostgresDb;
4	use crate::layers::layer::Property;
5
6	use crate::{
7	error::Result,
8	layers::{
9	external::{DataProvider, DataProviderDefinition},
10	layer::{
11	AddLayer, AddLayerCollection, CollectionItem, Layer, LayerCollection,
12	LayerCollectionListOptions, LayerCollectionListing, LayerListing,
13	ProviderLayerCollectionId, ProviderLayerId,
14	},
15	listing::{LayerCollectionId, LayerCollectionProvider},
16	storage::{
17	LayerDb, LayerProviderDb, LayerProviderListing, LayerProviderListingOptions,
18	INTERNAL_LAYER_DB_ROOT_COLLECTION_ID, INTERNAL_PROVIDER_ID,
19	},
20	LayerDbError,
21	},
22	};
23	use async_trait::async_trait;
24	use bb8_postgres::tokio_postgres::{
25	tls::{MakeTlsConnect, TlsConnect},
26	Socket,
27	};
28
29	use snafu::ResultExt;
30	use std::str::FromStr;
31	use uuid::Uuid;
32
33	/// delete all collections without parent collection
34	async fn _remove_collections_without_parent_collection(	3✔
35	transaction: &tokio_postgres::Transaction<'_>,	3✔
36	) -> Result<()> {	3✔
37	// HINT: a recursive delete statement seems reasonable, but hard to implement in postgres
38	// because you have a graph with potential loops
39
40	let remove_layer_collections_without_parents_stmt = transaction	3✔
41	.prepare(	3✔
42	"DELETE FROM layer_collections	3✔
43	WHERE id <> $1 -- do not delete root collection	3✔
44	AND id NOT IN (	3✔
45	SELECT child FROM collection_children	3✔
46	);",	3✔
47	)	3✔
48	.await?;	3✔
49	while 0 < transaction	5✔
50	.execute(	5✔
51	&remove_layer_collections_without_parents_stmt,	5✔
52	&[&INTERNAL_LAYER_DB_ROOT_COLLECTION_ID],	5✔
53	)	5✔
54	.await?	5✔
55	{	2✔
56	// whenever one collection is deleted, we have to check again if there are more	2✔
57	// collections without parents	2✔
58	}	2✔
59
60	Ok(())	3✔
61	}	3✔
62
63	/// delete all layers without parent collection
64	async fn _remove_layers_without_parent_collection(	5✔
65	transaction: &tokio_postgres::Transaction<'_>,	5✔
66	) -> Result<()> {	5✔
67	let remove_layers_without_parents_stmt = transaction	5✔
68	.prepare(	5✔
69	"DELETE FROM layers	5✔
70	WHERE id NOT IN (	5✔
71	SELECT layer FROM collection_layers	5✔
72	);",	5✔
73	)	5✔
74	.await?;	5✔
75	transaction	5✔
76	.execute(&remove_layers_without_parents_stmt, &[])	5✔
77	.await?;	5✔
78
79	Ok(())	5✔
80	}	5✔
81
82	#[async_trait]
83	impl<Tls> LayerDb for PostgresDb<Tls>
84	where
85	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
86	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
87	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
88	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
89	{
90	async fn add_layer(&self, layer: AddLayer, collection: &LayerCollectionId) -> Result<LayerId> {	6✔
91	let collection_id =	6✔
92	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	6✔
93	found: collection.0.clone(),	×
94	})?;	6✔
95
96	let mut conn = self.conn_pool.get().await?;	6✔
97
98	let layer = layer;	6✔
99		6✔
100	let layer_id = Uuid::new_v4();	6✔
101	let symbology = serde_json::to_value(&layer.symbology).context(crate::error::SerdeJson)?;	6✔
102
103	let metadata = serde_json::to_value(&layer.metadata).context(crate::error::SerdeJson)?;	6✔
104
105	let trans = conn.build_transaction().start().await?;	6✔
106
107	let stmt = trans	6✔
108	.prepare(	6✔
109	"	6✔
110	INSERT INTO layers (id, name, description, workflow, symbology, properties, metadata)	6✔
111	VALUES ($1, $2, $3, $4, $5, $6, $7);",	6✔
112	)	6✔
113	.await?;	9✔
114
115	trans
116	.execute(
117	&stmt,	6✔
118	&[	6✔
119	&layer_id,	6✔
120	&layer.name,	6✔
121	&layer.description,	6✔
122	&serde_json::to_value(&layer.workflow).context(crate::error::SerdeJson)?,	6✔
123	&symbology,	6✔
124	&layer.properties,	6✔
125	&metadata,	6✔
126	],
127	)
128	.await?;	6✔
129
130	let stmt = trans	6✔
131	.prepare(	6✔
132	"	6✔
133	INSERT INTO collection_layers (collection, layer)	6✔
134	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	6✔
135	)	6✔
136	.await?;	6✔
137
138	trans.execute(&stmt, &[&collection_id, &layer_id]).await?;	6✔
139
140	trans.commit().await?;	6✔
141
142	Ok(LayerId(layer_id.to_string()))	6✔
143	}	12✔
144
145	async fn add_layer_with_id(	×
146	&self,	×
147	id: &LayerId,	×
148	layer: AddLayer,	×
149	collection: &LayerCollectionId,	×
150	) -> Result<()> {	×
151	let layer_id =	×
152	Uuid::from_str(&id.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	×
153	found: collection.0.clone(),	×
154	})?;	×
155
156	let collection_id =	×
157	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	×
158	found: collection.0.clone(),	×
159	})?;	×
160
161	let mut conn = self.conn_pool.get().await?;	×
162
163	let layer = layer;	×
164
165	let symbology = serde_json::to_value(&layer.symbology).context(crate::error::SerdeJson)?;	×
166
167	let metadata = serde_json::to_value(&layer.metadata).context(crate::error::SerdeJson)?;	×
168
169	let trans = conn.build_transaction().start().await?;	×
170
171	let stmt = trans	×
172	.prepare(	×
173	"	×
174	INSERT INTO layers (id, name, description, workflow, symbology, properties, metadata)	×
175	VALUES ($1, $2, $3, $4, $5, $6, $7);",	×
176	)	×
177	.await?;	×
178
179	trans
180	.execute(
181	&stmt,	×
182	&[	×
183	&layer_id,	×
184	&layer.name,	×
185	&layer.description,	×
186	&serde_json::to_value(&layer.workflow).context(crate::error::SerdeJson)?,	×
187	&symbology,	×
188	&layer.properties,	×
189	&metadata,	×
190	],
191	)
192	.await?;	×
193
194	let stmt = trans	×
195	.prepare(	×
196	"	×
197	INSERT INTO collection_layers (collection, layer)	×
198	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	×
199	)	×
200	.await?;	×
201
202	trans.execute(&stmt, &[&collection_id, &layer_id]).await?;	×
203
204	trans.commit().await?;	×
205
206	Ok(())	×
207	}	×
208
209	async fn add_layer_to_collection(	2✔
210	&self,	2✔
211	layer: &LayerId,	2✔
212	collection: &LayerCollectionId,	2✔
213	) -> Result<()> {	2✔
214	let layer_id =	1✔
215	Uuid::from_str(&layer.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
216	found: layer.0.clone(),	1✔
217	})?;	2✔
218
219	let collection_id =	1✔
220	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	1✔
221	found: collection.0.clone(),	×
222	})?;	1✔
223
224	let conn = self.conn_pool.get().await?;	1✔
225
226	let stmt = conn	1✔
227	.prepare(	1✔
228	"	1✔
229	INSERT INTO collection_layers (collection, layer)	1✔
230	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	1✔
231	)	1✔
232	.await?;	1✔
233
234	conn.execute(&stmt, &[&collection_id, &layer_id]).await?;	1✔
235
236	Ok(())	1✔
237	}	4✔
238
239	async fn add_layer_collection(	8✔
240	&self,	8✔
241	collection: AddLayerCollection,	8✔
242	parent: &LayerCollectionId,	8✔
243	) -> Result<LayerCollectionId> {	8✔
244	let parent =	8✔
245	Uuid::from_str(&parent.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	8✔
246	found: parent.0.clone(),	×
247	})?;	8✔
248
249	let mut conn = self.conn_pool.get().await?;	8✔
250
251	let collection = collection;	8✔
252		8✔
253	let collection_id = Uuid::new_v4();	8✔
254
255	let trans = conn.build_transaction().start().await?;	8✔
256
257	let stmt = trans	8✔
258	.prepare(	8✔
259	"	8✔
260	INSERT INTO layer_collections (id, name, description, properties)	8✔
261	VALUES ($1, $2, $3, $4);",	8✔
262	)	8✔
263	.await?;	20✔
264
265	trans	8✔
266	.execute(	8✔
267	&stmt,	8✔
268	&[	8✔
269	&collection_id,	8✔
270	&collection.name,	8✔
271	&collection.description,	8✔
272	&collection.properties,	8✔
273	],	8✔
274	)	8✔
275	.await?;	8✔
276
277	let stmt = trans	8✔
278	.prepare(	8✔
279	"	8✔
280	INSERT INTO collection_children (parent, child)	8✔
281	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	8✔
282	)	8✔
283	.await?;	8✔
284
285	trans.execute(&stmt, &[&parent, &collection_id]).await?;	8✔
286
287	trans.commit().await?;	8✔
288
289	Ok(LayerCollectionId(collection_id.to_string()))	8✔
290	}	16✔
291
292	async fn add_layer_collection_with_id(	×
293	&self,	×
294	id: &LayerCollectionId,	×
295	collection: AddLayerCollection,	×
296	parent: &LayerCollectionId,	×
297	) -> Result<()> {	×
298	let collection_id =	×
299	Uuid::from_str(&id.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	×
300	found: id.0.clone(),	×
301	})?;	×
302
303	let parent =	×
304	Uuid::from_str(&parent.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	×
305	found: parent.0.clone(),	×
306	})?;	×
307
308	let mut conn = self.conn_pool.get().await?;	×
309
310	let trans = conn.build_transaction().start().await?;	×
311
312	let stmt = trans	×
313	.prepare(	×
314	"	×
315	INSERT INTO layer_collections (id, name, description, properties)	×
316	VALUES ($1, $2, $3, $4);",	×
317	)	×
318	.await?;	×
319
320	trans	×
321	.execute(	×
322	&stmt,	×
323	&[	×
324	&collection_id,	×
325	&collection.name,	×
326	&collection.description,	×
327	&collection.properties,	×
328	],	×
329	)	×
330	.await?;	×
331
332	let stmt = trans	×
333	.prepare(	×
334	"	×
335	INSERT INTO collection_children (parent, child)	×
336	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	×
337	)	×
338	.await?;	×
339
340	trans.execute(&stmt, &[&parent, &collection_id]).await?;	×
341
342	trans.commit().await?;	×
343
344	Ok(())	×
345	}	×
346
347	async fn add_collection_to_parent(	1✔
348	&self,	1✔
349	collection: &LayerCollectionId,	1✔
350	parent: &LayerCollectionId,	1✔
351	) -> Result<()> {	1✔
352	let collection =	1✔
353	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	1✔
354	found: collection.0.clone(),	×
355	})?;	1✔
356
357	let parent =	1✔
358	Uuid::from_str(&parent.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	1✔
359	found: parent.0.clone(),	×
360	})?;	1✔
361
362	let conn = self.conn_pool.get().await?;	1✔
363
364	let stmt = conn	1✔
365	.prepare(	1✔
366	"	1✔
367	INSERT INTO collection_children (parent, child)	1✔
368	VALUES ($1, $2) ON CONFLICT DO NOTHING;",	1✔
369	)	1✔
370	.await?;	1✔
371
372	conn.execute(&stmt, &[&parent, &collection]).await?;	1✔
373
374	Ok(())	1✔
375	}	2✔
376
377	async fn remove_layer_collection(&self, collection: &LayerCollectionId) -> Result<()> {	3✔
378	let collection =	3✔
379	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	3✔
380	found: collection.0.clone(),	×
381	})?;	3✔
382
383	if collection == INTERNAL_LAYER_DB_ROOT_COLLECTION_ID {	3✔
384	return Err(LayerDbError::CannotRemoveRootCollection.into());	1✔
385	}	2✔
386
387	let mut conn = self.conn_pool.get().await?;	2✔
388	let transaction = conn.transaction().await?;	2✔
389
390	// delete the collection!
391	// on delete cascade removes all entries from `collection_children` and `collection_layers`
392
393	let remove_layer_collection_stmt = transaction	2✔
394	.prepare(	2✔
395	"DELETE FROM layer_collections	2✔
396	WHERE id = $1;",	2✔
397	)	2✔
398	.await?;	2✔
399	transaction	2✔
400	.execute(&remove_layer_collection_stmt, &[&collection])	2✔
401	.await?;	2✔
402
403	_remove_collections_without_parent_collection(&transaction).await?;	4✔
404
405	_remove_layers_without_parent_collection(&transaction).await?;	4✔
406
407	transaction.commit().await.map_err(Into::into)	2✔
408	}	6✔
409
410	async fn remove_layer_from_collection(	2✔
411	&self,	2✔
412	layer: &LayerId,	2✔
413	collection: &LayerCollectionId,	2✔
414	) -> Result<()> {	2✔
415	let collection_uuid =	2✔
416	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
417	found: collection.0.clone(),	×
418	})?;	2✔
419
420	let layer_uuid =	2✔
421	Uuid::from_str(&layer.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	2✔
422	found: layer.0.clone(),	×
423	})?;	2✔
424
425	let mut conn = self.conn_pool.get().await?;	2✔
426	let transaction = conn.transaction().await?;	2✔
427
428	let remove_layer_collection_stmt = transaction	2✔
429	.prepare(	2✔
430	"DELETE FROM collection_layers	2✔
431	WHERE collection = $1	2✔
432	AND layer = $2;",	2✔
433	)	2✔
434	.await?;	2✔
435	let num_results = transaction	2✔
436	.execute(	2✔
437	&remove_layer_collection_stmt,	2✔
438	&[&collection_uuid, &layer_uuid],	2✔
439	)	2✔
440	.await?;	2✔
441
442	if num_results == 0 {	2✔
443	return Err(LayerDbError::NoLayerForGivenIdInCollection {	×
444	layer: layer.clone(),	×
445	collection: collection.clone(),	×
446	}	×
447	.into());	×
448	}	2✔
449		2✔
450	_remove_layers_without_parent_collection(&transaction).await?;	4✔
451
452	transaction.commit().await.map_err(Into::into)	2✔
453	}	4✔
454
455	async fn remove_layer_collection_from_parent(	1✔
456	&self,	1✔
457	collection: &LayerCollectionId,	1✔
458	parent: &LayerCollectionId,	1✔
459	) -> Result<()> {	1✔
460	let collection_uuid =	1✔
461	Uuid::from_str(&collection.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	1✔
462	found: collection.0.clone(),	×
463	})?;	1✔
464
465	let parent_collection_uuid =	1✔
466	Uuid::from_str(&parent.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	1✔
467	found: parent.0.clone(),	×
468	})?;	1✔
469
470	let mut conn = self.conn_pool.get().await?;	1✔
471	let transaction = conn.transaction().await?;	1✔
472
473	let remove_layer_collection_stmt = transaction	1✔
474	.prepare(	1✔
475	"DELETE FROM collection_children	1✔
476	WHERE child = $1	1✔
477	AND parent = $2;",	1✔
478	)	1✔
479	.await?;	1✔
480	let num_results = transaction	1✔
481	.execute(	1✔
482	&remove_layer_collection_stmt,	1✔
483	&[&collection_uuid, &parent_collection_uuid],	1✔
484	)	1✔
485	.await?;	1✔
486
487	if num_results == 0 {	1✔
488	return Err(LayerDbError::NoCollectionForGivenIdInCollection {	×
489	collection: collection.clone(),	×
490	parent: parent.clone(),	×
491	}	×
492	.into());	×
493	}	1✔
494		1✔
495	_remove_collections_without_parent_collection(&transaction).await?;	4✔
496
497	_remove_layers_without_parent_collection(&transaction).await?;	2✔
498
499	transaction.commit().await.map_err(Into::into)	1✔
500	}	2✔
501	}
502
503	#[async_trait]
504	impl<Tls> LayerCollectionProvider for PostgresDb<Tls>
505	where
506	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
507	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
508	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
509	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
510	{
511	#[allow(clippy::too_many_lines)]
512	async fn load_layer_collection(	12✔
513	&self,	12✔
514	collection_id: &LayerCollectionId,	12✔
515	options: LayerCollectionListOptions,	12✔
516	) -> Result<LayerCollection> {	12✔
517	let collection = Uuid::from_str(&collection_id.0).map_err(\|_\| {	12✔
518	crate::error::Error::IdStringMustBeUuid {	×
519	found: collection_id.0.clone(),	×
520	}	×
521	})?;	12✔
522
523	let conn = self.conn_pool.get().await?;	12✔
524
525	let stmt = conn	12✔
526	.prepare(	12✔
527	"	12✔
528	SELECT DISTINCT name, description, properties	12✔
529	FROM layer_collections	12✔
530	WHERE id = $1;",	12✔
531	)	12✔
532	.await?;	12✔
533
534	let row = conn.query_one(&stmt, &[&collection]).await?;	12✔
535
536	let name: String = row.get(0);	9✔
537	let description: String = row.get(1);	9✔
538	let properties: Vec<Property> = row.get(2);	9✔
539
540	let stmt = conn	9✔
541	.prepare(	9✔
542	"	9✔
543	SELECT DISTINCT id, name, description, properties, is_layer	9✔
544	FROM (	9✔
545	SELECT	9✔
546	concat(id, '') AS id,	9✔
547	name,	9✔
548	description,	9✔
549	properties,	9✔
550	FALSE AS is_layer	9✔
551	FROM layer_collections	9✔
552	JOIN collection_children cc ON (id = cc.child)	9✔
553	WHERE cc.parent = $1	9✔
554	) u UNION (	9✔
555	SELECT	9✔
556	concat(id, '') AS id,	9✔
557	name,	9✔
558	description,	9✔
559	properties,	9✔
560	TRUE AS is_layer	9✔
561	FROM layers uc	9✔
562	JOIN collection_layers cl ON (id = cl.layer)	9✔
563	WHERE cl.collection = $1	9✔
564	)	9✔
565	ORDER BY is_layer ASC, name ASC	9✔
566	LIMIT $2	9✔
567	OFFSET $3;	9✔
568	",	9✔
569	)	9✔
570	.await?;	9✔
571
572	let rows = conn	9✔
573	.query(	9✔
574	&stmt,	9✔
575	&[	9✔
576	&collection,	9✔
577	&i64::from(options.limit),	9✔
578	&i64::from(options.offset),	9✔
579	],	9✔
580	)	9✔
581	.await?;	9✔
582
583	let items = rows	9✔
584	.into_iter()	9✔
585	.map(\|row\| {	11✔
586	let is_layer: bool = row.get(4);	11✔
587		11✔
588	if is_layer {	11✔
589	Ok(CollectionItem::Layer(LayerListing {	5✔
590	id: ProviderLayerId {	5✔
591	provider_id: INTERNAL_PROVIDER_ID,	5✔
592	layer_id: LayerId(row.get(0)),	5✔
593	},	5✔
594	name: row.get(1),	5✔
595	description: row.get(2),	5✔
596	properties: row.get(3),	5✔
597	}))	5✔
598	} else {
599	Ok(CollectionItem::Collection(LayerCollectionListing {	6✔
600	id: ProviderLayerCollectionId {	6✔
601	provider_id: INTERNAL_PROVIDER_ID,	6✔
602	collection_id: LayerCollectionId(row.get(0)),	6✔
603	},	6✔
604	name: row.get(1),	6✔
605	description: row.get(2),	6✔
606	properties: row.get(3),	6✔
607	}))	6✔
608	}
609	})	11✔
610	.collect::<Result<Vec<CollectionItem>>>()?;	9✔
611
612	Ok(LayerCollection {	9✔
613	id: ProviderLayerCollectionId {	9✔
614	provider_id: INTERNAL_PROVIDER_ID,	9✔
615	collection_id: collection_id.clone(),	9✔
616	},	9✔
617	name,	9✔
618	description,	9✔
619	items,	9✔
620	entry_label: None,	9✔
621	properties,	9✔
622	})	9✔
623	}	24✔
624
625	async fn get_root_layer_collection_id(&self) -> Result<LayerCollectionId> {	5✔
626	Ok(LayerCollectionId(	5✔
627	INTERNAL_LAYER_DB_ROOT_COLLECTION_ID.to_string(),	5✔
628	))	5✔
629	}	5✔
630
631	async fn load_layer(&self, id: &LayerId) -> Result<Layer> {	5✔
632	let layer_id =	5✔
633	Uuid::from_str(&id.0).map_err(\|_\| crate::error::Error::IdStringMustBeUuid {	5✔
634	found: id.0.clone(),	×
635	})?;	5✔
636
637	let conn = self.conn_pool.get().await?;	5✔
638
639	let stmt = conn	5✔
640	.prepare(	5✔
641	"	5✔
642	SELECT	5✔
643	name,	5✔
644	description,	5✔
645	workflow,	5✔
646	symbology,	5✔
647	properties,	5✔
648	metadata	5✔
649	FROM layers	5✔
650	WHERE id = $1;",	5✔
651	)	5✔
652	.await?;	5✔
653
654	let row = conn	5✔
655	.query_one(&stmt, &[&layer_id])	5✔
656	.await	5✔
657	.map_err(\|_error\| LayerDbError::NoLayerForGivenId { id: id.clone() })?;	5✔
658
659	Ok(Layer {
660	id: ProviderLayerId {	2✔
661	provider_id: INTERNAL_PROVIDER_ID,	2✔
662	layer_id: id.clone(),	2✔
663	},	2✔
664	name: row.get(0),	2✔
665	description: row.get(1),	2✔
666	workflow: serde_json::from_value(row.get(2)).context(crate::error::SerdeJson)?,	2✔
667	symbology: serde_json::from_value(row.get(3)).context(crate::error::SerdeJson)?,	2✔
668	properties: row.get(4),	2✔
669	metadata: serde_json::from_value(row.get(5)).context(crate::error::SerdeJson)?,	2✔
670	})
671	}	10✔
672	}
673
674	#[async_trait]
675	impl<Tls> LayerProviderDb for PostgresDb<Tls>
676	where
677	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
678	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
679	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
680	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
681	{
682	async fn add_layer_provider(	1✔
683	&self,	1✔
684	provider: Box<dyn DataProviderDefinition>,	1✔
685	) -> Result<DataProviderId> {	1✔
686	let conn = self.conn_pool.get().await?;	1✔
687
688	let stmt = conn	1✔
689	.prepare(	1✔
690	"	1✔
691	INSERT INTO layer_providers (	1✔
692	id,	1✔
693	type_name,	1✔
694	name,	1✔
695	definition	1✔
696	)	1✔
697	VALUES ($1, $2, $3, $4)",	1✔
698	)	1✔
699	.await?;	1✔
700
701	let id = provider.id();	1✔
702	conn.execute(	1✔
703	&stmt,	1✔
704	&[	1✔
705	&id,	1✔
706	&provider.type_name(),	1✔
707	&provider.name(),	1✔
708	&serde_json::to_value(provider)?,	1✔
709	],
710	)
711	.await?;	1✔
712	Ok(id)	1✔
713	}	2✔
714
715	async fn list_layer_providers(	1✔
716	&self,	1✔
717	options: LayerProviderListingOptions,	1✔
718	) -> Result<Vec<LayerProviderListing>> {	1✔
719	// TODO: permission
720	let conn = self.conn_pool.get().await?;	1✔
721
722	let stmt = conn	1✔
723	.prepare(	1✔
724	"	1✔
725	SELECT	1✔
726	id,	1✔
727	name,	1✔
728	type_name	1✔
729	FROM	1✔
730	layer_providers	1✔
731	ORDER BY name ASC	1✔
732	LIMIT $1	1✔
733	OFFSET $2;",	1✔
734	)	1✔
735	.await?;	1✔
736
737	let rows = conn	1✔
738	.query(	1✔
739	&stmt,	1✔
740	&[&i64::from(options.limit), &i64::from(options.offset)],	1✔
741	)	1✔
742	.await?;	1✔
743
744	Ok(rows	1✔
745	.iter()	1✔
746	.map(\|row\| LayerProviderListing {	1✔
747	id: row.get(0),	1✔
748	name: row.get(1),	1✔
749	description: row.get(2),	1✔
750	})	1✔
751	.collect())	1✔
752	}	2✔
753
754	async fn load_layer_provider(&self, id: DataProviderId) -> Result<Box<dyn DataProvider>> {	1✔
755	// TODO: permissions
756	let conn = self.conn_pool.get().await?;	1✔
757
758	let stmt = conn	1✔
759	.prepare(	1✔
760	"	1✔
761	SELECT	1✔
762	definition	1✔
763	FROM	1✔
764	layer_providers	1✔
765	WHERE	1✔
766	id = $1",	1✔
767	)	1✔
768	.await?;	1✔
769
770	let row = conn.query_one(&stmt, &[&id]).await?;	1✔
771
772	let definition = serde_json::from_value::<Box<dyn DataProviderDefinition>>(row.get(0))?;	1✔
773
774	definition.initialize().await	1✔
775	}	2✔
776	}

geo-engine / geoengine / 5624216179

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