geo-engine/geoengine | Build 6510717572 | services/src/pro/projects/postgres_projectdb.rs | Coveralls

28

async fn list_plots<Tls>(

5✔

29

    conn: &PooledConnection<'_, PostgresConnectionManager<Tls>>,

5✔

30

    project_version_id: &ProjectVersionId,

5✔

31

) -> Result<Vec<String>, ProjectDbError>

5✔

32

where

5✔

33

    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,

5✔

34

    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,

5✔

35

    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,

5✔

36

    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,

5✔

37

5✔

38

    let stmt = conn

5✔

39

        .prepare(

5✔

40

5✔

41

                    SELECT name

5✔

42

                    FROM project_version_plots

5✔

43

                    WHERE project_version_id = $1;

5✔

44

",

5✔

45

5✔

46

        .await

5✔

47

        .context(PostgresProjectDbError)?;

5✔

49

    let plot_rows = conn

5✔

50

        .query(&stmt, &[project_version_id])

5✔

51

        .await

5✔

52

        .context(PostgresProjectDbError)?;

5✔

53

    let plot_names = plot_rows.iter().map(|row| row.get(0)).collect();

5✔

54

5✔

55

    Ok(plot_names)

5✔

56

5✔

58

async fn load_plots<Tls>(

13✔

59

    conn: &PooledConnection<'_, PostgresConnectionManager<Tls>>,

13✔

60

    project_version_id: &ProjectVersionId,

13✔

61

) -> Result<Vec<Plot>, ProjectDbError>

13✔

62

where

13✔

63

    Tls: MakeTlsConnect<Socket> + Clone + Send + Sync + 'static,

13✔

64

    <Tls as MakeTlsConnect<Socket>>::Stream: Send + Sync,

13✔

65

    <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,

13✔

66

    <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,

13✔

67

13✔

68

    let stmt = conn

13✔

69

        .prepare(

13✔

70

13✔

71

                SELECT

13✔

72

                    name, workflow_id

13✔

73

                FROM project_version_plots

13✔

74

                WHERE project_version_id = $1

13✔

75

                ORDER BY plot_index ASC

13✔

76

",

13✔

77

13✔

78

        .await

15✔

79

        .context(PostgresProjectDbError)?;

13✔

81

    let rows = conn

13✔

82

        .query(&stmt, &[project_version_id])

13✔

83

        .await

15✔

84

        .context(PostgresProjectDbError)?;

13✔

86

    let plots = rows

13✔

87

        .into_iter()

13✔

88

        .map(|row| Plot {

13✔

89

            workflow: WorkflowId(row.get(1)),

6✔

90

            name: row.get(0),

6✔

91

})

13✔

92

        .collect();

13✔

93

13✔

94

    Ok(plots)

13✔

95

13✔

97

async fn update_plots(

11✔

98

    trans: &Transaction<'_>,

11✔

99

    project_id: &ProjectId,

11✔

100

    project_version_id: &ProjectVersionId,

11✔

101

    plots: &[Plot],

11✔

102

) -> Result<(), ProjectDbError> {

11✔

103

    for (idx, plot) in plots.iter().enumerate() {

11✔

104

        let stmt = trans

6✔

105

            .prepare(

6✔

106

6✔

107

                    INSERT INTO project_version_plots (

6✔

108

                        project_id,

6✔

109

                        project_version_id,

6✔

110

                        plot_index,

6✔

111

                        name,

6✔

112

                        workflow_id)

6✔

113

                    VALUES ($1, $2, $3, $4, $5);

6✔

114

",

6✔

115

6✔

116

            .await

6✔

117

            .context(PostgresProjectDbError)?;

6✔

119

        trans

6✔

120

            .execute(

6✔

121

                &stmt,

6✔

122

&[

6✔

123

                    project_id,

6✔

124

                    project_version_id,

6✔

125

                    &(idx as i32),

6✔

126

                    &plot.name,

6✔

127

                    &plot.workflow,

6✔

128

],

6✔

129

6✔

130

            .await

6✔

131

            .context(PostgresProjectDbError)?;

6✔

134

    Ok(())

11✔

135

11✔

145

    async fn list_projects(

3✔

146

        &self,

3✔

147

        options: ProjectListOptions,

3✔

148

    ) -> Result<Vec<ProjectListing>, ProjectDbError> {

3✔

149

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

3✔

151

        let stmt = conn

3✔

152

            .prepare(&format!(

3✔

153

3✔

154

        SELECT p.id, p.project_id, p.name, p.description, p.changed

3✔

155

        FROM user_permitted_projects u JOIN project_versions p ON (u.project_id = p.project_id)

3✔

156

        WHERE

3✔

157

            u.user_id = $1

3✔

158

            AND p.changed >= ALL (SELECT changed FROM project_versions WHERE project_id = p.project_id)

3✔

159

        ORDER BY p.{}

3✔

160

        LIMIT $2

3✔

161

        OFFSET $3;",

3✔

162

                options.order.to_sql_string()

3✔

163

))

3✔

164

            .await.context(PostgresProjectDbError)?;

3✔

166

        let project_rows = conn

3✔

167

            .query(

3✔

168

                &stmt,

3✔

169

&[

3✔

170

                    &self.session.user.id,

3✔

171

                    &i64::from(options.limit),

3✔

172

                    &i64::from(options.offset),

3✔

173

],

3✔

174

3✔

175

            .await

3✔

176

            .context(PostgresProjectDbError)?;

3✔

178

        let mut project_listings = vec![];

3✔

179

        for project_row in project_rows {

8✔

180

            let project_version_id = ProjectVersionId(project_row.get(0));

5✔

181

            let project_id = ProjectId(project_row.get(1));

5✔

182

            let name = project_row.get(2);

5✔

183

            let description = project_row.get(3);

5✔

184

            let changed = project_row.get(4);

5✔

186

            let stmt = conn

5✔

187

                .prepare(

5✔

188

5✔

189

                    SELECT name

5✔

190

                    FROM project_version_layers

5✔

191

                    WHERE project_version_id = $1;",

5✔

192

5✔

193

                .await

5✔

194

                .context(PostgresProjectDbError)?;

5✔

196

            let layer_rows = conn

5✔

197

                .query(&stmt, &[&project_version_id])

5✔

198

                .await

5✔

199

                .context(PostgresProjectDbError)?;

5✔

200

            let layer_names = layer_rows.iter().map(|row| row.get(0)).collect();

5✔

201

5✔

202

            project_listings.push(ProjectListing {

5✔

203

                id: project_id,

5✔

204

                name,

5✔

205

                description,

5✔

206

                layer_names,

5✔

207

                plot_names: list_plots(&conn, &project_version_id).await?,

10✔

208

                changed,

5✔

211

        Ok(project_listings)

3✔

212

6✔

214

    async fn create_project(&self, create: CreateProject) -> Result<ProjectId, ProjectDbError> {

22✔

215

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

22✔

217

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

22✔

219

        let trans = conn

22✔

220

            .build_transaction()

22✔

221

            .start()

22✔

222

            .await

22✔

223

            .context(PostgresProjectDbError)?;

22✔

225

        let stmt = trans

22✔

226

            .prepare("INSERT INTO projects (id) VALUES ($1);")

22✔

227

            .await

22✔

228

            .context(PostgresProjectDbError)?;

22✔

230

        trans

22✔

231

            .execute(&stmt, &[&project.id])

22✔

232

            .await

22✔

233

            .context(PostgresProjectDbError)?;

22✔

235

        let stmt = trans

22✔

236

            .prepare(

22✔

237

                "INSERT INTO project_versions (

22✔

238

id,

22✔

239

                    project_id,

22✔

240

                    name,

22✔

241

                    description,

22✔

242

                    bounds,

22✔

243

                    time_step,

22✔

244

                    changed)

22✔

245

                    VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP);",

22✔

246

22✔

247

            .await

51✔

248

            .context(PostgresProjectDbError)?;

22✔

250

        let version_id = ProjectVersionId::new();

22✔

251

22✔

252

        trans

22✔

253

            .execute(

22✔

254

                &stmt,

22✔

255

&[

22✔

256

                    &version_id,

22✔

257

                    &project.id,

22✔

258

                    &project.name,

22✔

259

                    &project.description,

22✔

260

                    &project.bounds,

22✔

261

                    &project.time_step,

22✔

262

],

22✔

263

22✔

264

            .await

22✔

265

            .context(PostgresProjectDbError)?;

22✔

267

        let stmt = trans

22✔

268

            .prepare(

22✔

269

                "INSERT INTO

22✔

270

                    project_version_authors (project_version_id, user_id)

22✔

271

                VALUES

22✔

272

                    ($1, $2);",

22✔

273

22✔

274

            .await

22✔

275

            .context(PostgresProjectDbError)?;

22✔

277

        trans

22✔

278

            .execute(&stmt, &[&version_id, &self.session.user.id])

22✔

279

            .await

22✔

280

            .context(PostgresProjectDbError)?;

22✔

282

        let stmt = trans

22✔

283

            .prepare(

22✔

284

                "INSERT INTO permissions (role_id, permission, project_id) VALUES ($1, $2, $3);",

22✔

285

22✔

286

            .await

28✔

287

            .context(PostgresProjectDbError)?;

22✔

289

        trans

22✔

290

            .execute(

22✔

291

                &stmt,

22✔

292

                &[&self.session.user.id, &Permission::Owner, &project.id],

22✔

293

22✔

294

            .await

22✔

295

            .context(PostgresProjectDbError)?;

22✔

297

        trans.commit().await.context(PostgresProjectDbError)?;

22✔

299

        Ok(project.id)

22✔

300

44✔

302

    async fn load_project(&self, project: ProjectId) -> Result<Project, ProjectDbError> {

13✔

303

        self.load_project_version(project, LoadVersion::Latest)

13✔

304

            .await

515✔

305

26✔

308

    async fn update_project(&self, update: UpdateProject) -> Result<(), ProjectDbError> {

11✔

309

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

12✔

311

        let trans = conn

11✔

312

            .build_transaction()

11✔

313

            .start()

11✔

314

            .await

13✔

315

            .context(PostgresProjectDbError)?;

11✔

317

        self.ensure_permission_in_tx(update.id, Permission::Owner, &trans)

11✔

318

            .await

23✔

319

            .boxed_context(AccessFailedProjectDbError { project: update.id })?;

11✔

321

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

507✔

323

        let project = project.update_project(update)?;

11✔

325

        let stmt = trans

11✔

326

            .prepare(

11✔

327

11✔

328

                INSERT INTO project_versions (

11✔

329

id,

11✔

330

                    project_id,

11✔

331

                    name,

11✔

332

                    description,

11✔

333

                    bounds,

11✔

334

                    time_step,

11✔

335

                    changed)

11✔

336

                VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP);",

11✔

337

11✔

338

            .await

12✔

339

            .context(PostgresProjectDbError)?;

11✔

341

        trans

11✔

342

            .execute(

11✔

343

                &stmt,

11✔

344

&[

11✔

345

                    &project.version.id,

11✔

346

                    &project.id,

11✔

347

                    &project.name,

11✔

348

                    &project.description,

11✔

349

                    &project.bounds,

11✔

350

                    &project.time_step,

11✔

351

],

11✔

352

11✔

353

            .await

13✔

354

            .context(PostgresProjectDbError)?;

11✔

356

        let stmt = trans

11✔

357

            .prepare(

11✔

358

                "INSERT INTO

11✔

359

                    project_version_authors (project_version_id, user_id)

11✔

360

                VALUES

11✔

361

                    ($1, $2);",

11✔

362

11✔

363

            .await

13✔

364

            .context(PostgresProjectDbError)?;

11✔

366

        trans

11✔

367

            .execute(&stmt, &[&project.version.id, &self.session.user.id])

11✔

368

            .await

11✔

369

            .context(PostgresProjectDbError)?;

11✔

371

        for (idx, layer) in project.layers.iter().enumerate() {

11✔

372

            let stmt = trans

11✔

373

                .prepare(

11✔

374

11✔

375

                INSERT INTO project_version_layers (

11✔

376

                    project_id,

11✔

377

                    project_version_id,

11✔

378

                    layer_index,

11✔

379

                    name,

11✔

380

                    workflow_id,

11✔

381

                    symbology,

11✔

382

                    visibility)

11✔

383

                VALUES ($1, $2, $3, $4, $5, $6, $7);",

11✔

384

11✔

385

                .await

15✔

386

                .context(PostgresProjectDbError)?;

11✔

388

            trans

11✔

389

                .execute(

11✔

390

                    &stmt,

11✔

391

&[

11✔

392

                        &project.id,

11✔

393

                        &project.version.id,

11✔

394

                        &(idx as i32),

11✔

395

                        &layer.name,

11✔

396

                        &layer.workflow,

11✔

397

                        &layer.symbology,

11✔

398

                        &layer.visibility,

11✔

399

],

11✔

400

11✔

401

                .await

11✔

402

                .context(PostgresProjectDbError)?;

11✔

405

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

12✔

407

        trans.commit().await.context(PostgresProjectDbError)?;

13✔

409

        Ok(())

11✔

410

22✔

412

    async fn delete_project(&self, project: ProjectId) -> Result<(), ProjectDbError> {

2✔

413

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

2✔

414

        let trans = conn

2✔

415

            .build_transaction()

2✔

416

            .start()

2✔

417

            .await

2✔

418

            .context(PostgresProjectDbError)?;

2✔

420

        self.ensure_permission_in_tx(project, Permission::Owner, &trans)

2✔

421

            .await

4✔

422

            .boxed_context(AccessFailedProjectDbError { project })?;

2✔

424

        let stmt = trans

2✔

425

            .prepare("DELETE FROM projects WHERE id = $1;")

2✔

426

            .await

2✔

427

            .context(PostgresProjectDbError)?;

2✔

429

        let rows_affected = trans

2✔

430

            .execute(&stmt, &[&project])

2✔

431

            .await

2✔

432

            .context(PostgresProjectDbError)?;

2✔

434

        trans.commit().await.context(PostgresProjectDbError)?;

2✔

436

        ensure!(

2✔

437

            rows_affected == 1,

2✔

441

        Ok(())

2✔

442

4✔

445

    async fn load_project_version(

15✔

446

        &self,

15✔

447

        project: ProjectId,

15✔

448

        version: LoadVersion,

15✔

449

    ) -> Result<Project, ProjectDbError> {

15✔

450

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

18✔

451

        let trans = conn

15✔

452

            .build_transaction()

15✔

453

            .start()

15✔

454

            .await

17✔

455

            .context(PostgresProjectDbError)?;

15✔

457

        self.ensure_permission_in_tx(project, Permission::Owner, &trans)

15✔

458

            .await

71✔

459

            .boxed_context(AccessFailedProjectDbError { project })?;

15✔

461

        let rows = if let LoadVersion::Version(version) = version {

13✔

492

            let stmt = trans

13✔

493

                .prepare(

13✔

494

13✔

495

            SELECT

13✔

496

                p.project_id,

13✔

497

                p.id,

13✔

498

                p.name,

13✔

499

                p.description,

13✔

500

                p.bounds,

13✔

501

                p.time_step,

13✔

502

                p.changed,

13✔

503

                a.user_id

13✔

504

            FROM

13✔

505

                project_versions p JOIN project_version_authors a ON (p.id = a.project_version_id)

13✔

506

            WHERE project_id = $1 AND p.changed >= ALL(

13✔

507

                SELECT changed FROM project_versions WHERE project_id = $1

13✔

508

)",

13✔

509

13✔

510

                .await

132✔

511

                .context(PostgresProjectDbError)?;

13✔

513

            let rows = trans

13✔

514

                .query(&stmt, &[&project])

13✔

515

                .await

16✔

516

                .context(PostgresProjectDbError)?;

13✔

518

            if rows.is_empty() {

13✔

520

13✔

521

13✔

522

            rows

13✔

525

        let row = &rows[0];

13✔

526

13✔

527

        let project_id = ProjectId(row.get(0));

13✔

528

        let version_id = ProjectVersionId(row.get(1));

13✔

529

        let name = row.get(2);

13✔

530

        let description = row.get(3);

13✔

531

        let bounds = row.get(4);

13✔

532

        let time_step = row.get(5);

13✔

533

        let changed = row.get(6);

13✔

534

        let _author_id = UserId(row.get(7));

13✔

536

        let stmt = trans

13✔

537

            .prepare(

13✔

538

13✔

539

        SELECT

13✔

540

            name, workflow_id, symbology, visibility

13✔

541

        FROM project_version_layers

13✔

542

        WHERE project_version_id = $1

13✔

543

        ORDER BY layer_index ASC",

13✔

544

13✔

545

            .await

284✔

546

            .context(PostgresProjectDbError)?;

13✔

548

        let rows = trans

13✔

549

            .query(&stmt, &[&version_id])

13✔

550

            .await

16✔

551

            .context(PostgresProjectDbError)?;

13✔

553

        trans.commit().await.context(PostgresProjectDbError)?;

15✔

555

        let mut layers = vec![];

13✔

556

        for row in rows {

21✔

557

            layers.push(ProjectLayer {

8✔

558

                workflow: WorkflowId(row.get(1)),

8✔

559

                name: row.get(0),

8✔

560

                symbology: row.get(2),

8✔

561

                visibility: row.get(3),

8✔

562

});

8✔

563

8✔

566

            id: project_id,

13✔

567

            version: ProjectVersion {

13✔

568

                id: version_id,

13✔

569

                changed,

13✔

570

},

13✔

571

            name,

13✔

572

            description,

13✔

573

            layers,

13✔

574

            plots: load_plots(&conn, &version_id).await?,

30✔

575

            bounds,

13✔

576

            time_step,

13✔

578

30✔

580

    async fn list_project_versions(

6✔

581

        &self,

6✔

582

        project: ProjectId,

6✔

583

    ) -> Result<Vec<ProjectVersion>, ProjectDbError> {

6✔

584

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

6✔

585

        let trans = conn

6✔

586

            .build_transaction()

6✔

587

            .start()

6✔

588

            .await

6✔

589

            .context(PostgresProjectDbError)?;

6✔

591

        self.ensure_permission_in_tx(project, Permission::Read, &trans)

6✔

592

            .await

12✔

593

            .boxed_context(AccessFailedProjectDbError { project })?;

6✔

595

        let stmt = trans

6✔

596

            .prepare(

6✔

597

6✔

598

                SELECT

6✔

599

                    p.id, p.changed, a.user_id

6✔

600

                FROM

6✔

601

                    project_versions p JOIN project_version_authors a ON (p.id = a.project_version_id)

6✔

602

                WHERE

6✔

603

                    project_id = $1

6✔

604

                ORDER BY

6✔

605

                    p.changed DESC, a.user_id DESC",

6✔

606

6✔

607

            .await.context(PostgresProjectDbError)?;

6✔

609

        let rows = trans

6✔

610

            .query(&stmt, &[&project])

6✔

611

            .await

6✔

612

            .context(PostgresProjectDbError)?;

6✔

614

        trans.commit().await.context(PostgresProjectDbError)?;

6✔

616

        Ok(rows

6✔

617

            .iter()

6✔

618

            .map(|row| ProjectVersion {

18✔

619

                id: ProjectVersionId(row.get(0)),

18✔

620

                changed: row.get(1),

18✔

621

})

18✔

622

            .collect())

6✔

623

12✔

geo-engine / geoengine / 6510717572

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

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

geo-engine / geoengine / 6510717572

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