Initial Commit
This commit is contained in:
343
src/lib/outbound/postgres/user_repository.rs
Normal file
343
src/lib/outbound/postgres/user_repository.rs
Normal file
@@ -0,0 +1,343 @@
|
||||
use anyhow::Context;
|
||||
use sqlx::postgres::PgDatabaseError;
|
||||
use sqlx::Executor;
|
||||
use sqlx::QueryBuilder;
|
||||
use sqlx::Row;
|
||||
|
||||
use crate::domain::api::models::user::*;
|
||||
use crate::domain::api::ports::UserRepository;
|
||||
|
||||
use super::Postgres;
|
||||
|
||||
impl UserRepository for Postgres {
|
||||
async fn create_user(&self, req: CreateUserRequest) -> Result<User, CreateUserError> {
|
||||
let mut tx = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.context("Failed to start sql transaction")?;
|
||||
|
||||
let id = uuid::Uuid::new_v4();
|
||||
let email = req.email();
|
||||
let password = req.password();
|
||||
|
||||
let query = sqlx::query(
|
||||
r#"
|
||||
INSERT INTO users (id, email, password)
|
||||
VALUES ($1, $2, $3)
|
||||
RETURNING
|
||||
id,
|
||||
email,
|
||||
password,
|
||||
verified
|
||||
"#,
|
||||
)
|
||||
.bind(id)
|
||||
.bind(email.to_string())
|
||||
.bind(password.to_string());
|
||||
|
||||
let execute = tx.execute(query).await;
|
||||
|
||||
if let Err(ref e) = execute {
|
||||
if let Some(e) = e.as_database_error() {
|
||||
let e = e.downcast_ref::<PgDatabaseError>();
|
||||
if e.code() == "23505" {
|
||||
return Err(CreateUserError::Duplicate {
|
||||
email: email.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// fallthrough and handle other errors with `?`
|
||||
execute.map_err(|e| CreateUserError::Unknown(e.into()))?;
|
||||
|
||||
tx.commit()
|
||||
.await
|
||||
.context("failed to commit SQL transaction")?;
|
||||
|
||||
Ok(User::new(id, email.clone(), password.clone(), false.into()))
|
||||
}
|
||||
|
||||
async fn create_activation_token(&self, ent: &User) -> Result<ActivationToken, anyhow::Error> {
|
||||
let mut tx = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.context("Failed to start sql transaction")?;
|
||||
|
||||
let user_id = ent.id();
|
||||
|
||||
let token = ActivationToken::new();
|
||||
|
||||
let query = sqlx::query(
|
||||
r#"
|
||||
INSERT INTO activation_token (user_id, token)
|
||||
VALUES ($1, $2)"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(token.to_string());
|
||||
|
||||
tx.execute(query)
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?;
|
||||
tx.commit()
|
||||
.await
|
||||
.context("failed to commit SQL transaction")?;
|
||||
|
||||
Ok(token)
|
||||
}
|
||||
|
||||
async fn create_password_reset_token(
|
||||
&self,
|
||||
ent: &User,
|
||||
) -> Result<PasswordResetToken, anyhow::Error> {
|
||||
let _ = self.delete_password_reset_tokens_for_user(ent).await;
|
||||
|
||||
let mut tx = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.context("Failed to start sql transaction")?;
|
||||
|
||||
let user_id = ent.id();
|
||||
|
||||
let token = PasswordResetToken::new();
|
||||
|
||||
let query = sqlx::query(
|
||||
r#"
|
||||
INSERT INTO forgot_password (user_id, token)
|
||||
VALUES ($1, $2)"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(token.to_string());
|
||||
|
||||
tx.execute(query)
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?;
|
||||
tx.commit()
|
||||
.await
|
||||
.context("failed to commit SQL transaction")?;
|
||||
|
||||
Ok(token)
|
||||
}
|
||||
|
||||
async fn all_users(&self) -> Vec<User> {
|
||||
todo!()
|
||||
}
|
||||
|
||||
async fn find_user_by_id(&self, id: uuid::Uuid) -> Result<Option<User>, anyhow::Error> {
|
||||
let query = sqlx::query(
|
||||
r#"
|
||||
SELECT
|
||||
id,
|
||||
email,
|
||||
password,
|
||||
verified
|
||||
FROM users WHERE id = $1
|
||||
"#,
|
||||
)
|
||||
.bind(id);
|
||||
|
||||
let row = self
|
||||
.pool
|
||||
.fetch_optional(query)
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?;
|
||||
|
||||
let Some(row) = row else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let id = row.get("id");
|
||||
let email = EmailAddress::new(row.get("email"))?;
|
||||
let password = Password::new_hashed(row.get("password"));
|
||||
let verified = Verified::new(row.get("verified"));
|
||||
|
||||
Ok(Some(User::new(id, email, password, verified)))
|
||||
}
|
||||
|
||||
async fn find_user_by_email(
|
||||
&self,
|
||||
email: &EmailAddress,
|
||||
) -> Result<Option<User>, anyhow::Error> {
|
||||
let query = sqlx::query(
|
||||
r#"
|
||||
SELECT
|
||||
id,
|
||||
email,
|
||||
password,
|
||||
verified
|
||||
FROM users WHERE email = $1
|
||||
"#,
|
||||
)
|
||||
.bind(email.to_string());
|
||||
|
||||
let row = self
|
||||
.pool
|
||||
.fetch_optional(query)
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?;
|
||||
|
||||
let Some(row) = row else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let id = row.get("id");
|
||||
let email = EmailAddress::new(row.get("email"))?;
|
||||
let password = Password::new_hashed(row.get("password"));
|
||||
let verified = Verified::new(row.get("verified"));
|
||||
|
||||
Ok(Some(User::new(id, email, password, verified)))
|
||||
}
|
||||
|
||||
async fn find_user_by_activation_token(
|
||||
&self,
|
||||
token: &ActivationToken,
|
||||
) -> Result<Option<User>, anyhow::Error> {
|
||||
let query = sqlx::query(
|
||||
r#"
|
||||
SELECT
|
||||
user_id,
|
||||
token
|
||||
FROM activation_token WHERE token = $1
|
||||
"#,
|
||||
)
|
||||
.bind(token.to_string());
|
||||
|
||||
let Some(row) = self
|
||||
.pool
|
||||
.fetch_optional(query)
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let id = row.get("user_id");
|
||||
|
||||
Ok(self.find_user_by_id(id).await?)
|
||||
}
|
||||
|
||||
async fn find_user_by_password_reset_token(
|
||||
&self,
|
||||
token: &PasswordResetToken,
|
||||
) -> Result<Option<User>, anyhow::Error> {
|
||||
let query = sqlx::query(
|
||||
r#"
|
||||
SELECT
|
||||
user_id,
|
||||
token
|
||||
FROM forgot_password WHERE token = $1
|
||||
"#,
|
||||
)
|
||||
.bind(token.to_string());
|
||||
|
||||
let Some(row) = self
|
||||
.pool
|
||||
.fetch_optional(query)
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let id = row.get("user_id");
|
||||
|
||||
Ok(self.find_user_by_id(id).await?)
|
||||
}
|
||||
|
||||
async fn update_user(
|
||||
&self,
|
||||
ent: &User,
|
||||
req: UpdateUserRequest,
|
||||
) -> Result<(User, User), UpdateUserError> {
|
||||
let mut tx = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.context("Failed to start sql transaction")?;
|
||||
|
||||
let mut query = QueryBuilder::new("UPDATE users SET ");
|
||||
|
||||
let mut new_email = ent.email();
|
||||
let mut new_password = ent.password();
|
||||
let mut new_verified = ent.verified();
|
||||
|
||||
if let Some(email) = req.email() {
|
||||
new_email = email;
|
||||
query.push(" email = ");
|
||||
query.push_bind(email.to_string());
|
||||
};
|
||||
|
||||
if let Some(password) = req.password() {
|
||||
new_password = password;
|
||||
query.push(" password = ");
|
||||
query.push_bind(password.to_string());
|
||||
};
|
||||
|
||||
if let Some(verified) = req.verified() {
|
||||
new_verified = verified;
|
||||
query.push(" verified = ");
|
||||
query.push_bind::<bool>((*verified).into());
|
||||
};
|
||||
|
||||
query.push(" WHERE id = ");
|
||||
query.push_bind(ent.id());
|
||||
|
||||
tx.execute(query.build())
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?;
|
||||
tx.commit()
|
||||
.await
|
||||
.context("failed to commit SQL transaction")?;
|
||||
|
||||
Ok((
|
||||
ent.clone(),
|
||||
User::new(
|
||||
ent.id().clone(),
|
||||
new_email.clone(),
|
||||
new_password.clone(),
|
||||
new_verified.clone(),
|
||||
),
|
||||
))
|
||||
}
|
||||
|
||||
async fn delete_activation_token_for_user(&self, ent: &User) -> Result<(), anyhow::Error> {
|
||||
let mut tx = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.context("Failed to start sql transaction")?;
|
||||
|
||||
let query = sqlx::query("DELETE FROM activation_token WHERE user_id = $1").bind(ent.id());
|
||||
|
||||
tx.execute(query)
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?;
|
||||
tx.commit()
|
||||
.await
|
||||
.context("failed to commit SQL transaction")?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete_password_reset_tokens_for_user(&self, ent: &User) -> Result<(), anyhow::Error> {
|
||||
let mut tx = self
|
||||
.pool
|
||||
.begin()
|
||||
.await
|
||||
.context("Failed to start sql transaction")?;
|
||||
|
||||
let query = sqlx::query("DELETE FROM forgot_password WHERE user_id = $1").bind(ent.id());
|
||||
|
||||
tx.execute(query)
|
||||
.await
|
||||
.context("failed to execute SQL transaction")?;
|
||||
tx.commit()
|
||||
.await
|
||||
.context("failed to commit SQL transaction")?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user