this repo has no description
use Arc<Pool<Postgres>> so the db connection can be shared across threads. needed for ingestion
+11
-7
+6
-4
src/backfill/mod.rs
+6
-4
src/backfill/mod.rs
···
6
6
//! 4. convert cbor data to json
7
7
//! 5. store in db (limit to DB_MAX_REQ / 4 to avoid err)
8
8
9
+
use std::sync::Arc;
10
+
9
11
use ipld_core::cid::multibase::Base;
10
12
use jacquard::url::Url;
11
13
use sqlx::{Pool, Postgres, query};
···
34
36
}
35
37
36
38
pub async fn backfill(
37
-
conn: &Pool<Postgres>,
39
+
conn: Arc<Pool<Postgres>>,
38
40
time: Option<std::time::Instant>,
39
41
) -> Result<(), Error> {
40
42
let car = load_car(
···
51
53
// erase all old records and return if it fails
52
54
// we dont use diffs bc theyre complex and the overhead is minimal rn
53
55
// only real overhead is network latency which would be ~= anyway
54
-
let _ = query!("DELETE FROM records").execute(conn).await?;
56
+
let _ = query!("DELETE FROM records").execute(&*conn).await?;
55
57
56
58
let data = parse_car(&car)
57
59
.await?
···
85
87
},
86
88
);
87
89
88
-
if let Err(err) = query.build().execute(conn).await {
90
+
if let Err(err) = query.build().execute(&*conn).await {
89
91
// couldnt backfill so go nuclear
90
92
// this is program startup so its prolly safe lol
91
93
println!("Got error \"{}\"\nDeleting records and exiting...", err);
92
-
let _ = query!("DELETE FROM records").execute(conn).await?;
94
+
let _ = query!("DELETE FROM records").execute(&*conn).await?;
93
95
panic!()
94
96
};
95
97
}
+4
-2
src/db.rs
+4
-2
src/db.rs
···
1
1
//! create a connection pool and setup tables before making avaliable
2
2
3
+
use std::sync::Arc;
4
+
3
5
use crate::config;
4
6
use sqlx::{Pool, Postgres, migrate, postgres::PgPool};
5
7
6
-
pub async fn conn() -> Pool<Postgres> {
8
+
pub async fn conn() -> Arc<Pool<Postgres>> {
7
9
let conn = match PgPool::connect(&config::DATABASE_URL).await {
8
10
Ok(val) => val,
9
11
Err(err) => {
···
15
17
panic!("Error running migrations: {}", err);
16
18
});
17
19
18
-
conn
20
+
Arc::new(conn)
19
21
}