For now? I'm experimenting on an old concept.
Migrate to SQLX: done.
+234
-217
+1
-1
server/src/client_communication.rs
+1
-1
server/src/client_communication.rs
···
34
34
use crate::user::User;
35
35
use crate::{
36
36
AppState, LuminaError, authentication_error_elog, error_elog, http_code_elog, incoming_elog,
37
-
info_elog, registration_error_elog, warn_elog,
37
+
info_elog, registration_error_elog,
38
38
};
39
39
use base64::Engine;
40
40
use base64::engine::general_purpose::STANDARD;
+110
-98
server/src/database.rs
+110
-98
server/src/database.rs
···
27
27
use bb8::Pool;
28
28
use bb8_redis::RedisConnectionManager;
29
29
use cynthia_con::{CynthiaColors, CynthiaStyles};
30
+
use sqlx::Postgres;
30
31
use sqlx::postgres::PgPool;
31
32
use std::time::Duration;
32
-
use sqlx::{Postgres};
33
33
34
34
struct DatabaseConfig {
35
35
postgres_username: String,
···
37
37
postgres_host: String,
38
38
postgres_port: u16,
39
39
postgres_dbname: String,
40
-
redis_url: String,
41
40
}
42
41
43
42
pub(crate) async fn setup() -> Result<PgConn, LuminaError> {
···
64
63
};
65
64
66
65
{
67
-
let pg_config: DatabaseConfig = {
68
-
let mut uuu = (
69
-
"unspecified database".to_string(),
70
-
"unspecified host".to_string(),
71
-
"unknown port".to_string(),
72
-
);
73
-
let mut pg_config = DatabaseConfig {
74
-
postgres_username: std::env::var("LUMINA_POSTGRES_USERNAME")
75
-
.unwrap_or("lumina".to_string()),
76
-
postgres_password: std::env::var("LUMINA_POSTGRES_PASSWORD").ok(),
77
-
postgres_host: std::env::var("LUMINA_POSTGRES_HOST")
78
-
.unwrap_or("localhost".to_string()),
79
-
postgres_port: std::env::var("LUMINA_POSTGRES_PORT")
80
-
.ok()
81
-
.and_then(|p| p.parse::<u16>().ok())
82
-
.unwrap_or(5432),
83
-
postgres_dbname: std::env::var("LUMINA_POSTGRES_DATABASE")
84
-
.unwrap_or("lumina_config".to_string()),
85
-
redis_url,
86
-
};
87
-
pg_config.postgres_username =
88
-
std::env::var("LUMINA_POSTGRES_USERNAME").unwrap_or("lumina".to_string());
89
-
let dbname =
90
-
std::env::var("LUMINA_POSTGRES_DATABASE").unwrap_or("lumina_config".to_string());
91
-
uuu.0 = dbname.clone();
92
-
pg_config.postgres_dbname = dbname;
93
-
let port = match std::env::var("LUMINA_POSTGRES_PORT") {
94
-
Err(..) => {
95
-
warn_elog!(
96
-
ev_log,
97
-
"No Postgres database port provided under environment variable 'LUMINA_POSTGRES_PORT'. Using default value '5432'."
98
-
);
99
-
"5432".to_string()
100
-
}
101
-
Ok(c) => c,
102
-
};
103
-
uuu.2 = port.clone();
104
-
// Parse the port as u16, if it fails, return an error
105
-
pg_config.postgres_port = port
106
-
.parse::<u16>()
107
-
.map_err(|_| LuminaError::ConfInvalid(LUMINA_POSTGRES_PORT))?;
108
-
match std::env::var("LUMINA_POSTGRES_HOST") {
109
-
Ok(val) => {
110
-
uuu.1 = val.clone();
111
-
pg_config.postgres_host = val;
112
-
}
113
-
Err(_) => {
114
-
warn_elog!(
115
-
ev_log,
116
-
"No Postgres database host provided under environment variable 'LUMINA_POSTGRES_HOST'. Using default value 'localhost'."
117
-
);
118
-
// Default to localhost if not set
119
-
uuu.1 = "localhost".to_string();
120
-
pg_config.postgres_host = "localhost".to_string();
121
-
}
122
-
};
123
-
match std::env::var("LUMINA_POSTGRES_PASSWORD") {
124
-
Ok(val) => {
125
-
pg_config.postgres_password = Some(val);
126
-
}
127
-
Err(_) => {
128
-
warn_elog!(
129
-
ev_log,
130
-
"No Postgres database password provided under environment variable 'LUMINA_POSTGRES_PASSWORD'. Trying passwordless authentication."
131
-
);
132
-
}
133
-
};
66
+
let uri = if let Ok(uri) = std::env::var("DATABASE_URL") {
134
67
info_elog!(
135
68
ev_log,
136
-
"Using Postgres database at: {} on host: {} at port: {}",
137
-
uuu.0.color_bright_cyan().style_bold(),
138
-
uuu.1.color_bright_cyan().style_bold(),
139
-
uuu.2.color_bright_cyan().style_bold(),
69
+
"DATABASE_URL set, using that to connect to Postgres",
140
70
);
141
-
pg_config
142
-
};
71
+
uri
72
+
} else {
73
+
let pg_config: DatabaseConfig = {
74
+
let mut uuu = (
75
+
"unspecified database".to_string(),
76
+
"unspecified host".to_string(),
77
+
"unknown port".to_string(),
78
+
);
79
+
let mut pg_config = DatabaseConfig {
80
+
postgres_username: std::env::var("LUMINA_POSTGRES_USERNAME")
81
+
.unwrap_or("lumina".to_string()),
82
+
postgres_password: std::env::var("LUMINA_POSTGRES_PASSWORD").ok(),
83
+
postgres_host: std::env::var("LUMINA_POSTGRES_HOST")
84
+
.unwrap_or("localhost".to_string()),
85
+
postgres_port: std::env::var("LUMINA_POSTGRES_PORT")
86
+
.ok()
87
+
.and_then(|p| p.parse::<u16>().ok())
88
+
.unwrap_or(5432),
89
+
postgres_dbname: std::env::var("LUMINA_POSTGRES_DATABASE")
90
+
.unwrap_or("lumina_config".to_string()),
91
+
};
92
+
pg_config.postgres_username =
93
+
std::env::var("LUMINA_POSTGRES_USERNAME").unwrap_or("lumina".to_string());
94
+
let dbname = std::env::var("LUMINA_POSTGRES_DATABASE")
95
+
.unwrap_or("lumina_config".to_string());
96
+
uuu.0 = dbname.clone();
97
+
pg_config.postgres_dbname = dbname;
98
+
let port = match std::env::var("LUMINA_POSTGRES_PORT") {
99
+
Err(..) => {
100
+
warn_elog!(
101
+
ev_log,
102
+
"No Postgres database port provided under environment variable 'LUMINA_POSTGRES_PORT'. Using default value '5432'."
103
+
);
104
+
"5432".to_string()
105
+
}
106
+
Ok(c) => c,
107
+
};
108
+
uuu.2 = port.clone();
109
+
// Parse the port as u16, if it fails, return an error
110
+
pg_config.postgres_port = port
111
+
.parse::<u16>()
112
+
.map_err(|_| LuminaError::ConfInvalid(LUMINA_POSTGRES_PORT))?;
113
+
match std::env::var("LUMINA_POSTGRES_HOST") {
114
+
Ok(val) => {
115
+
uuu.1 = val.clone();
116
+
pg_config.postgres_host = val;
117
+
}
118
+
Err(_) => {
119
+
warn_elog!(
120
+
ev_log,
121
+
"No Postgres database host provided under environment variable 'LUMINA_POSTGRES_HOST'. Using default value 'localhost'."
122
+
);
123
+
// Default to localhost if not set
124
+
uuu.1 = "localhost".to_string();
125
+
pg_config.postgres_host = "localhost".to_string();
126
+
}
127
+
};
128
+
match std::env::var("LUMINA_POSTGRES_PASSWORD") {
129
+
Ok(val) => {
130
+
pg_config.postgres_password = Some(val);
131
+
}
132
+
Err(_) => {
133
+
warn_elog!(
134
+
ev_log,
135
+
"No Postgres database password provided under environment variable 'LUMINA_POSTGRES_PASSWORD'. Trying passwordless authentication."
136
+
);
137
+
}
138
+
};
139
+
info_elog!(
140
+
ev_log,
141
+
"Using Postgres database at: {} on host: {} at port: {}",
142
+
uuu.0.color_bright_cyan().style_bold(),
143
+
uuu.1.color_bright_cyan().style_bold(),
144
+
uuu.2.color_bright_cyan().style_bold(),
145
+
);
146
+
pg_config
147
+
};
143
148
144
-
// Create Postgres connection pool
145
-
let uri = format!(
146
-
"postgres://{}{}@{}:{}/{}",
147
-
pg_config.postgres_username,
148
-
pg_config
149
-
.postgres_password
150
-
.as_deref()
151
-
.map(|a| format!(":{}", a))
152
-
.unwrap_or_default(),
153
-
pg_config.postgres_host,
154
-
pg_config.postgres_port,
155
-
pg_config.postgres_dbname
156
-
);
149
+
// Create Postgres connection pool
150
+
format!(
151
+
"postgres://{}{}@{}:{}/{}",
152
+
pg_config.postgres_username,
153
+
pg_config
154
+
.postgres_password
155
+
.as_deref()
156
+
.map(|a| format!(":{}", a))
157
+
.unwrap_or_default(),
158
+
pg_config.postgres_host,
159
+
pg_config.postgres_port,
160
+
pg_config.postgres_dbname
161
+
)
162
+
};
157
163
let pg_pool: sqlx::Pool<Postgres> = PgPool::connect(uri.as_str()).await?;
158
164
{
159
165
// This is where previously the database schema was created if it did not exist, but now
···
376
382
.query_async(&mut **redis_conn)
377
383
.await
378
384
.unwrap_or(None)
379
-
.map(|a: String| time::OffsetDateTime::parse(a.as_str(), &time::format_description::well_known::Rfc3339));
385
+
.map(|a: String| {
386
+
time::OffsetDateTime::parse(a.as_str(), &time::format_description::well_known::Rfc3339)
387
+
});
380
388
381
389
let query = if let Some(Ok(timestamp)) = last_check {
382
-
sqlx::query!("SELECT DISTINCT tlid FROM timelines WHERE timestamp > $1", timestamp)
383
-
.fetch_all(
384
-
pg_pool,
385
-
)
386
-
.await
390
+
sqlx::query!(
391
+
"SELECT DISTINCT tlid FROM timelines WHERE timestamp > $1",
392
+
timestamp
393
+
)
394
+
.fetch_all(pg_pool)
395
+
.await
387
396
} else if let Some(Err(_)) = last_check {
388
-
panic!("timeline_cache_last_check returned an error, this means there's probably been tampering with the Redis DB.");
397
+
panic!(
398
+
"timeline_cache_last_check returned an error, this means there's probably been tampering with the Redis DB."
399
+
);
389
400
} else {
390
401
// First run, don't invalidate anything
391
402
let _: () = redis::cmd("SET")
···
403
414
match query {
404
415
Ok(timelines) => {
405
416
for timeline in timelines {
406
-
407
417
let _ = timeline::invalidate_timeline_cache(redis_conn, timeline.tlid).await;
408
418
}
409
419
···
444
454
/// ```rust
445
455
/// Vec<(String, String)> // (email, username)
446
456
/// ```
447
-
pub async fn list_users_and_emails(pool: &PgPool) -> Result<Vec<(String, String)>, sqlx::Error> {
457
+
pub async fn list_users_and_emails(
458
+
pool: &PgPool,
459
+
) -> Result<Vec<(String, String)>, sqlx::Error> {
448
460
let recs = sqlx::query!(
449
461
r#"
450
462
SELECT email, username
···
457
469
for rec in recs {
458
470
res.push((rec.email, rec.username));
459
471
}
460
-
Ok(res)
472
+
Ok(res)
461
473
}
462
474
}
+2
server/src/errors.rs
+2
server/src/errors.rs
···
38
38
RegexError,
39
39
SerializationError(serde_json::Error),
40
40
JoinFaillure,
41
+
AuthenticationNoSuchUser,
41
42
}
42
43
43
44
impl From<LuminaDbError> for LuminaError {
···
114
115
LuminaError::RegexError => "Regex error".to_string(),
115
116
LuminaError::SerializationError(s) => format!("Serialization error: {}", s),
116
117
LuminaError::JoinFaillure => "Process join failure".to_string(),
118
+
LuminaError::AuthenticationNoSuchUser => "No such user".to_string(),
117
119
LuminaError::Unknown => "Unknown error".to_string(),
118
120
}
119
121
)
+18
-12
server/src/helpers/events.rs
+18
-12
server/src/helpers/events.rs
···
26
26
use crate::LuminaError;
27
27
use crate::database::PgConn;
28
28
use cynthia_con::{CynthiaColors, CynthiaStyles};
29
-
use time::OffsetDateTime;
29
+
use time::{OffsetDateTime, PrimitiveDateTime};
30
30
31
31
/// Levels of logging supported by the Logger.
32
32
#[derive(Debug)]
···
86
86
/// asynchronously inserts a log entry in the logs table.
87
87
pub async fn log(&self, level: EventType, message: &str) {
88
88
// Get the current timestamp.
89
-
let now = OffsetDateTime::now_utc();
89
+
let now_odt = OffsetDateTime::now_utc();
90
+
let now_pdt = PrimitiveDateTime::new(now_odt.date(), now_odt.time());
90
91
91
92
// Determine the appropriate prefix for stdout.
92
93
// These prefixes are colored and styled matching helpers::prefixes().
···
166
167
.chars()
167
168
.filter(|c| !c.is_control() || c.is_whitespace())
168
169
.collect();
169
-
let ts = now
170
-
.format(&time::format_description::well_known::Rfc3339)
171
-
.unwrap();
172
-
173
-
if let Ok(pg_conn) = db_conn.postgres_pool.get().await {
174
-
let _ = pg_conn
175
-
.execute(
176
-
"INSERT INTO logs (type, message, timestamp) VALUES ($1, $2, $3)",
177
-
&[&level_str, &message_db, &ts],
170
+
match sqlx::query!(
171
+
"INSERT INTO logs (type, message, timestamp) VALUES ($1, $2, $3)",
172
+
&level_str,
173
+
&message_db,
174
+
&now_pdt,
175
+
)
176
+
.execute(&db_conn.postgres_pool)
177
+
.await
178
+
{
179
+
Ok(_) => (),
180
+
Err(p) => {
181
+
panic!(
182
+
"{0}\n\n\n{1}\n\n\n\n{0}",
183
+
"Could not write logs to database! Crashing.", p
178
184
)
179
-
.await;
185
+
}
180
186
}
181
187
}
182
188
EventLogger::OnlyStdout => {
+10
-19
server/src/main.rs
+10
-19
server/src/main.rs
···
156
156
error_elog!(ev_log, "While connecting to postgres database: {}", a);
157
157
None
158
158
}
159
-
Err(LuminaError::Bb8RunErrorPg(a)) => {
160
-
error_elog!(ev_log, "While setting up database pool: {}", a);
161
-
None
162
-
}
163
159
Err(LuminaError::DbError(crate::errors::LuminaDbError::Redis(a))) => {
164
160
error_elog!(ev_log, "While connecting to Redis: {}", a);
165
161
None
···
203
199
if cfg!(debug_assertions) {
204
200
let redis_pool = db.get_redis_pool();
205
201
let mut redis_conn = redis_pool.get().await.unwrap();
206
-
timeline::invalidate_timeline_cache(
207
-
&mut redis_conn,
208
-
"00000000-0000-0000-0000-000000000000",
209
-
)
210
-
.await
211
-
.unwrap();
202
+
timeline::invalidate_timeline_cache(&mut redis_conn, Uuid::nil())
203
+
.await
204
+
.unwrap();
212
205
let global = timeline::fetch_timeline_post_ids(
213
206
ev_log.clone(),
214
207
&db,
215
-
"00000000-0000-0000-0000-000000000000",
208
+
&Uuid::nil(),
216
209
None,
217
210
)
218
211
.await
···
227
220
228
221
match db.recreate().await.into() {
229
222
DbConn::PgsqlConnection(pg_pool, _) => {
230
-
let client = pg_pool.get().await.unwrap();
231
223
// Insert Hello World post and timeline entry if not exists
232
224
let user_1_: Result<user::User, LuminaError> =
233
225
match user::User::create_user(
···
277
269
println!(
278
270
"Created two users with password 'MyTestPassw9292!' and usernames 'testuser1' and 'testuser2'."
279
271
);
280
-
let _ = client
281
-
.execute(
282
-
"INSERT INTO post_text (id, author_id, content, created_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP) ON CONFLICT (id) DO NOTHING",
283
-
&[&generated_uuid, &user_1.id, &hello_content],
272
+
sqlx::query!("INSERT INTO post_text (id, author_id, content, created_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP) ON CONFLICT (id) DO NOTHING",
273
+
&generated_uuid, &user_1.id, &hello_content
284
274
)
285
-
.await;
275
+
.execute(&pg_pool)
276
+
.await.unwrap_or_default();
286
277
let add_clone = ev_log.clone();
287
278
timeline::add_to_timeline(
288
279
add_clone,
289
280
&db,
290
-
"00000000-0000-0000-0000-000000000000",
291
-
generated_uuid.to_string().as_str(),
281
+
&Uuid::nil(),
282
+
&generated_uuid,
292
283
)
293
284
.await
294
285
.unwrap_or(());
+4
-5
server/src/tests.rs
+4
-5
server/src/tests.rs
···
16
16
* along with this program. If not, see <https://www.gnu.org/licenses/>.
17
17
*/
18
18
19
+
use uuid::Uuid;
20
+
19
21
use crate::database::{self, DatabaseConnections};
20
22
use crate::errors::LuminaError;
21
23
use crate::timeline;
···
26
28
let result = database::setup()
27
29
.await
28
30
.expect("Database setup should succeed.");
29
-
assert!(
30
-
result.get_postgres_pool().get().await.is_ok(),
31
-
"Should get Postgres connection"
32
-
);
33
31
assert!(
34
32
result.get_redis_pool().get().await.is_ok(),
35
33
"Should get Redis connection"
···
75
73
let db = database::setup().await.expect("DB setup");
76
74
let redis_pool = db.get_redis_pool();
77
75
let mut conn = redis_pool.get().await.expect("Redis conn");
78
-
let timeline_id = "test-timeline-invalidation";
76
+
// Global timeline
77
+
let timeline_id = Uuid::nil();
79
78
80
79
// Set a test cache key
81
80
let cache_key = format!("timeline_cache:{}:page:0", timeline_id);
+44
-48
server/src/timeline.rs
+44
-48
server/src/timeline.rs
···
175
175
async fn fetch_timeline_total_count(db: &DbConn, timeline_id: &str) -> Result<usize, LuminaError> {
176
176
match db {
177
177
DbConn::PgsqlConnection(pg_pool, _redis_pool) => {
178
-
let client = pg_pool.get().await?;
179
178
let timeline_uuid = Uuid::parse_str(timeline_id).map_err(|_| LuminaError::UUidError)?;
180
-
let row = client
181
-
.query_one(
182
-
"SELECT COUNT(*) FROM timelines WHERE tlid = $1",
183
-
&[&timeline_uuid],
184
-
)
185
-
.await?;
179
+
let row = sqlx::query!(
180
+
"SELECT COUNT(*) AS count FROM timelines WHERE tlid = $1",
181
+
&timeline_uuid
182
+
)
183
+
.fetch_one(pg_pool)
184
+
.await?;
186
185
187
-
let count: i64 = row.get(0);
186
+
let count: i64 = row.count.unwrap_or(0);
188
187
Ok(count as usize)
189
188
}
190
189
}
···
199
198
) -> Result<Vec<String>, LuminaError> {
200
199
match db {
201
200
DbConn::PgsqlConnection(pg_pool, _redis_pool) => {
202
-
let client = pg_pool.get().await?;
203
201
let timeline_uuid = Uuid::parse_str(timeline_id).map_err(|_| LuminaError::UUidError)?;
204
-
let rows = client
205
-
.query(
202
+
let rows =
203
+
sqlx::query!(
206
204
"SELECT item_id FROM timelines WHERE tlid = $1 ORDER BY timestamp DESC LIMIT $2 OFFSET $3",
207
-
&[&timeline_uuid, &(limit as i64), &(offset as i64)],
205
+
&timeline_uuid, &(limit as i64), &(offset as i64),
208
206
)
207
+
.fetch_all(pg_pool)
209
208
.await
210
209
?;
211
210
212
211
let post_ids = rows
213
212
.into_iter()
214
-
.map(|row| row.get::<_, Uuid>(0).to_string())
213
+
.map(|row| row.item_id.to_string())
215
214
.collect();
216
215
Ok(post_ids)
217
216
}
···
223
222
pub async fn fetch_timeline_post_ids(
224
223
event_logger: EventLogger,
225
224
db: &DbConn,
226
-
timeline_id: &str,
225
+
timeline: &Uuid,
227
226
page: Option<usize>,
228
227
) -> Result<(Vec<String>, usize, bool), LuminaError> {
228
+
let timeline_id = timeline.to_string();
229
229
let page = page.unwrap_or(0);
230
230
let offset = page * TIMELINE_PAGE_SIZE;
231
231
···
241
241
.await?;
242
242
243
243
// Check if this timeline should be cached
244
-
let should_cache = is_high_traffic_timeline(&mut redis_conn, timeline_id).await?;
244
+
let should_cache = is_high_traffic_timeline(&mut redis_conn, &timeline_id).await?;
245
245
246
246
// Try to get from cache if it's a high-traffic timeline
247
247
if should_cache
248
248
&& let Some(cached_page) =
249
-
get_cached_timeline_page(&mut redis_conn, timeline_id, page).await?
249
+
get_cached_timeline_page(&mut redis_conn, &timeline_id, page).await?
250
250
{
251
251
let has_more = (page + 1) * TIMELINE_PAGE_SIZE < cached_page.total_count;
252
252
return Ok((cached_page.post_ids, cached_page.total_count, has_more));
···
255
255
// Cache miss or low-traffic timeline - fetch from database
256
256
if timeline_id == GLOBAL_TIMELINE_ID || should_cache {
257
257
// Get total count
258
-
let total_count = fetch_timeline_total_count(db, timeline_id).await?;
258
+
let total_count = fetch_timeline_total_count(db, &timeline_id).await?;
259
259
260
260
// Get page data
261
-
let post_ids = fetch_timeline_from_db(db, timeline_id, offset, TIMELINE_PAGE_SIZE).await?;
261
+
let post_ids = fetch_timeline_from_db(db, &timeline_id, offset, TIMELINE_PAGE_SIZE).await?;
262
262
263
263
// Cache the result if it's high-traffic
264
264
if should_cache {
265
-
match cache_timeline_page(&mut redis_conn, timeline_id, page, &post_ids, total_count)
265
+
match cache_timeline_page(&mut redis_conn, &timeline_id, page, &post_ids, total_count)
266
266
.await
267
267
{
268
268
Ok(_) => info_elog!(
···
324
324
);
325
325
// For now, only global timeline is supported.
326
326
if timeline_name == "global" {
327
-
let timeline_uuid =
327
+
let global_timeline_uuid =
328
328
Uuid::parse_str(GLOBAL_TIMELINE_ID).map_err(|_| LuminaError::UUidError)?;
329
329
let (post_ids, total_count, has_more) =
330
-
fetch_timeline_post_ids(event_logger, db, GLOBAL_TIMELINE_ID, page).await?;
331
-
Ok((timeline_uuid, post_ids, total_count, has_more))
330
+
fetch_timeline_post_ids(event_logger, db, &global_timeline_uuid, page).await?;
331
+
Ok((global_timeline_uuid, post_ids, total_count, has_more))
332
332
} else {
333
333
// Handle other timelines in the future
334
334
error_elog!(
···
344
344
pub async fn add_to_timeline(
345
345
event_logger: EventLogger,
346
346
db: &DbConn,
347
-
timeline_id: &str,
348
-
item_id: &str,
347
+
timeline: &Uuid,
348
+
item: &Uuid,
349
349
) -> Result<(), LuminaError> {
350
350
// Add to database
351
351
match db {
352
352
DbConn::PgsqlConnection(pg_pool, redis_pool) => {
353
-
let client = pg_pool.get().await?;
354
-
let timeline_uuid = Uuid::parse_str(timeline_id).map_err(|_| LuminaError::UUidError)?;
355
-
let item_uuid = Uuid::parse_str(item_id).map_err(|_| LuminaError::UUidError)?;
356
-
client
357
-
.execute(
358
-
"INSERT INTO timelines (tlid, item_id, timestamp) VALUES ($1, $2, NOW())",
359
-
&[&timeline_uuid, &item_uuid],
360
-
)
361
-
.await?;
353
+
sqlx::query!(
354
+
"INSERT INTO timelines (tlid, item_id, timestamp) VALUES ($1, $2, NOW())",
355
+
*timeline,
356
+
item,
357
+
)
358
+
.execute(pg_pool)
359
+
.await?;
362
360
363
361
// Invalidate cache
364
362
let mut redis_conn = redis_pool.get().await?;
365
-
if let Err(e) = invalidate_timeline_cache(&mut redis_conn, timeline_id).await {
363
+
if let Err(e) = invalidate_timeline_cache(&mut redis_conn, *timeline).await {
366
364
error_elog!(
367
365
event_logger,
368
366
"Failed to invalidate cache for timeline {}: {:?}",
369
-
timeline_id,
367
+
timeline.to_string(),
370
368
e
371
369
);
372
370
}
···
381
379
pub async fn remove_from_timeline(
382
380
event_logger: EventLogger,
383
381
db: &DbConn,
384
-
timeline_id: &str,
385
-
item_id: &str,
382
+
timeline: &Uuid,
383
+
item: &Uuid,
386
384
) -> Result<(), LuminaError> {
387
385
// Remove from database
388
386
match db {
389
387
DbConn::PgsqlConnection(pg_pool, redis_pool) => {
390
-
let client = pg_pool.get().await?;
391
-
let timeline_uuid = Uuid::parse_str(timeline_id).map_err(|_| LuminaError::UUidError)?;
392
-
let item_uuid = Uuid::parse_str(item_id).map_err(|_| LuminaError::UUidError)?;
393
-
client
394
-
.execute(
395
-
"DELETE FROM timelines WHERE tlid = $1 AND item_id = $2",
396
-
&[&timeline_uuid, &item_uuid],
397
-
)
398
-
.await?;
388
+
sqlx::query!(
389
+
"DELETE FROM timelines WHERE tlid = $1 AND item_id = $2",
390
+
&timeline,
391
+
&item
392
+
)
393
+
.execute(pg_pool)
394
+
.await?;
399
395
400
396
// Invalidate cache
401
397
let mut redis_conn = redis_pool.get().await?;
402
-
if let Err(e) = invalidate_timeline_cache(&mut redis_conn, timeline_id).await {
398
+
if let Err(e) = invalidate_timeline_cache(&mut redis_conn, *timeline).await {
403
399
error_elog!(
404
400
event_logger,
405
401
"Failed to invalidate cache for timeline {}: {:?}",
406
-
timeline_id,
402
+
timeline.to_string(),
407
403
e
408
404
);
409
405
}
+45
-34
server/src/user.rs
+45
-34
server/src/user.rs
···
22
22
use crate::{LuminaError, database::DbConn, helpers::events::EventLogger, info_elog};
23
23
use cynthia_con::CynthiaColors;
24
24
use uuid::Uuid;
25
-
use crate::database::DatabaseConnections;
26
25
27
26
#[derive(Debug, Clone)]
28
27
pub struct User {
···
63
62
async fn get_hashed_password(self, database: &DbConn) -> Result<String, LuminaError> {
64
63
match database {
65
64
DbConn::PgsqlConnection(pg_pool, _) => {
66
-
67
65
let row = sqlx::query!("SELECT password FROM users WHERE id = $1", &self.id)
68
66
.fetch_one(pg_pool)
69
67
.await?;
···
96
94
let username_exists =
97
95
sqlx::query!("SELECT * FROM users WHERE username = $1", &username)
98
96
.fetch_optional(pg_pool)
99
-
.await?;
97
+
.await?;
100
98
if !username_exists.is_none() {
101
99
return Err(LuminaError::RegisterUsernameInUse);
102
100
}
···
118
116
identifier: String,
119
117
db: &DbConn,
120
118
) -> Result<User, LuminaError> {
121
-
let identifyer_type = if identifier.contains('@') {
122
-
"email"
119
+
let DbConn::PgsqlConnection(pg_pool, _) = db;
120
+
// todo: Find a way to not repeat here, without it 'never matching' (which is what happens if you
121
+
// parameterise the left side of the WHERE...)
122
+
if identifier.contains('@') {
123
+
match sqlx::query!("SELECT id, email, username, coalesce(foreign_instance_id, '') as foreign_instance_id FROM users WHERE email = $1", &identifier).fetch_optional(pg_pool).await?
124
+
{
125
+
None => Err(LuminaError::AuthenticationNoSuchUser),
126
+
Some(user) => Ok(
127
+
User {
128
+
id: user.id,
129
+
email: user.email,
130
+
username: user.username,
131
+
foreign_instance_id: user.foreign_instance_id.unwrap_or("".to_string()),
132
+
}
133
+
),
134
+
}
123
135
} else {
124
-
"username"
125
-
};
126
-
match db {
127
-
DbConn::PgsqlConnection(pg_pool, _) => {
128
-
let user = sqlx::query!("SELECT id, email, username, coalesce(foreign_instance_id, '') as foreign_instance_id FROM users WHERE $1 = $2", identifyer_type, &identifier)
129
-
130
-
.fetch_one(pg_pool)
131
-
.await
132
-
?;
133
-
Ok(User {
134
-
id: user.id,
135
-
email: user.email,
136
-
username: user.username,
137
-
foreign_instance_id: user.foreign_instance_id.unwrap_or("".to_string()),
138
-
})
139
-
}
136
+
match sqlx::query!("SELECT id, email, username, coalesce(foreign_instance_id, '') as foreign_instance_id FROM users WHERE username = $1", &identifier).fetch_optional(pg_pool).await?
137
+
{
138
+
None => Err(LuminaError::AuthenticationNoSuchUser),
139
+
Some(user) => Ok(
140
+
User {
141
+
id: user.id,
142
+
email: user.email,
143
+
username: user.username,
144
+
foreign_instance_id: user.foreign_instance_id.unwrap_or("".to_string()),
145
+
}
146
+
)
147
+
}
140
148
}
141
149
}
142
150
···
151
159
DbConn::PgsqlConnection(pg_pool, _) => {
152
160
let session_key = Uuid::new_v4().to_string();
153
161
let id = sqlx::query!(
154
-
"INSERT INTO sessions (user_id, session_key) VALUES ($1, $2) RETURNING id",
155
-
&user_id, &session_key,
156
-
).fetch_one(pg_pool)
157
-
.await?;
162
+
"INSERT INTO sessions (user_id, session_key) VALUES ($1, $2) RETURNING id",
163
+
&user_id,
164
+
&session_key,
165
+
)
166
+
.fetch_one(pg_pool)
167
+
.await?;
158
168
info_elog!(
159
169
ev_log,
160
170
"New session created by {}",
···
229
239
.unwrap_or(false);
230
240
if username_exists {
231
241
// Fallback to DB check if in bloom filter
232
-
let username_db = sqlx::query!("SELECT * FROM users WHERE username = $1", &username)
233
-
.fetch_optional(pg_pool)
234
-
.await?;
242
+
let username_db =
243
+
sqlx::query!("SELECT * FROM users WHERE username = $1", &username)
244
+
.fetch_optional(pg_pool)
245
+
.await?;
235
246
if !username_db.is_none() {
236
247
return Err(LuminaError::RegisterUsernameInUse);
237
248
}
238
249
}
239
250
// Fallback to DB check if not in bloom filter
240
-
let email_db =
241
-
sqlx::query!("SELECT * FROM users WHERE email = $1", &email)
242
-
.fetch_optional(pg_pool)
251
+
let email_db = sqlx::query!("SELECT * FROM users WHERE email = $1", &email)
252
+
.fetch_optional(pg_pool)
243
253
.await?;
244
254
if !email_db.is_none() {
245
255
// Update bloom filter after DB check
···
251
261
.unwrap_or(());
252
262
return Err(LuminaError::RegisterEmailInUse);
253
263
}
254
-
let username_db = sqlx::query!("SELECT * FROM users WHERE username = $1", &username)
255
-
.fetch_optional(pg_pool)
256
-
.await?;
264
+
let username_db =
265
+
sqlx::query!("SELECT * FROM users WHERE username = $1", &username)
266
+
.fetch_optional(pg_pool)
267
+
.await?;
257
268
if !username_db.is_none() {
258
269
let _: () = redis::cmd("BF.ADD")
259
270
.arg(&username_key)