matrix/authentication-service
1
0
Fork 0
mirror of https://github.com/matrix-org/matrix-authentication-service.git synced 2025-07-29 22:01:14 +03:00
Code Activity

strorage: browser session and user password repositories

Browse Source
This commit is contained in:
Quentin Gliech
2023-01-03 15:21:47 +01:00
parent 4790897892
commit 53172d6a3f
25 changed files with 914 additions and 726 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/storage/src/oauth2/access_token.rs
View File

@ -175,6 +175,7 @@ pub async fn lookup_active_access_token(
let browser_session = BrowserSession {
id: res.user_session_id.into(),
created_at: res.user_session_created_at,
finished_at: None,
user,
last_authentication,
};

1
crates/storage/src/oauth2/authorization_grant.rs
View File

@ -224,6 +224,7 @@ impl GrantLookup {
id: user_session_id.into(),
user,
created_at: user_session_created_at,
finished_at: None,
last_authentication,
};

13
crates/storage/src/oauth2/mod.rs
View File

@ -23,8 +23,8 @@ use uuid::Uuid;
use self::client::lookup_clients;
use crate::{
pagination::{process_page, QueryBuilderExt},
user::lookup_active_session,
Clock, DatabaseError, DatabaseInconsistencyError,
user::BrowserSessionRepository,
Clock, DatabaseError, DatabaseInconsistencyError, Repository,
};
pub mod access_token;
@ -134,11 +134,10 @@ pub async fn get_paginated_user_oauth_sessions(
// ideal
let mut browser_sessions: HashMap<Ulid, BrowserSession> = HashMap::new();
for id in browser_session_ids {
let v = lookup_active_session(&mut *conn, id)
.await?
.ok_or_else(|| {
DatabaseInconsistencyError::on("oauth2_sessions").column("user_session_id")
})?;
let v = conn.browser_session().lookup(id).await?.ok_or_else(|| {
DatabaseInconsistencyError::on("oauth2_sessions").column("user_session_id")
})?;
browser_sessions.insert(id, v);
}

1
crates/storage/src/oauth2/refresh_token.rs
View File

@ -204,6 +204,7 @@ pub async fn lookup_active_refresh_token(
let browser_session = BrowserSession {
id: res.user_session_id.into(),
created_at: res.user_session_created_at,
finished_at: None,
user,
last_authentication,
};

35
crates/storage/src/repository.rs
View File

@ -19,7 +19,10 @@ use crate::{
PgUpstreamOAuthLinkRepository, PgUpstreamOAuthProviderRepository,
PgUpstreamOAuthSessionRepository,
},
user::{PgUserEmailRepository, PgUserRepository},
user::{
PgBrowserSessionRepository, PgUserEmailRepository, PgUserPasswordRepository,
PgUserRepository,
},
};
pub trait Repository {
@ -43,11 +46,21 @@ pub trait Repository {
where
Self: 'c;
type UserPasswordRepository<'c>
where
Self: 'c;
type BrowserSessionRepository<'c>
where
Self: 'c;
fn upstream_oauth_link(&mut self) -> Self::UpstreamOAuthLinkRepository<'_>;
fn upstream_oauth_provider(&mut self) -> Self::UpstreamOAuthProviderRepository<'_>;
fn upstream_oauth_session(&mut self) -> Self::UpstreamOAuthSessionRepository<'_>;
fn user(&mut self) -> Self::UserRepository<'_>;
fn user_email(&mut self) -> Self::UserEmailRepository<'_>;
fn user_password(&mut self) -> Self::UserPasswordRepository<'_>;
fn browser_session(&mut self) -> Self::BrowserSessionRepository<'_>;
}
impl Repository for PgConnection {
@ -56,6 +69,8 @@ impl Repository for PgConnection {
type UpstreamOAuthSessionRepository<'c> = PgUpstreamOAuthSessionRepository<'c> where Self: 'c;
type UserRepository<'c> = PgUserRepository<'c> where Self: 'c;
type UserEmailRepository<'c> = PgUserEmailRepository<'c> where Self: 'c;
type UserPasswordRepository<'c> = PgUserPasswordRepository<'c> where Self: 'c;
type BrowserSessionRepository<'c> = PgBrowserSessionRepository<'c> where Self: 'c;
fn upstream_oauth_link(&mut self) -> Self::UpstreamOAuthLinkRepository<'_> {
PgUpstreamOAuthLinkRepository::new(self)
@ -76,6 +91,14 @@ impl Repository for PgConnection {
fn user_email(&mut self) -> Self::UserEmailRepository<'_> {
PgUserEmailRepository::new(self)
}
fn user_password(&mut self) -> Self::UserPasswordRepository<'_> {
PgUserPasswordRepository::new(self)
}
fn browser_session(&mut self) -> Self::BrowserSessionRepository<'_> {
PgBrowserSessionRepository::new(self)
}
}
impl<'t> Repository for Transaction<'t, Postgres> {
@ -84,6 +107,8 @@ impl<'t> Repository for Transaction<'t, Postgres> {
type UpstreamOAuthSessionRepository<'c> = PgUpstreamOAuthSessionRepository<'c> where Self: 'c;
type UserRepository<'c> = PgUserRepository<'c> where Self: 'c;
type UserEmailRepository<'c> = PgUserEmailRepository<'c> where Self: 'c;
type UserPasswordRepository<'c> = PgUserPasswordRepository<'c> where Self: 'c;
type BrowserSessionRepository<'c> = PgBrowserSessionRepository<'c> where Self: 'c;
fn upstream_oauth_link(&mut self) -> Self::UpstreamOAuthLinkRepository<'_> {
PgUpstreamOAuthLinkRepository::new(self)
@ -104,4 +129,12 @@ impl<'t> Repository for Transaction<'t, Postgres> {
fn user_email(&mut self) -> Self::UserEmailRepository<'_> {
PgUserEmailRepository::new(self)
}
fn user_password(&mut self) -> Self::UserPasswordRepository<'_> {
PgUserPasswordRepository::new(self)
}
fn browser_session(&mut self) -> Self::BrowserSessionRepository<'_> {
PgBrowserSessionRepository::new(self)
}
}

105
crates/storage/src/user/authentication.rs
View File

@ -1,105 +0,0 @@
// Copyright 2022 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use mas_data_model::{Authentication, BrowserSession, Password, UpstreamOAuthLink};
use rand::Rng;
use sqlx::PgExecutor;
use ulid::Ulid;
use uuid::Uuid;
use crate::Clock;
#[tracing::instrument(
skip_all,
fields(
user.id = %user_session.user.id,
%user_password.id,
%user_session.id,
user_session_authentication.id,
),
err,
)]
pub async fn authenticate_session_with_password(
executor: impl PgExecutor<'_>,
mut rng: impl Rng + Send,
clock: &Clock,
user_session: &mut BrowserSession,
user_password: &Password,
) -> Result<(), sqlx::Error> {
let created_at = clock.now();
let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
tracing::Span::current().record(
"user_session_authentication.id",
tracing::field::display(id),
);
sqlx::query!(
r#"
INSERT INTO user_session_authentications
(user_session_authentication_id, user_session_id, created_at)
VALUES ($1, $2, $3)
"#,
Uuid::from(id),
Uuid::from(user_session.id),
created_at,
)
.execute(executor)
.await?;
user_session.last_authentication = Some(Authentication { id, created_at });
Ok(())
}
#[tracing::instrument(
skip_all,
fields(
user.id = %user_session.user.id,
%upstream_oauth_link.id,
%user_session.id,
user_session_authentication.id,
),
err,
)]
pub async fn authenticate_session_with_upstream(
executor: impl PgExecutor<'_>,
mut rng: impl Rng + Send,
clock: &Clock,
user_session: &mut BrowserSession,
upstream_oauth_link: &UpstreamOAuthLink,
) -> Result<(), sqlx::Error> {
let created_at = clock.now();
let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
tracing::Span::current().record(
"user_session_authentication.id",
tracing::field::display(id),
);
sqlx::query!(
r#"
INSERT INTO user_session_authentications
(user_session_authentication_id, user_session_id, created_at)
VALUES ($1, $2, $3)
"#,
Uuid::from(id),
Uuid::from(user_session.id),
created_at,
)
.execute(executor)
.await?;
user_session.last_authentication = Some(Authentication { id, created_at });
Ok(())
}

250
crates/storage/src/user/mod.rs
View File

@ -14,27 +14,22 @@
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use mas_data_model::{Authentication, BrowserSession, User};
use rand::{Rng, RngCore};
use sqlx::{PgConnection, PgExecutor, QueryBuilder};
use tracing::{info_span, Instrument};
use mas_data_model::User;
use rand::RngCore;
use sqlx::PgConnection;
use ulid::Ulid;
use uuid::Uuid;
use crate::{
pagination::{process_page, QueryBuilderExt},
tracing::ExecuteExt,
Clock, DatabaseError, DatabaseInconsistencyError, LookupResultExt,
};
use crate::{tracing::ExecuteExt, Clock, DatabaseError, LookupResultExt};
mod authentication;
mod email;
mod password;
mod session;
pub use self::{
authentication::{authenticate_session_with_password, authenticate_session_with_upstream},
email::{PgUserEmailRepository, UserEmailRepository},
password::{add_user_password, lookup_user_password},
password::{PgUserPasswordRepository, UserPasswordRepository},
session::{BrowserSessionRepository, PgBrowserSessionRepository},
};
#[async_trait]
@ -218,234 +213,3 @@ impl<'c> UserRepository for PgUserRepository<'c> {
Ok(exists)
}
}
#[derive(sqlx::FromRow)]
struct SessionLookup {
user_session_id: Uuid,
user_session_created_at: DateTime<Utc>,
user_id: Uuid,
user_username: String,
user_primary_user_email_id: Option<Uuid>,
last_authentication_id: Option<Uuid>,
last_authd_at: Option<DateTime<Utc>>,
}
impl TryInto<BrowserSession> for SessionLookup {
type Error = DatabaseInconsistencyError;
fn try_into(self) -> Result<BrowserSession, Self::Error> {
let id = Ulid::from(self.user_id);
let user = User {
id,
username: self.user_username,
sub: id.to_string(),
primary_user_email_id: self.user_primary_user_email_id.map(Into::into),
};
let last_authentication = match (self.last_authentication_id, self.last_authd_at) {
(Some(id), Some(created_at)) => Some(Authentication {
id: id.into(),
created_at,
}),
(None, None) => None,
_ => {
return Err(DatabaseInconsistencyError::on(
"user_session_authentications",
))
}
};
Ok(BrowserSession {
id: self.user_session_id.into(),
user,
created_at: self.user_session_created_at,
last_authentication,
})
}
}
#[tracing::instrument(
skip_all,
fields(user_session.id = %id),
err,
)]
pub async fn lookup_active_session(
executor: impl PgExecutor<'_>,
id: Ulid,
) -> Result<Option<BrowserSession>, DatabaseError> {
let res = sqlx::query_as!(
SessionLookup,
r#"
SELECT s.user_session_id
, s.created_at AS "user_session_created_at"
, u.user_id
, u.username AS "user_username"
, u.primary_user_email_id AS "user_primary_user_email_id"
, a.user_session_authentication_id AS "last_authentication_id?"
, a.created_at AS "last_authd_at?"
FROM user_sessions s
INNER JOIN users u
USING (user_id)
LEFT JOIN user_session_authentications a
USING (user_session_id)
WHERE s.user_session_id = $1 AND s.finished_at IS NULL
ORDER BY a.created_at DESC
LIMIT 1
"#,
Uuid::from(id),
)
.fetch_one(executor)
.await
.to_option()?;
let Some(res) = res else { return Ok(None) };
Ok(Some(res.try_into()?))
}
#[tracing::instrument(
skip_all,
fields(
%user.id,
%user.username,
),
err,
)]
pub async fn get_paginated_user_sessions(
executor: impl PgExecutor<'_>,
user: &User,
before: Option<Ulid>,
after: Option<Ulid>,
first: Option<usize>,
last: Option<usize>,
) -> Result<(bool, bool, Vec<BrowserSession>), DatabaseError> {
let mut query = QueryBuilder::new(
r#"
SELECT
s.user_session_id,
u.user_id,
u.username,
s.created_at,
a.user_session_authentication_id AS "last_authentication_id",
a.created_at AS "last_authd_at",
ue.user_email_id AS "user_email_id",
ue.email AS "user_email",
ue.created_at AS "user_email_created_at",
ue.confirmed_at AS "user_email_confirmed_at"
FROM user_sessions s
INNER JOIN users u
USING (user_id)
LEFT JOIN user_session_authentications a
USING (user_session_id)
LEFT JOIN user_emails ue
ON ue.user_email_id = u.primary_user_email_id
"#,
);
query
.push(" WHERE s.finished_at IS NULL AND s.user_id = ")
.push_bind(Uuid::from(user.id))
.generate_pagination("s.user_session_id", before, after, first, last)?;
let span = info_span!("Fetch paginated user emails", db.statement = query.sql());
let page: Vec<SessionLookup> = query
.build_query_as()
.fetch_all(executor)
.instrument(span)
.await?;
let (has_previous_page, has_next_page, page) = process_page(page, first, last)?;
let page: Result<Vec<_>, _> = page.into_iter().map(TryInto::try_into).collect();
Ok((has_previous_page, has_next_page, page?))
}
#[tracing::instrument(
skip_all,
fields(
%user.id,
user_session.id,
),
err,
)]
pub async fn start_session(
executor: impl PgExecutor<'_>,
mut rng: impl Rng + Send,
clock: &Clock,
user: User,
) -> Result<BrowserSession, sqlx::Error> {
let created_at = clock.now();
let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
tracing::Span::current().record("user_session.id", tracing::field::display(id));
sqlx::query!(
r#"
INSERT INTO user_sessions (user_session_id, user_id, created_at)
VALUES ($1, $2, $3)
"#,
Uuid::from(id),
Uuid::from(user.id),
created_at,
)
.execute(executor)
.await?;
let session = BrowserSession {
id,
user,
created_at,
last_authentication: None,
};
Ok(session)
}
#[tracing::instrument(
skip_all,
fields(%user.id),
err,
)]
pub async fn count_active_sessions(
executor: impl PgExecutor<'_>,
user: &User,
) -> Result<i64, DatabaseError> {
let res = sqlx::query_scalar!(
r#"
SELECT COUNT(*) as "count!"
FROM user_sessions s
WHERE s.user_id = $1 AND s.finished_at IS NULL
"#,
Uuid::from(user.id),
)
.fetch_one(executor)
.await?;
Ok(res)
}
#[tracing::instrument(
skip_all,
fields(%user_session.id),
err,
)]
pub async fn end_session(
executor: impl PgExecutor<'_>,
clock: &Clock,
user_session: &BrowserSession,
) -> Result<(), DatabaseError> {
let now = clock.now();
let res = sqlx::query!(
r#"
UPDATE user_sessions
SET finished_at = $1
WHERE user_session_id = $2
"#,
now,
Uuid::from(user_session.id),
)
.execute(executor)
.instrument(info_span!("End session"))
.await?;
DatabaseError::ensure_affected_rows(&res, 1)
}

229
crates/storage/src/user/password.rs
View File

@ -12,63 +12,42 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use mas_data_model::{Password, User};
use rand::Rng;
use sqlx::PgExecutor;
use rand::RngCore;
use sqlx::PgConnection;
use ulid::Ulid;
use uuid::Uuid;
use crate::{Clock, DatabaseError, DatabaseInconsistencyError, LookupResultExt};
use crate::{
tracing::ExecuteExt, Clock, DatabaseError, DatabaseInconsistencyError, LookupResultExt,
};
#[tracing::instrument(
skip_all,
fields(
%user.id,
%user.username,
user_password.id,
user_password.version = version,
),
err,
)]
pub async fn add_user_password(
executor: impl PgExecutor<'_>,
mut rng: impl Rng + Send,
clock: &Clock,
user: &User,
version: u16,
hashed_password: String,
upgraded_from: Option<Password>,
) -> Result<Password, DatabaseError> {
let created_at = clock.now();
let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
tracing::Span::current().record("user_password.id", tracing::field::display(id));
#[async_trait]
pub trait UserPasswordRepository: Send + Sync {
type Error;
let upgraded_from_id = upgraded_from.map(|p| p.id);
async fn active(&mut self, user: &User) -> Result<Option<Password>, Self::Error>;
async fn add(
&mut self,
rng: &mut (dyn RngCore + Send),
clock: &Clock,
user: &User,
version: u16,
hashed_password: String,
upgraded_from: Option<&Password>,
) -> Result<Password, Self::Error>;
}
sqlx::query!(
r#"
INSERT INTO user_passwords
(user_password_id, user_id, hashed_password, version, upgraded_from_id, created_at)
VALUES ($1, $2, $3, $4, $5, $6)
"#,
Uuid::from(id),
Uuid::from(user.id),
hashed_password,
i32::from(version),
upgraded_from_id.map(Uuid::from),
created_at,
)
.execute(executor)
.await?;
pub struct PgUserPasswordRepository<'c> {
conn: &'c mut PgConnection,
}
Ok(Password {
id,
hashed_password,
version,
upgraded_from_id,
created_at,
})
impl<'c> PgUserPasswordRepository<'c> {
pub fn new(conn: &'c mut PgConnection) -> Self {
Self { conn }
}
}
struct UserPasswordLookup {
@ -79,57 +58,115 @@ struct UserPasswordLookup {
created_at: DateTime<Utc>,
}
#[tracing::instrument(
skip_all,
fields(
%user.id,
%user.username,
),
err,
)]
pub async fn lookup_user_password(
executor: impl PgExecutor<'_>,
user: &User,
) -> Result<Option<Password>, DatabaseError> {
let res = sqlx::query_as!(
UserPasswordLookup,
r#"
SELECT up.user_password_id
, up.hashed_password
, up.version
, up.upgraded_from_id
, up.created_at
FROM user_passwords up
WHERE up.user_id = $1
ORDER BY up.created_at DESC
LIMIT 1
"#,
Uuid::from(user.id),
)
.fetch_one(executor)
.await
.to_option()?;
#[async_trait]
impl<'c> UserPasswordRepository for PgUserPasswordRepository<'c> {
type Error = DatabaseError;
let Some(res) = res else { return Ok(None) };
#[tracing::instrument(
name = "db.user_password.active",
skip_all,
fields(
db.statement,
%user.id,
%user.username,
),
err,
)]
async fn active(&mut self, user: &User) -> Result<Option<Password>, Self::Error> {
let res = sqlx::query_as!(
UserPasswordLookup,
r#"
SELECT up.user_password_id
, up.hashed_password
, up.version
, up.upgraded_from_id
, up.created_at
FROM user_passwords up
WHERE up.user_id = $1
ORDER BY up.created_at DESC
LIMIT 1
"#,
Uuid::from(user.id),
)
.traced()
.fetch_one(&mut *self.conn)
.await
.to_option()?;
let id = Ulid::from(res.user_password_id);
let Some(res) = res else { return Ok(None) };
let version = res.version.try_into().map_err(|e| {
DatabaseInconsistencyError::on("user_passwords")
.column("version")
.row(id)
.source(e)
})?;
let id = Ulid::from(res.user_password_id);
let upgraded_from_id = res.upgraded_from_id.map(Ulid::from);
let created_at = res.created_at;
let hashed_password = res.hashed_password;
let version = res.version.try_into().map_err(|e| {
DatabaseInconsistencyError::on("user_passwords")
.column("version")
.row(id)
.source(e)
})?;
Ok(Some(Password {
id,
hashed_password,
version,
upgraded_from_id,
created_at,
}))
let upgraded_from_id = res.upgraded_from_id.map(Ulid::from);
let created_at = res.created_at;
let hashed_password = res.hashed_password;
Ok(Some(Password {
id,
hashed_password,
version,
upgraded_from_id,
created_at,
}))
}
#[tracing::instrument(
name = "db.user_password.add",
skip_all,
fields(
db.statement,
%user.id,
%user.username,
user_password.id,
user_password.version = version,
),
err,
)]
async fn add(
&mut self,
rng: &mut (dyn RngCore + Send),
clock: &Clock,
user: &User,
version: u16,
hashed_password: String,
upgraded_from: Option<&Password>,
) -> Result<Password, Self::Error> {
let created_at = clock.now();
let id = Ulid::from_datetime_with_source(created_at.into(), rng);
tracing::Span::current().record("user_password.id", tracing::field::display(id));
let upgraded_from_id = upgraded_from.map(|p| p.id);
sqlx::query!(
r#"
INSERT INTO user_passwords
(user_password_id, user_id, hashed_password, version, upgraded_from_id, created_at)
VALUES ($1, $2, $3, $4, $5, $6)
"#,
Uuid::from(id),
Uuid::from(user.id),
hashed_password,
i32::from(version),
upgraded_from_id.map(Uuid::from),
created_at,
)
.traced()
.execute(&mut *self.conn)
.await?;
Ok(Password {
id,
hashed_password,
version,
upgraded_from_id,
created_at,
})
}
}

425
crates/storage/src/user/session.rs Normal file
View File

@ -0,0 +1,425 @@
// Copyright 2022 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use mas_data_model::{Authentication, BrowserSession, Password, UpstreamOAuthLink, User};
use rand::RngCore;
use sqlx::{PgConnection, QueryBuilder};
use ulid::Ulid;
use uuid::Uuid;
use crate::{
pagination::{process_page, Page, QueryBuilderExt},
tracing::ExecuteExt,
Clock, DatabaseError, DatabaseInconsistencyError, LookupResultExt,
};
#[async_trait]
pub trait BrowserSessionRepository: Send + Sync {
type Error;
async fn lookup(&mut self, id: Ulid) -> Result<Option<BrowserSession>, Self::Error>;
async fn add(
&mut self,
rng: &mut (dyn RngCore + Send),
clock: &Clock,
user: &User,
) -> Result<BrowserSession, Self::Error>;
async fn finish(
&mut self,
clock: &Clock,
user_session: BrowserSession,
) -> Result<BrowserSession, Self::Error>;
async fn list_active_paginated(
&mut self,
user: &User,
before: Option<Ulid>,
after: Option<Ulid>,
first: Option<usize>,
last: Option<usize>,
) -> Result<Page<BrowserSession>, Self::Error>;
async fn count_active(&mut self, user: &User) -> Result<usize, Self::Error>;
async fn authenticate_with_password(
&mut self,
rng: &mut (dyn RngCore + Send),
clock: &Clock,
user_session: BrowserSession,
user_password: &Password,
) -> Result<BrowserSession, Self::Error>;
async fn authenticate_with_upstream(
&mut self,
rng: &mut (dyn RngCore + Send),
clock: &Clock,
user_session: BrowserSession,
upstream_oauth_link: &UpstreamOAuthLink,
) -> Result<BrowserSession, Self::Error>;
}
pub struct PgBrowserSessionRepository<'c> {
conn: &'c mut PgConnection,
}
impl<'c> PgBrowserSessionRepository<'c> {
pub fn new(conn: &'c mut PgConnection) -> Self {
Self { conn }
}
}
#[derive(sqlx::FromRow)]
struct SessionLookup {
user_session_id: Uuid,
user_session_created_at: DateTime<Utc>,
user_session_finished_at: Option<DateTime<Utc>>,
user_id: Uuid,
user_username: String,
user_primary_user_email_id: Option<Uuid>,
last_authentication_id: Option<Uuid>,
last_authd_at: Option<DateTime<Utc>>,
}
impl TryInto<BrowserSession> for SessionLookup {
type Error = DatabaseInconsistencyError;
fn try_into(self) -> Result<BrowserSession, Self::Error> {
let id = Ulid::from(self.user_id);
let user = User {
id,
username: self.user_username,
sub: id.to_string(),
primary_user_email_id: self.user_primary_user_email_id.map(Into::into),
};
let last_authentication = match (self.last_authentication_id, self.last_authd_at) {
(Some(id), Some(created_at)) => Some(Authentication {
id: id.into(),
created_at,
}),
(None, None) => None,
_ => {
return Err(DatabaseInconsistencyError::on(
"user_session_authentications",
))
}
};
Ok(BrowserSession {
id: self.user_session_id.into(),
user,
created_at: self.user_session_created_at,
finished_at: self.user_session_finished_at,
last_authentication,
})
}
}
#[async_trait]
impl<'c> BrowserSessionRepository for PgBrowserSessionRepository<'c> {
type Error = DatabaseError;
#[tracing::instrument(
name = "db.browser_session.lookup",
skip_all,
fields(
db.statement,
user_session.id = %id,
),
err,
)]
async fn lookup(&mut self, id: Ulid) -> Result<Option<BrowserSession>, Self::Error> {
let res = sqlx::query_as!(
SessionLookup,
r#"
SELECT s.user_session_id
, s.created_at AS "user_session_created_at"
, s.finished_at AS "user_session_finished_at"
, u.user_id
, u.username AS "user_username"
, u.primary_user_email_id AS "user_primary_user_email_id"
, a.user_session_authentication_id AS "last_authentication_id?"
, a.created_at AS "last_authd_at?"
FROM user_sessions s
INNER JOIN users u
USING (user_id)
LEFT JOIN user_session_authentications a
USING (user_session_id)
WHERE s.user_session_id = $1
ORDER BY a.created_at DESC
LIMIT 1
"#,
Uuid::from(id),
)
.traced()
.fetch_one(&mut *self.conn)
.await
.to_option()?;
let Some(res) = res else { return Ok(None) };
Ok(Some(res.try_into()?))
}
#[tracing::instrument(
name = "db.browser_session.add",
skip_all,
fields(
db.statement,
%user.id,
user_session.id,
),
err,
)]
async fn add(
&mut self,
rng: &mut (dyn RngCore + Send),
clock: &Clock,
user: &User,
) -> Result<BrowserSession, Self::Error> {
let created_at = clock.now();
let id = Ulid::from_datetime_with_source(created_at.into(), rng);
tracing::Span::current().record("user_session.id", tracing::field::display(id));
sqlx::query!(
r#"
INSERT INTO user_sessions (user_session_id, user_id, created_at)
VALUES ($1, $2, $3)
"#,
Uuid::from(id),
Uuid::from(user.id),
created_at,
)
.traced()
.execute(&mut *self.conn)
.await?;
let session = BrowserSession {
id,
// XXX
user: user.clone(),
created_at,
finished_at: None,
last_authentication: None,
};
Ok(session)
}
#[tracing::instrument(
name = "db.browser_session.finish",
skip_all,
fields(
db.statement,
%user_session.id,
),
err,
)]
async fn finish(
&mut self,
clock: &Clock,
mut user_session: BrowserSession,
) -> Result<BrowserSession, Self::Error> {
let finished_at = clock.now();
let res = sqlx::query!(
r#"
UPDATE user_sessions
SET finished_at = $1
WHERE user_session_id = $2
"#,
finished_at,
Uuid::from(user_session.id),
)
.traced()
.execute(&mut *self.conn)
.await?;
user_session.finished_at = Some(finished_at);
DatabaseError::ensure_affected_rows(&res, 1)?;
Ok(user_session)
}
#[tracing::instrument(
name = "db.browser_session.list_active_paginated",
skip_all,
fields(
db.statement,
%user.id,
),
err,
)]
async fn list_active_paginated(
&mut self,
user: &User,
before: Option<Ulid>,
after: Option<Ulid>,
first: Option<usize>,
last: Option<usize>,
) -> Result<Page<BrowserSession>, Self::Error> {
// TODO: ordering of last authentication is wrong
let mut query = QueryBuilder::new(
r#"
SELECT DISTINCT ON (s.user_session_id)
s.user_session_id,
u.user_id,
u.username,
s.created_at,
a.user_session_authentication_id AS "last_authentication_id",
a.created_at AS "last_authd_at",
FROM user_sessions s
INNER JOIN users u
USING (user_id)
LEFT JOIN user_session_authentications a
USING (user_session_id)
"#,
);
query
.push(" WHERE s.finished_at IS NULL AND s.user_id = ")
.push_bind(Uuid::from(user.id))
.generate_pagination("s.user_session_id", before, after, first, last)?;
let page: Vec<SessionLookup> = query
.build_query_as()
.traced()
.fetch_all(&mut *self.conn)
.await?;
let (has_previous_page, has_next_page, edges) = process_page(page, first, last)?;
let edges: Result<Vec<_>, _> = edges.into_iter().map(TryInto::try_into).collect();
Ok(Page {
has_previous_page,
has_next_page,
edges: edges?,
})
}
#[tracing::instrument(
name = "db.browser_session.count_active",
skip_all,
fields(
db.statement,
%user.id,
),
err,
)]
async fn count_active(&mut self, user: &User) -> Result<usize, Self::Error> {
let res = sqlx::query_scalar!(
r#"
SELECT COUNT(*) as "count!"
FROM user_sessions s
WHERE s.user_id = $1 AND s.finished_at IS NULL
"#,
Uuid::from(user.id),
)
.traced()
.fetch_one(&mut *self.conn)
.await?;
res.try_into().map_err(DatabaseError::to_invalid_operation)
}
#[tracing::instrument(
name = "db.browser_session.authenticate_with_password",
skip_all,
fields(
db.statement,
%user_session.id,
%user_password.id,
user_session_authentication.id,
),
err,
)]
async fn authenticate_with_password(
&mut self,
rng: &mut (dyn RngCore + Send),
clock: &Clock,
mut user_session: BrowserSession,
user_password: &Password,
) -> Result<BrowserSession, Self::Error> {
let _user_password = user_password;
let created_at = clock.now();
let id = Ulid::from_datetime_with_source(created_at.into(), rng);
tracing::Span::current().record(
"user_session_authentication.id",
tracing::field::display(id),
);
sqlx::query!(
r#"
INSERT INTO user_session_authentications
(user_session_authentication_id, user_session_id, created_at)
VALUES ($1, $2, $3)
"#,
Uuid::from(id),
Uuid::from(user_session.id),
created_at,
)
.traced()
.execute(&mut *self.conn)
.await?;
user_session.last_authentication = Some(Authentication { id, created_at });
Ok(user_session)
}
#[tracing::instrument(
name = "db.browser_session.authenticate_with_upstream",
skip_all,
fields(
db.statement,
%user_session.id,
%upstream_oauth_link.id,
user_session_authentication.id,
),
err,
)]
async fn authenticate_with_upstream(
&mut self,
rng: &mut (dyn RngCore + Send),
clock: &Clock,
mut user_session: BrowserSession,
upstream_oauth_link: &UpstreamOAuthLink,
) -> Result<BrowserSession, Self::Error> {
let _upstream_oauth_link = upstream_oauth_link;
let created_at = clock.now();
let id = Ulid::from_datetime_with_source(created_at.into(), rng);
tracing::Span::current().record(
"user_session_authentication.id",
tracing::field::display(id),
);
sqlx::query!(
r#"
INSERT INTO user_session_authentications
(user_session_authentication_id, user_session_id, created_at)
VALUES ($1, $2, $3)
"#,
Uuid::from(id),
Uuid::from(user_session.id),
created_at,
)
.traced()
.execute(&mut *self.conn)
.await?;
user_session.last_authentication = Some(Authentication { id, created_at });
Ok(user_session)
}
}