4056270646

Build Type

push

github

Committed by GitHub

Commit Message

Merge #706

Run Details

108 of 108 new or added lines in 5 files covered. (100.0%)

87598 of 99672 relevant lines covered (87.89%)

77046.89 hits per line

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

90.31

/services/src/pro/projects/postgres_projectdb.rs

use crate::pro::contexts::PostgresContext;
use crate::pro::users::UserId;
use crate::pro::users::UserSession;
use crate::projects::Layer;
use crate::projects::Plot;
use crate::projects::{
    CreateProject, Project, ProjectDb, ProjectId, ProjectListOptions, ProjectListing,
    ProjectVersion, ProjectVersionId, UpdateProject,
};
use crate::util::user_input::Validated;
use crate::util::Identifier;
use crate::workflows::workflow::WorkflowId;
use crate::{
    error::{self, Result},
    pro::projects::projectdb::ProjectPermission,
};
use async_trait::async_trait;
use bb8_postgres::PostgresConnectionManager;
use bb8_postgres::{
    bb8::Pool, tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect,
    tokio_postgres::Socket,
};
use snafu::ResultExt;

use bb8_postgres::bb8::PooledConnection;
use bb8_postgres::tokio_postgres::Transaction;

use super::LoadVersion;
use super::ProProjectDb;
use super::UserProjectPermission;

pub struct PostgresProjectDb<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,
{
    conn_pool: Pool<PostgresConnectionManager<Tls>>,
}

impl<Tls> PostgresProjectDb<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,
{
    pub fn new(conn_pool: Pool<PostgresConnectionManager<Tls>>) -> Self {
        Self { conn_pool }
    }

    async fn list_plots(
        &self,
        conn: &PooledConnection<'_, PostgresConnectionManager<Tls>>,
        project_version_id: &ProjectVersionId,
    ) -> Result<Vec<String>> {
        let stmt = conn
            .prepare(
                "
                    SELECT name
                    FROM project_version_plots
                    WHERE project_version_id = $1;
                ",
            )
            .await?;

        let plot_rows = conn.query(&stmt, &[project_version_id]).await?;
        let plot_names = plot_rows.iter().map(|row| row.get(0)).collect();

        Ok(plot_names)
    }

    async fn load_plots(
        &self,
        conn: &PooledConnection<'_, PostgresConnectionManager<Tls>>,
        project_version_id: &ProjectVersionId,
    ) -> Result<Vec<Plot>> {
        let stmt = conn
            .prepare(
                "
                SELECT  
                    name, workflow_id
                FROM project_version_plots
                WHERE project_version_id = $1
                ORDER BY plot_index ASC
                ",
            )
            .await?;

        let rows = conn.query(&stmt, &[project_version_id]).await?;

        let plots = rows
            .into_iter()
            .map(|row| Plot {
                workflow: WorkflowId(row.get(1)),
                name: row.get(0),
            })
            .collect();

        Ok(plots)
    }

    async fn update_plots(
        &self,
        trans: &Transaction<'_>,
        project_id: &ProjectId,
        project_version_id: &ProjectVersionId,
        plots: &[Plot],
    ) -> Result<()> {
        for (idx, plot) in plots.iter().enumerate() {
            let stmt = trans
                .prepare(
                    "
                    INSERT INTO project_version_plots (
                        project_id,
                        project_version_id,
                        plot_index,
                        name,
                        workflow_id)
                    VALUES ($1, $2, $3, $4, $5);
                    ",
                )
                .await?;

            trans
                .execute(
                    &stmt,
                    &[
                        project_id,
                        project_version_id,
                        &(idx as i32),
                        &plot.name,
                        &plot.workflow,
                    ],
                )
                .await?;
        }

        Ok(())
    }
}

#[async_trait]
impl<Tls> ProjectDb<UserSession> for PostgresProjectDb<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 list(
        &self,
        session: &UserSession,
        options: Validated<ProjectListOptions>,
    ) -> Result<Vec<ProjectListing>> {
        // TODO: project filters
        let options = options.user_input;

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

        let stmt = conn
            .prepare(&format!(
                "
        SELECT p.id, p.project_id, p.name, p.description, p.changed 
        FROM user_project_permissions u JOIN project_versions p ON (u.project_id = p.project_id)
        WHERE
            u.user_id = $1
            AND latest IS TRUE
        ORDER BY p.{}
        LIMIT $2
        OFFSET $3;",
                options.order.to_sql_string()
            ))
            .await?;

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

        let mut project_listings = vec![];
        for project_row in project_rows {
            let project_version_id = ProjectVersionId(project_row.get(0));
            let project_id = ProjectId(project_row.get(1));
            let name = project_row.get(2);
            let description = project_row.get(3);
            let changed = project_row.get(4);

            let stmt = conn
                .prepare(
                    "
                    SELECT name
                    FROM project_version_layers
                    WHERE project_version_id = $1;",
                )
                .await?;

            let layer_rows = conn.query(&stmt, &[&project_version_id]).await?;
            let layer_names = layer_rows.iter().map(|row| row.get(0)).collect();

            project_listings.push(ProjectListing {
                id: project_id,
                name,
                description,
                layer_names,
                plot_names: self.list_plots(&conn, &project_version_id).await?,
                changed,
            });
        }
        Ok(project_listings)
    }

    async fn create(
        &self,
        session: &UserSession,
        create: Validated<CreateProject>,
    ) -> Result<ProjectId> {
        let mut conn = self.conn_pool.get().await?;

        let project: Project = Project::from_create_project(create.user_input);

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

        let stmt = trans
            .prepare("INSERT INTO projects (id) VALUES ($1);")
            .await?;

        trans.execute(&stmt, &[&project.id]).await?;

        let stmt = trans
            .prepare(
                "INSERT INTO project_versions (
                    id,
                    project_id,
                    name,
                    description,
                    bounds,
                    time_step,
                    author_user_id,
                    changed,
                    latest)
                    VALUES ($1, $2, $3, $4, $5, $6, $7, CURRENT_TIMESTAMP, TRUE);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &ProjectVersionId::new(),
                    &project.id,
                    &project.name,
                    &project.description,
                    &project.bounds,
                    &project.time_step,
                    &session.user.id,
                ],
            )
            .await?;

        let stmt = trans
            .prepare(
                "INSERT INTO user_project_permissions (user_id, project_id, permission) VALUES ($1, $2, $3);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[&session.user.id, &project.id, &ProjectPermission::Owner],
            )
            .await?;

        trans.commit().await?;

        Ok(project.id)
    }

    async fn load(&self, session: &UserSession, project: ProjectId) -> Result<Project> {
        self.load_version(session, project, LoadVersion::Latest)
            .await
    }

    #[allow(clippy::too_many_lines)]
    async fn update(&self, session: &UserSession, update: Validated<UpdateProject>) -> Result<()> {
        let update = update.user_input;

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

        PostgresContext::check_user_project_permission(
            &conn,
            session.user.id,
            update.id,
            &[ProjectPermission::Write, ProjectPermission::Owner],
        )
        .await?;

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

        let project = self.load(session, update.id).await?; // TODO: move inside transaction?

        let stmt = trans
            .prepare("UPDATE project_versions SET latest = FALSE WHERE project_id = $1 AND latest IS TRUE;")
            .await?;
        trans.execute(&stmt, &[&project.id]).await?;

        let project = project.update_project(update)?;

        let stmt = trans
            .prepare(
                "
                INSERT INTO project_versions (
                    id,
                    project_id,
                    name,
                    description,
                    bounds,
                    time_step,
                    author_user_id,
                    changed,
                    latest)
                VALUES ($1, $2, $3, $4, $5, $6, $7, CURRENT_TIMESTAMP, TRUE);",
            )
            .await?;

        trans
            .execute(
                &stmt,
                &[
                    &project.version.id,
                    &project.id,
                    &project.name,
                    &project.description,
                    &project.bounds,
                    &project.time_step,
                    &session.user.id,
                ],
            )
            .await?;

        for (idx, layer) in project.layers.iter().enumerate() {
            let stmt = trans
                .prepare(
                    "
                INSERT INTO project_version_layers (
                    project_id,
                    project_version_id,
                    layer_index,
                    name,
                    workflow_id,
                    symbology,
                    visibility)
                VALUES ($1, $2, $3, $4, $5, $6, $7);",
                )
                .await?;

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

            trans
                .execute(
                    &stmt,
                    &[
                        &project.id,
                        &project.version.id,
                        &(idx as i32),
                        &layer.name,
                        &layer.workflow,
                        &symbology,
                        &layer.visibility,
                    ],
                )
                .await?;
        }

        self.update_plots(&trans, &project.id, &project.version.id, &project.plots)
            .await?;

        trans.commit().await?;

        Ok(())
    }

    async fn delete(&self, session: &UserSession, project: ProjectId) -> Result<()> {
        let conn = self.conn_pool.get().await?;

        PostgresContext::check_user_project_permission(
            &conn,
            session.user.id,
            project,
            &[ProjectPermission::Owner],
        )
        .await?;

        let stmt = conn.prepare("DELETE FROM projects WHERE id = $1;").await?;

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

        Ok(())
    }
}

#[async_trait]
impl<Tls> ProProjectDb for PostgresProjectDb<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_version(
        &self,
        session: &UserSession,
        project: ProjectId,
        version: LoadVersion,
    ) -> Result<Project> {
        let conn = self.conn_pool.get().await?;

        PostgresContext::check_user_project_permission(
            &conn,
            session.user.id,
            project,
            &[
                ProjectPermission::Read,
                ProjectPermission::Write,
                ProjectPermission::Owner,
            ],
        )
        .await?;

        let row = if let LoadVersion::Version(version) = version {
            let stmt = conn
                .prepare(
                    "
        SELECT 
            p.project_id, 
            p.id, 
            p.name, 
            p.description,
            p.bounds,
            p.time_step,
            p.changed,
            p.author_user_id
        FROM user_project_permissions u JOIN project_versions p ON (u.project_id = p.project_id)
        WHERE u.user_id = $1 AND u.project_id = $2 AND p.project_version = $3",
                )
                .await?;

            conn.query_one(&stmt, &[&session.user.id, &project, &version])
                .await?
        } else {
            let stmt = conn
                .prepare(
                    "
        SELECT  
            p.project_id, 
            p.id, 
            p.name, 
            p.description,
            p.bounds,
            p.time_step,
            p.changed,
            p.author_user_id
        FROM user_project_permissions u JOIN project_versions p ON (u.project_id = p.project_id)
        WHERE u.user_id = $1 AND u.project_id = $2 AND latest IS TRUE",
                )
                .await?;

            conn.query_one(&stmt, &[&session.user.id, &project]).await?
        };

        let project_id = ProjectId(row.get(0));
        let version_id = ProjectVersionId(row.get(1));
        let name = row.get(2);
        let description = row.get(3);
        let bounds = row.get(4);
        let time_step = row.get(5);
        let changed = row.get(6);
        let _author_id = UserId(row.get(7));

        let stmt = conn
            .prepare(
                "
        SELECT  
            name, workflow_id, symbology, visibility
        FROM project_version_layers
        WHERE project_version_id = $1
        ORDER BY layer_index ASC",
            )
            .await?;

        let rows = conn.query(&stmt, &[&version_id]).await?;

        let mut layers = vec![];
        for row in rows {
            layers.push(Layer {
                workflow: WorkflowId(row.get(1)),
                name: row.get(0),
                symbology: serde_json::from_value(row.get(2)).context(error::SerdeJson)?,
                visibility: row.get(3),
            });
        }

        Ok(Project {
            id: project_id,
            version: ProjectVersion {
                id: version_id,
                changed,
            },
            name,
            description,
            layers,
            plots: self.load_plots(&conn, &version_id).await?,
            bounds,
            time_step,
        })
    }

    async fn versions(
        &self,
        session: &UserSession,
        project: ProjectId,
    ) -> Result<Vec<ProjectVersion>> {
        let conn = self.conn_pool.get().await?;

        PostgresContext::check_user_project_permission(
            &conn,
            session.user.id,
            project,
            &[
                ProjectPermission::Read,
                ProjectPermission::Write,
                ProjectPermission::Owner,
            ],
        )
        .await?;

        let stmt = conn
            .prepare(
                "
                SELECT id, changed, author_user_id
                FROM project_versions WHERE project_id = $1 
                ORDER BY latest DESC, changed DESC, author_user_id DESC",
            )
            .await?;

        let rows = conn.query(&stmt, &[&project]).await?;

        Ok(rows
            .iter()
            .map(|row| ProjectVersion {
                id: ProjectVersionId(row.get(0)),
                changed: row.get(1),
            })
            .collect())
    }

    async fn list_permissions(
        &self,
        session: &UserSession,
        project: ProjectId,
    ) -> Result<Vec<UserProjectPermission>> {
        let conn = self.conn_pool.get().await?;

        PostgresContext::check_user_project_permission(
            &conn,
            session.user.id,
            project,
            &[
                ProjectPermission::Read,
                ProjectPermission::Write,
                ProjectPermission::Owner,
            ],
        )
        .await?;

        let stmt = conn
            .prepare(
                "
        SELECT user_id, project_id, permission FROM user_project_permissions WHERE project_id = $1;",
            )
            .await?;

        let rows = conn.query(&stmt, &[&project]).await?;

        Ok(rows
            .into_iter()
            .map(|row| UserProjectPermission {
                project: ProjectId(row.get(1)),
                permission: row.get(2),
                user: session.user.id,
            })
            .collect())
    }

    async fn add_permission(
        &self,
        session: &UserSession,
        permission: UserProjectPermission,
    ) -> Result<()> {
        let conn = self.conn_pool.get().await?;

        PostgresContext::check_user_project_permission(
            &conn,
            session.user.id,
            permission.project,
            &[ProjectPermission::Owner],
        )
        .await?;

        let stmt = conn
            .prepare(
                "
        INSERT INTO user_project_permissions (user_id, project_id, permission)
        VALUES ($1, $2, $3);",
            )
            .await?;

        conn.execute(
            &stmt,
            &[
                &permission.user,
                &permission.project,
                &permission.permission,
            ],
        )
        .await?;

        Ok(())
    }

    async fn remove_permission(
        &self,
        session: &UserSession,
        permission: UserProjectPermission,
    ) -> Result<()> {
        let conn = self.conn_pool.get().await?;

        PostgresContext::check_user_project_permission(
            &conn,
            session.user.id,
            permission.project,
            &[ProjectPermission::Owner],
        )
        .await?;

        let stmt = conn
            .prepare(
                "
            DELETE FROM user_project_permissions 
            WHERE user_id = $1 AND project_id = $2 AND permission = $3;",
            )
            .await?;

        conn.execute(
            &stmt,
            &[
                &session.user.id,
                &permission.project,
                &permission.permission,
            ],
        )
        .await?;

        Ok(())
    }
}

1	use crate::pro::contexts::PostgresContext;
2	use crate::pro::users::UserId;
3	use crate::pro::users::UserSession;
4	use crate::projects::Layer;
5	use crate::projects::Plot;
6	use crate::projects::{
7	CreateProject, Project, ProjectDb, ProjectId, ProjectListOptions, ProjectListing,
8	ProjectVersion, ProjectVersionId, UpdateProject,
9	};
10	use crate::util::user_input::Validated;
11	use crate::util::Identifier;
12	use crate::workflows::workflow::WorkflowId;
13	use crate::{
14	error::{self, Result},
15	pro::projects::projectdb::ProjectPermission,
16	};
17	use async_trait::async_trait;
18	use bb8_postgres::PostgresConnectionManager;
19	use bb8_postgres::{
20	bb8::Pool, tokio_postgres::tls::MakeTlsConnect, tokio_postgres::tls::TlsConnect,
21	tokio_postgres::Socket,
22	};
23	use snafu::ResultExt;
24
25	use bb8_postgres::bb8::PooledConnection;
26	use bb8_postgres::tokio_postgres::Transaction;
27
28	use super::LoadVersion;
29	use super::ProProjectDb;
30	use super::UserProjectPermission;
31
32	pub struct PostgresProjectDb<Tls>
33	where
34	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
35	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
36	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
37	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
38	{
39	conn_pool: Pool<PostgresConnectionManager<Tls>>,
40	}
41
42	impl<Tls> PostgresProjectDb<Tls>
43	where
44	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
45	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
46	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
47	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
48	{
49	pub fn new(conn_pool: Pool<PostgresConnectionManager<Tls>>) -> Self {	20✔
50	Self { conn_pool }	20✔
51	}	20✔
52
53	async fn list_plots(	4✔
54	&self,	4✔
55	conn: &PooledConnection<'_, PostgresConnectionManager<Tls>>,	4✔
56	project_version_id: &ProjectVersionId,	4✔
57	) -> Result<Vec<String>> {	4✔
58	let stmt = conn	4✔
59	.prepare(	4✔
60	"	4✔
61	SELECT name	4✔
62	FROM project_version_plots	4✔
63	WHERE project_version_id = $1;	4✔
64	",	4✔
65	)	4✔
66	.await?;	4✔
67
68	let plot_rows = conn.query(&stmt, &[project_version_id]).await?;	4✔
69	let plot_names = plot_rows.iter().map(\|row\| row.get(0)).collect();	4✔
70		4✔
71	Ok(plot_names)	4✔
72	}	4✔
73
74	async fn load_plots(	8✔
75	&self,	8✔
76	conn: &PooledConnection<'_, PostgresConnectionManager<Tls>>,	8✔
77	project_version_id: &ProjectVersionId,	8✔
78	) -> Result<Vec<Plot>> {	8✔
79	let stmt = conn	8✔
80	.prepare(	8✔
81	"	8✔
82	SELECT	8✔
83	name, workflow_id	8✔
84	FROM project_version_plots	8✔
85	WHERE project_version_id = $1	8✔
86	ORDER BY plot_index ASC	8✔
87	",	8✔
88	)	8✔
89	.await?;	8✔
90
91	let rows = conn.query(&stmt, &[project_version_id]).await?;	8✔
92
93	let plots = rows	8✔
94	.into_iter()	8✔
95	.map(\|row\| Plot {	8✔
96	workflow: WorkflowId(row.get(1)),	6✔
97	name: row.get(0),	6✔
98	})	8✔
99	.collect();	8✔
100		8✔
101	Ok(plots)	8✔
102	}	8✔
103
104	async fn update_plots(	6✔
105	&self,	6✔
106	trans: &Transaction<'_>,	6✔
107	project_id: &ProjectId,	6✔
108	project_version_id: &ProjectVersionId,	6✔
109	plots: &[Plot],	6✔
110	) -> Result<()> {	6✔
111	for (idx, plot) in plots.iter().enumerate() {	6✔
112	let stmt = trans	6✔
113	.prepare(	6✔
114	"	6✔
115	INSERT INTO project_version_plots (	6✔
116	project_id,	6✔
117	project_version_id,	6✔
118	plot_index,	6✔
119	name,	6✔
120	workflow_id)	6✔
121	VALUES ($1, $2, $3, $4, $5);	6✔
122	",	6✔
123	)	6✔
124	.await?;	6✔
125
126	trans	6✔
127	.execute(	6✔
128	&stmt,	6✔
129	&[	6✔
130	project_id,	6✔
131	project_version_id,	6✔
132	&(idx as i32),	6✔
133	&plot.name,	6✔
134	&plot.workflow,	6✔
135	],	6✔
136	)	6✔
137	.await?;	6✔
138	}
139
140	Ok(())	6✔
141	}	6✔
142	}
143
144	#[async_trait]
145	impl<Tls> ProjectDb<UserSession> for PostgresProjectDb<Tls>
146	where
147	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
148	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
149	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
150	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
151	{
152	async fn list(	2✔
153	&self,	2✔
154	session: &UserSession,	2✔
155	options: Validated<ProjectListOptions>,	2✔
156	) -> Result<Vec<ProjectListing>> {	2✔
157	// TODO: project filters
158	let options = options.user_input;	2✔
159
160	let conn = self.conn_pool.get().await?;	2✔
161
162	let stmt = conn	2✔
163	.prepare(&format!(	2✔
164	"	2✔
165	SELECT p.id, p.project_id, p.name, p.description, p.changed	2✔
166	FROM user_project_permissions u JOIN project_versions p ON (u.project_id = p.project_id)	2✔
167	WHERE	2✔
168	u.user_id = $1	2✔
169	AND latest IS TRUE	2✔
170	ORDER BY p.{}	2✔
171	LIMIT $2	2✔
172	OFFSET $3;",	2✔
173	options.order.to_sql_string()	2✔
174	))	2✔
175	.await?;	2✔
176
177	let project_rows = conn	2✔
178	.query(	2✔
179	&stmt,	2✔
180	&[	2✔
181	&session.user.id,	2✔
182	&i64::from(options.limit),	2✔
183	&i64::from(options.offset),	2✔
184	],	2✔
185	)	2✔
186	.await?;	2✔
187
188	let mut project_listings = vec![];	2✔
189	for project_row in project_rows {	6✔
190	let project_version_id = ProjectVersionId(project_row.get(0));	4✔
191	let project_id = ProjectId(project_row.get(1));	4✔
192	let name = project_row.get(2);	4✔
193	let description = project_row.get(3);	4✔
194	let changed = project_row.get(4);	4✔
195
196	let stmt = conn	4✔
197	.prepare(	4✔
198	"	4✔
199	SELECT name	4✔
200	FROM project_version_layers	4✔
201	WHERE project_version_id = $1;",	4✔
202	)	4✔
203	.await?;	4✔
204
205	let layer_rows = conn.query(&stmt, &[&project_version_id]).await?;	4✔
206	let layer_names = layer_rows.iter().map(\|row\| row.get(0)).collect();	4✔
207		4✔
208	project_listings.push(ProjectListing {	4✔
209	id: project_id,	4✔
210	name,	4✔
211	description,	4✔
212	layer_names,	4✔
213	plot_names: self.list_plots(&conn, &project_version_id).await?,	8✔
214	changed,	4✔
215	});
216	}
217	Ok(project_listings)	2✔
218	}	4✔
219
220	async fn create(	20✔
221	&self,	20✔
222	session: &UserSession,	20✔
223	create: Validated<CreateProject>,	20✔
224	) -> Result<ProjectId> {	20✔
225	let mut conn = self.conn_pool.get().await?;	20✔
226
227	let project: Project = Project::from_create_project(create.user_input);	20✔
228
229	let trans = conn.build_transaction().start().await?;	20✔
230
231	let stmt = trans	20✔
232	.prepare("INSERT INTO projects (id) VALUES ($1);")	20✔
233	.await?;	18✔
234
235	trans.execute(&stmt, &[&project.id]).await?;	21✔
236
237	let stmt = trans	20✔
238	.prepare(	20✔
239	"INSERT INTO project_versions (	20✔
240	id,	20✔
241	project_id,	20✔
242	name,	20✔
243	description,	20✔
244	bounds,	20✔
245	time_step,	20✔
246	author_user_id,	20✔
247	changed,	20✔
248	latest)	20✔
249	VALUES ($1, $2, $3, $4, $5, $6, $7, CURRENT_TIMESTAMP, TRUE);",	20✔
250	)	20✔
251	.await?;	28✔
252
253	trans	20✔
254	.execute(	20✔
255	&stmt,	20✔
256	&[	20✔
257	&ProjectVersionId::new(),	20✔
258	&project.id,	20✔
259	&project.name,	20✔
260	&project.description,	20✔
261	&project.bounds,	20✔
262	&project.time_step,	20✔
263	&session.user.id,	20✔
264	],	20✔
265	)	20✔
266	.await?;	20✔
267
268	let stmt = trans	20✔
269	.prepare(	20✔
270	"INSERT INTO user_project_permissions (user_id, project_id, permission) VALUES ($1, $2, $3);",	20✔
271	)	20✔
272	.await?;	23✔
273
274	trans	20✔
275	.execute(	20✔
276	&stmt,	20✔
277	&[&session.user.id, &project.id, &ProjectPermission::Owner],	20✔
278	)	20✔
279	.await?;	20✔
280
281	trans.commit().await?;	20✔
282
283	Ok(project.id)	20✔
284	}	40✔
285
286	async fn load(&self, session: &UserSession, project: ProjectId) -> Result<Project> {	8✔
287	self.load_version(session, project, LoadVersion::Latest)	8✔
288	.await	58✔
289	}	16✔
290
291	#[allow(clippy::too_many_lines)]
292	async fn update(&self, session: &UserSession, update: Validated<UpdateProject>) -> Result<()> {	6✔
293	let update = update.user_input;	6✔
294
295	let mut conn = self.conn_pool.get().await?;	6✔
296
297	PostgresContext::check_user_project_permission(	6✔
298	&conn,	6✔
299	session.user.id,	6✔
300	update.id,	6✔
301	&[ProjectPermission::Write, ProjectPermission::Owner],	6✔
302	)	6✔
303	.await?;	18✔
304
305	let trans = conn.build_transaction().start().await?;	6✔
306
307	let project = self.load(session, update.id).await?; // TODO: move inside transaction?	54✔
308
309	let stmt = trans	6✔
310	.prepare("UPDATE project_versions SET latest = FALSE WHERE project_id = $1 AND latest IS TRUE;")	6✔
311	.await?;	6✔
312	trans.execute(&stmt, &[&project.id]).await?;	6✔
313
314	let project = project.update_project(update)?;	6✔
315
316	let stmt = trans	6✔
317	.prepare(	6✔
318	"	6✔
319	INSERT INTO project_versions (	6✔
320	id,	6✔
321	project_id,	6✔
322	name,	6✔
323	description,	6✔
324	bounds,	6✔
325	time_step,	6✔
326	author_user_id,	6✔
327	changed,	6✔
328	latest)	6✔
329	VALUES ($1, $2, $3, $4, $5, $6, $7, CURRENT_TIMESTAMP, TRUE);",	6✔
330	)	6✔
331	.await?;	14✔
332
333	trans	6✔
334	.execute(	6✔
335	&stmt,	6✔
336	&[	6✔
337	&project.version.id,	6✔
338	&project.id,	6✔
339	&project.name,	6✔
340	&project.description,	6✔
341	&project.bounds,	6✔
342	&project.time_step,	6✔
343	&session.user.id,	6✔
344	],	6✔
345	)	6✔
346	.await?;	6✔
347
348	for (idx, layer) in project.layers.iter().enumerate() {	6✔
349	let stmt = trans	6✔
350	.prepare(	6✔
351	"	6✔
352	INSERT INTO project_version_layers (	6✔
353	project_id,	6✔
354	project_version_id,	6✔
355	layer_index,	6✔
356	name,	6✔
357	workflow_id,	6✔
358	symbology,	6✔
359	visibility)	6✔
360	VALUES ($1, $2, $3, $4, $5, $6, $7);",	6✔
361	)	6✔
362	.await?;	10✔
363
364	let symbology = serde_json::to_value(&layer.symbology).context(error::SerdeJson)?;	6✔
365
366	trans	6✔
367	.execute(	6✔
368	&stmt,	6✔
369	&[	6✔
370	&project.id,	6✔
371	&project.version.id,	6✔
372	&(idx as i32),	6✔
373	&layer.name,	6✔
374	&layer.workflow,	6✔
375	&symbology,	6✔
376	&layer.visibility,	6✔
377	],	6✔
378	)	6✔
379	.await?;	6✔
380	}
381
382	self.update_plots(&trans, &project.id, &project.version.id, &project.plots)	6✔
383	.await?;	12✔
384
385	trans.commit().await?;	6✔
386
387	Ok(())	6✔
388	}	12✔
389
390	async fn delete(&self, session: &UserSession, project: ProjectId) -> Result<()> {	2✔
391	let conn = self.conn_pool.get().await?;	2✔
392
393	PostgresContext::check_user_project_permission(	2✔
394	&conn,	2✔
395	session.user.id,	2✔
396	project,	2✔
397	&[ProjectPermission::Owner],	2✔
398	)	2✔
399	.await?;	×
400
401	let stmt = conn.prepare("DELETE FROM projects WHERE id = $1;").await?;	2✔
402
403	conn.execute(&stmt, &[&project]).await?;	2✔
404
405	Ok(())	2✔
406	}	4✔
407	}
408
409	#[async_trait]
410	impl<Tls> ProProjectDb for PostgresProjectDb<Tls>
411	where
412	Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,
413	<Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,
414	<Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
415	<<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
416	{
417	#[allow(clippy::too_many_lines)]
418	async fn load_version(	10✔
419	&self,	10✔
420	session: &UserSession,	10✔
421	project: ProjectId,	10✔
422	version: LoadVersion,	10✔
423	) -> Result<Project> {	10✔
424	let conn = self.conn_pool.get().await?;	10✔
425
426	PostgresContext::check_user_project_permission(	10✔
427	&conn,	10✔
428	session.user.id,	10✔
429	project,	10✔
430	&[	10✔
431	ProjectPermission::Read,	10✔
432	ProjectPermission::Write,	10✔
433	ProjectPermission::Owner,	10✔
434	],	10✔
435	)	10✔
436	.await?;	20✔
437
438	let row = if let LoadVersion::Version(version) = version {	8✔
439	let stmt = conn	×
440	.prepare(	×
441	"	×
442	SELECT	×
443	p.project_id,	×
444	p.id,	×
445	p.name,	×
446	p.description,	×
447	p.bounds,	×
448	p.time_step,	×
449	p.changed,	×
450	p.author_user_id	×
451	FROM user_project_permissions u JOIN project_versions p ON (u.project_id = p.project_id)	×
452	WHERE u.user_id = $1 AND u.project_id = $2 AND p.project_version = $3",	×
453	)	×
454	.await?;	×
455
456	conn.query_one(&stmt, &[&session.user.id, &project, &version])	×
457	.await?	×
458	} else {
459	let stmt = conn	8✔
460	.prepare(	8✔
461	"	8✔
462	SELECT	8✔
463	p.project_id,	8✔
464	p.id,	8✔
465	p.name,	8✔
466	p.description,	8✔
467	p.bounds,	8✔
468	p.time_step,	8✔
469	p.changed,	8✔
470	p.author_user_id	8✔
471	FROM user_project_permissions u JOIN project_versions p ON (u.project_id = p.project_id)	8✔
472	WHERE u.user_id = $1 AND u.project_id = $2 AND latest IS TRUE",	8✔
473	)	8✔
474	.await?;	8✔
475
476	conn.query_one(&stmt, &[&session.user.id, &project]).await?	8✔
477	};
478
479	let project_id = ProjectId(row.get(0));	8✔
480	let version_id = ProjectVersionId(row.get(1));	8✔
481	let name = row.get(2);	8✔
482	let description = row.get(3);	8✔
483	let bounds = row.get(4);	8✔
484	let time_step = row.get(5);	8✔
485	let changed = row.get(6);	8✔
486	let _author_id = UserId(row.get(7));	8✔
487
488	let stmt = conn	8✔
489	.prepare(	8✔
490	"	8✔
491	SELECT	8✔
492	name, workflow_id, symbology, visibility	8✔
493	FROM project_version_layers	8✔
494	WHERE project_version_id = $1	8✔
495	ORDER BY layer_index ASC",	8✔
496	)	8✔
497	.await?;	12✔
498
499	let rows = conn.query(&stmt, &[&version_id]).await?;	8✔
500
501	let mut layers = vec![];	8✔
502	for row in rows {	12✔
503	layers.push(Layer {
504	workflow: WorkflowId(row.get(1)),	4✔
505	name: row.get(0),	4✔
506	symbology: serde_json::from_value(row.get(2)).context(error::SerdeJson)?,	4✔
507	visibility: row.get(3),	4✔
508	});
509	}
510
511	Ok(Project {
512	id: project_id,	8✔
513	version: ProjectVersion {	8✔
514	id: version_id,	8✔
515	changed,	8✔
516	},	8✔
517	name,	8✔
518	description,	8✔
519	layers,	8✔
520	plots: self.load_plots(&conn, &version_id).await?,	16✔
521	bounds,	8✔
522	time_step,	8✔
523	})
524	}	20✔
525
526	async fn versions(	6✔
527	&self,	6✔
528	session: &UserSession,	6✔
529	project: ProjectId,	6✔
530	) -> Result<Vec<ProjectVersion>> {	6✔
531	let conn = self.conn_pool.get().await?;	6✔
532
533	PostgresContext::check_user_project_permission(	6✔
534	&conn,	6✔
535	session.user.id,	6✔
536	project,	6✔
537	&[	6✔
538	ProjectPermission::Read,	6✔
539	ProjectPermission::Write,	6✔
540	ProjectPermission::Owner,	6✔
541	],	6✔
542	)	6✔
543	.await?;	12✔
544
545	let stmt = conn	6✔
546	.prepare(	6✔
547	"	6✔
548	SELECT id, changed, author_user_id	6✔
549	FROM project_versions WHERE project_id = $1	6✔
550	ORDER BY latest DESC, changed DESC, author_user_id DESC",	6✔
551	)	6✔
552	.await?;	6✔
553
554	let rows = conn.query(&stmt, &[&project]).await?;	7✔
555
556	Ok(rows	6✔
557	.iter()	6✔
558	.map(\|row\| ProjectVersion {	18✔
559	id: ProjectVersionId(row.get(0)),	18✔
560	changed: row.get(1),	18✔
561	})	18✔
562	.collect())	6✔
563	}	12✔
564
565	async fn list_permissions(	4✔
566	&self,	4✔
567	session: &UserSession,	4✔
568	project: ProjectId,	4✔
569	) -> Result<Vec<UserProjectPermission>> {	4✔
570	let conn = self.conn_pool.get().await?;	4✔
571
572	PostgresContext::check_user_project_permission(	4✔
573	&conn,	4✔
574	session.user.id,	4✔
575	project,	4✔
576	&[	4✔
577	ProjectPermission::Read,	4✔
578	ProjectPermission::Write,	4✔
579	ProjectPermission::Owner,	4✔
580	],	4✔
581	)	4✔
582	.await?;	8✔
583
584	let stmt = conn	4✔
585	.prepare(	4✔
586	"	4✔
587	SELECT user_id, project_id, permission FROM user_project_permissions WHERE project_id = $1;",	4✔
588	)	4✔
589	.await?;	4✔
590
591	let rows = conn.query(&stmt, &[&project]).await?;	4✔
592
593	Ok(rows	4✔
594	.into_iter()	4✔
595	.map(\|row\| UserProjectPermission {	6✔
596	project: ProjectId(row.get(1)),	6✔
597	permission: row.get(2),	6✔
598	user: session.user.id,	6✔
599	})	6✔
600	.collect())	4✔
601	}	8✔
602
603	async fn add_permission(	2✔
604	&self,	2✔
605	session: &UserSession,	2✔
606	permission: UserProjectPermission,	2✔
607	) -> Result<()> {	2✔
608	let conn = self.conn_pool.get().await?;	2✔
609
610	PostgresContext::check_user_project_permission(	2✔
611	&conn,	2✔
612	session.user.id,	2✔
613	permission.project,	2✔
614	&[ProjectPermission::Owner],	2✔
615	)	2✔
616	.await?;	2✔
617
618	let stmt = conn	2✔
619	.prepare(	2✔
620	"	2✔
621	INSERT INTO user_project_permissions (user_id, project_id, permission)	2✔
622	VALUES ($1, $2, $3);",	2✔
623	)	2✔
624	.await?;	1✔
625
626	conn.execute(	2✔
627	&stmt,	2✔
628	&[	2✔
629	&permission.user,	2✔
630	&permission.project,	2✔
631	&permission.permission,	2✔
632	],	2✔
633	)	2✔
634	.await?;	2✔
635
636	Ok(())	2✔
637	}	4✔
638
639	async fn remove_permission(	×
640	&self,	×
641	session: &UserSession,	×
642	permission: UserProjectPermission,	×
643	) -> Result<()> {	×
644	let conn = self.conn_pool.get().await?;	×
645
646	PostgresContext::check_user_project_permission(	×
647	&conn,	×
648	session.user.id,	×
649	permission.project,	×
650	&[ProjectPermission::Owner],	×
651	)	×
652	.await?;	×
653
654	let stmt = conn	×
655	.prepare(	×
656	"	×
657	DELETE FROM user_project_permissions	×
658	WHERE user_id = $1 AND project_id = $2 AND permission = $3;",	×
659	)	×
660	.await?;	×
661
662	conn.execute(	×
663	&stmt,	×
664	&[	×
665	&session.user.id,	×
666	&permission.project,	×
667	&permission.permission,	×
668	],	×
669	)	×
670	.await?;	×
671
672	Ok(())	×
673	}	×
674	}

geo-engine / geoengine / 4056270646

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