path: root/src/db/mod.rs

use std::{error::Error, mem::replace, sync::Arc};

use anyhow::Result;
use chrono::{DateTime, Duration, Utc};
use password_hash::SaltString;
use rocket::{
    fairing::{self, Fairing},
    futures::lock::{MappedMutexGuard, Mutex, MutexGuard},
    http::Status,
    request::{FromRequest, Outcome},
    Request, Response, Sentinel,
};
use serde_json::Value;
use sqlx::{query, query_as, query_scalar, PgPool, Postgres, Transaction};

use crate::{
    crypto::WrappedKeyBundle,
    types::{
        oauth::ScopeSet, AccountResetID, AttachedClient, Avatar, AvatarID, Device, DeviceCommand,
        DeviceCommands, DeviceID, DevicePush, DeviceUpdate, HawkKey, KeyFetchID, OauthAccessToken,
        OauthAccessType, OauthAuthorization, OauthAuthorizationID, OauthRefreshToken, OauthTokenID,
        PasswordChangeID, SecretKey, SessionID, User, UserID, UserSession, VerifyCode, VerifyHash,
    },
};

// we implement a completely custom db type set instead of using rocket_db_pools for two reasons:
//  1. rocket_db_pools doesn't support sqlx 0.6
//  2. we want one transaction per *request*, including all guards

#[derive(Clone)]
pub struct Db {
    db: Arc<PgPool>,
}

impl Db {
    pub async fn connect(db: &str) -> Result<Self> {
        Ok(Self { db: Arc::new(PgPool::connect(db).await?) })
    }

    pub async fn begin(&self) -> Result<DbConn> {
        Ok(DbConn(Mutex::new(ConnState::Capable(self.clone()))))
    }

    pub async fn migrate(&self) -> Result<()> {
        sqlx::migrate!().run(&*self.db).await?;
        Ok(())
    }
}

struct ActiveConn {
    tx: Transaction<'static, Postgres>,
    always_commit: bool,
}

#[allow(clippy::large_enum_variant)]
enum ConnState {
    None,
    Capable(Db),
    Active(ActiveConn),
    Done,
}

pub struct DbConn(Mutex<ConnState>);

struct DbWrap(Db);
struct DbConnWrap(DbConn);
type DbLock<'c> = MappedMutexGuard<'c, ConnState, ActiveConn>;

#[rocket::async_trait]
impl Fairing for Db {
    fn info(&self) -> fairing::Info {
        fairing::Info {
            name: "db access",
            kind: fairing::Kind::Ignite | fairing::Kind::Response | fairing::Kind::Singleton,
        }
    }

    async fn on_ignite(&self, rocket: rocket::Rocket<rocket::Build>) -> fairing::Result {
        Ok(rocket.manage(DbWrap(self.clone())).manage(self.clone()))
    }

    async fn on_response<'r>(&self, req: &'r Request<'_>, resp: &mut Response<'r>) {
        let conn = req.local_cache(|| DbConnWrap(DbConn(Mutex::new(ConnState::None))));
        let s = replace(&mut *conn.0 .0.lock().await, ConnState::Done);
        if let ConnState::Active(ActiveConn { tx, always_commit }) = s {
            // don't commit if the request failed, unless explicitly asked to.
            // this is used by key fetch to invalidate tokens even if the header
            // signature of the request is incorrect.
            if !always_commit && resp.status() != Status::Ok {
                return;
            }
            if let Err(e) = tx.commit().await {
                resp.set_status(Status::InternalServerError);
                error!("commit failed: {}", e);
            }
        }
    }
}

impl DbConn {
    pub async fn commit(self) -> Result<(), sqlx::Error> {
        match self.0.into_inner() {
            ConnState::None => Ok(()),
            ConnState::Capable(_) => Ok(()),
            ConnState::Active(ac) => ac.tx.commit().await,
            ConnState::Done => Ok(()),
        }
    }

    fn register<'r>(req: &'r Request<'_>) -> &'r DbConn {
        &req.local_cache(|| {
            let db = req.rocket().state::<DbWrap>().unwrap().0.clone();
            DbConnWrap(DbConn(Mutex::new(ConnState::Capable(db))))
        })
        .0
    }

    // defer opening a transaction and thus locking the mutex holding it.
    // without deferral the order of arguments in route signatures is important,
    // which may be surprising: placing a DbConn before a guard that also uses
    // the database causes a concurrent transaction error.
    // HACK maybe we should return errors instead of panicking, but there's no
    // way an error here is not a severe bug
    async fn get(&self) -> sqlx::Result<DbLock<'_>> {
        let mut m = match self.0.try_lock() {
            Some(m) => m,
            None => panic!("attempted to open concurrent transactions"),
        };
        match &*m {
            ConnState::Capable(db) => {
                *m = ConnState::Active(ActiveConn {
                    tx: db.db.begin().await?,
                    always_commit: false,
                });
            },
            ConnState::None | ConnState::Done => panic!("db connection requested after teardown"),
            _ => (),
        }
        Ok(MutexGuard::map(m, |g| match g {
            ConnState::Active(ref mut tx) => tx,
            _ => unreachable!(),
        }))
    }
}

impl<'r> Sentinel for &'r Db {
    fn abort(rocket: &rocket::Rocket<rocket::Ignite>) -> bool {
        rocket.state::<DbWrap>().is_none()
    }
}

#[rocket::async_trait]
impl<'r> FromRequest<'r> for &'r Db {
    type Error = anyhow::Error;

    async fn from_request(req: &'r Request<'_>) -> Outcome<Self, Self::Error> {
        Outcome::Success(&req.rocket().state::<DbWrap>().unwrap().0)
    }
}

impl<'r> Sentinel for &'r DbConn {
    fn abort(rocket: &rocket::Rocket<rocket::Ignite>) -> bool {
        rocket.state::<DbWrap>().is_none()
    }
}

#[rocket::async_trait]
impl<'r> FromRequest<'r> for &'r DbConn {
    type Error = anyhow::Error;

    async fn from_request(req: &'r Request<'_>) -> Outcome<Self, Self::Error> {
        Outcome::Success(DbConn::register(req))
    }
}

//
//
//

//
//
//

impl DbConn {
    pub(crate) async fn always_commit(&self) -> Result<()> {
        self.get().await?.always_commit = true;
        Ok(())
    }

    //
    //
    //

    pub(crate) async fn add_session(
        &self,
        id: SessionID,
        user_id: &UserID,
        key: HawkKey,
        verified: bool,
        verify_code: Option<&str>,
    ) -> sqlx::Result<DateTime<Utc>> {
        query_scalar!(
            r#"insert into user_session (session_id, user_id, req_hmac_key, device_id, verified,
                                         verify_code)
               values ($1, $2, $3, null, $4, $5)
               returning created_at"#,
            id as _,
            user_id as _,
            key as _,
            verified,
            verify_code,
        )
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn use_session(&self, id: &SessionID) -> sqlx::Result<UserSession> {
        query_as!(
            UserSession,
            r#"update user_session
               set last_active = now()
               where session_id = $1
               returning user_id as "uid: UserID", req_hmac_key as "req_hmac_key: HawkKey",
                         device_id as "device_id: DeviceID", created_at, verified, verify_code"#,
            id as _
        )
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn use_session_from_refresh(
        &self,
        id: &OauthTokenID,
    ) -> sqlx::Result<(SessionID, UserSession)> {
        query!(
            r#"update user_session
               set last_active = now()
               where session_id = (
                   select session_id from oauth_token where kind = 'refresh' and id = $1
               )
               returning user_id as "uid: UserID", req_hmac_key as "req_hmac_key: HawkKey",
                         device_id as "device_id: DeviceID", session_id as "session_id: SessionID",
                         created_at, verified, verify_code"#,
            id as _
        )
        .map(|r| {
            (
                r.session_id,
                UserSession {
                    uid: r.uid,
                    req_hmac_key: r.req_hmac_key,
                    device_id: r.device_id,
                    created_at: r.created_at,
                    verified: r.verified,
                    verify_code: r.verify_code,
                },
            )
        })
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn delete_session(&self, user: &UserID, id: &SessionID) -> sqlx::Result<()> {
        query_scalar!(
            r#"delete from user_session
               where user_id = $1 and session_id = $2
               returning 1"#,
            user as _,
            id as _
        )
        .fetch_one(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn set_session_device<'d>(
        &self,
        id: &SessionID,
        dev: Option<&'d DeviceID>,
    ) -> sqlx::Result<()> {
        query!(
            r#"update user_session set device_id = $1 where session_id = $2"#,
            dev as _,
            id as _
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn set_session_verified(&self, id: &SessionID) -> sqlx::Result<()> {
        query_scalar!(
            r#"update user_session
               set verified = true, verify_code = null
               where session_id = $1
               returning 1"#,
            id as _
        )
        .fetch_one(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    //
    //
    //

    pub(crate) async fn enqueue_command(
        &self,
        target: &DeviceID,
        sender: &Option<DeviceID>,
        command: &str,
        payload: &Value,
        ttl: u32,
    ) -> sqlx::Result<i64> {
        let expires = Utc::now() + Duration::seconds(ttl as i64);
        query!(
            r#"insert into device_commands (device_id, command, payload, expires, sender)
               values ($1, $2, $3, $4, $5)
               returning index"#,
            target as _,
            command,
            payload,
            expires,
            sender.as_ref().map(ToString::to_string)
        )
        .map(|x| x.index)
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn get_commands(
        &self,
        user: &UserID,
        dev: &DeviceID,
        min_index: i64,
        limit: i64,
    ) -> sqlx::Result<(bool, Vec<DeviceCommand>)> {
        // NOTE while fxa api docs state that command queries return only commands enqueued
        // *after* index, what pushbox actually does is return commands *starting at* index!
        let mut results = query_as!(
            DeviceCommand,
            r#"select index, command, payload, expires, sender
               from device_commands join device using (device_id)
               where index >= $1 and device_id = $2 and user_id = $3
               order by index
               limit $4"#,
            min_index,
            dev as _,
            user as _,
            limit + 1
        )
        .fetch_all(&mut self.get().await?.tx)
        .await?;
        let more = results.len() > limit as usize;
        results.truncate(limit as usize);
        Ok((more, results))
    }

    //
    //
    //

    pub(crate) async fn get_devices(&self, user: &UserID) -> sqlx::Result<Vec<Device>> {
        query_as!(
            Device,
            r#"select d.device_id as "device_id: DeviceID", d.name, d.type as type_,
                      d.push as "push: DevicePush",
                      d.available_commands as "available_commands: DeviceCommands",
                      d.push_expired, d.location, coalesce(us.last_active, to_timestamp(0)) as "last_active!"
               from device d left join user_session us using (device_id)
               where d.user_id = $1"#,
            user as _
        )
        .fetch_all(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn get_device(&self, user: &UserID, dev: &DeviceID) -> sqlx::Result<Device> {
        query_as!(
            Device,
            r#"select d.device_id as "device_id: DeviceID", d.name, d.type as type_,
                      d.push as "push: DevicePush",
                      d.available_commands as "available_commands: DeviceCommands",
                      d.push_expired, d.location, coalesce(us.last_active, to_timestamp(0)) as "last_active!"
               from device d left join user_session us using (device_id)
               where d.user_id = $1 and d.device_id = $2"#,
            user as _,
            dev as _
        )
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn change_device<'d>(
        &self,
        user: &UserID,
        id: &DeviceID,
        dev: DeviceUpdate<'d>,
    ) -> sqlx::Result<Device> {
        query_as!(
            Device,
            r#"select device_id as "device_id!: DeviceID", name as "name!", type as "type_!",
                      push as "push: DevicePush",
                      available_commands as "available_commands!: DeviceCommands",
                      push_expired as "push_expired!", location as "location!",
                      coalesce(last_active, to_timestamp(0)) as "last_active!"
               from insert_or_update_device($1, $2, $3, $4, $5, $6, $7) as iud
                    left join user_session using (device_id)"#,
            id as _,
            user as _,
            // these two are not optional but are Option anyway. the db will
            // refuse insertions that don't have them set.
            dev.name,
            dev.type_,
            dev.push as _,
            dev.available_commands as _,
            dev.location
        )
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn set_push_expired(&self, dev: &DeviceID) -> sqlx::Result<()> {
        query!(
            r#"update device
               set push_expired = true
               where device_id = $1"#,
            dev as _
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn delete_device(&self, user: &UserID, dev: &DeviceID) -> sqlx::Result<()> {
        query_scalar!(
            r#"delete from device where user_id = $1 and device_id = $2 returning 1"#,
            user as _,
            dev as _
        )
        .fetch_one(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    //
    //
    //

    pub(crate) async fn add_key_fetch(
        &self,
        id: KeyFetchID,
        hmac_key: &HawkKey,
        keys: &WrappedKeyBundle,
    ) -> sqlx::Result<()> {
        query!(
            r#"insert into key_fetch (id, hmac_key, keys) values ($1, $2, $3)"#,
            id as _,
            hmac_key as _,
            &keys.to_bytes()[..]
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn finish_key_fetch(
        &self,
        id: &KeyFetchID,
    ) -> sqlx::Result<(HawkKey, Vec<u8>)> {
        query!(
            r#"delete from key_fetch
               where id = $1 and expires_at > now()
               returning hmac_key as "hmac_key: HawkKey", keys"#,
            id as _
        )
        .map(|r| (r.hmac_key, r.keys))
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    //
    //
    //

    pub(crate) async fn get_refresh_token(
        &self,
        id: &OauthTokenID,
    ) -> sqlx::Result<OauthRefreshToken> {
        query_as!(
            OauthRefreshToken,
            r#"select user_id as "user_id: UserID", client_id, scope as "scope: ScopeSet",
                      session_id as "session_id: SessionID"
               from oauth_token
               where id = $1 and kind = 'refresh'"#,
            id as _
        )
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn get_access_token(
        &self,
        id: &OauthTokenID,
    ) -> sqlx::Result<OauthAccessToken> {
        query_as!(
            OauthAccessToken,
            r#"select user_id as "user_id: UserID", client_id, scope as "scope: ScopeSet",
                      parent_refresh as "parent_refresh: OauthTokenID",
                      parent_session as "parent_session: SessionID",
                      expires_at as "expires_at!"
               from oauth_token
               where id = $1 and kind = 'access' and expires_at > now()"#,
            id as _
        )
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn add_refresh_token(
        &self,
        id: &OauthTokenID,
        token: OauthRefreshToken,
    ) -> sqlx::Result<()> {
        query!(
            r#"insert into oauth_token (id, kind, user_id, client_id, scope, session_id)
               values ($1, 'refresh', $2, $3, $4, $5)"#,
            id as _,
            token.user_id as _,
            token.client_id,
            token.scope as _,
            token.session_id as _
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn add_access_token(
        &self,
        id: &OauthTokenID,
        token: OauthAccessToken,
    ) -> sqlx::Result<()> {
        query!(
            r#"insert into oauth_token (id, kind, user_id, client_id, scope, session_id,
                                        parent_refresh, parent_session, expires_at)
               values ($1, 'access', $2, $3, $4, null, $5, $6, $7)"#,
            id as _,
            token.user_id as _,
            token.client_id,
            token.scope as _,
            token.parent_refresh as _,
            token.parent_session as _,
            token.expires_at,
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn delete_oauth_token(&self, id: &OauthTokenID) -> sqlx::Result<bool> {
        Ok(query!(r#"delete from oauth_token where id = $1"#, id as _)
            .execute(&mut self.get().await?.tx)
            .await?
            .rows_affected()
            > 0)
    }

    pub(crate) async fn delete_refresh_token(&self, id: &OauthTokenID) -> sqlx::Result<()> {
        query!(r#"delete from oauth_token where id = $1 and kind = 'refresh'"#, id as _)
            .execute(&mut self.get().await?.tx)
            .await?;
        Ok(())
    }

    //
    //
    //

    pub(crate) async fn add_oauth_authorization(
        &self,
        id: &OauthAuthorizationID,
        auth: OauthAuthorization,
    ) -> sqlx::Result<()> {
        query!(
            r#"insert into oauth_authorization (id, user_id, client_id, scope, access_type,
                                                code_challenge, keys_jwe, auth_at)
               values ($1, $2, $3, $4, $5, $6, $7, $8)"#,
            id as _,
            auth.user_id as _,
            auth.client_id,
            auth.scope as _,
            auth.access_type as _,
            auth.code_challenge,
            auth.keys_jwe,
            auth.auth_at,
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn take_oauth_authorization(
        &self,
        id: &OauthAuthorizationID,
    ) -> sqlx::Result<OauthAuthorization> {
        query_as!(
            OauthAuthorization,
            r#"delete from oauth_authorization
               where id = $1 and expires_at > now()
               returning user_id as "user_id: UserID", client_id, scope as "scope: ScopeSet",
                         access_type as "access_type: OauthAccessType",
                         code_challenge, keys_jwe, auth_at"#,
            id as _
        )
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    //
    //
    //

    pub(crate) async fn user_email_exists(&self, email: &str) -> sqlx::Result<bool> {
        Ok(query_scalar!(r#"select 1 from users where email = lower($1)"#, email)
            .fetch_optional(&mut self.get().await?.tx)
            .await?
            .is_some())
    }

    pub(crate) async fn add_user(&self, user: User) -> sqlx::Result<UserID> {
        let id = UserID::random();
        query_scalar!(
            r#"insert into users (user_id, auth_salt, email, ka, wrapwrap_kb, verify_hash,
                                  display_name)
               values ($1, $2, $3, $4, $5, $6, $7)"#,
            id as _,
            user.auth_salt.as_str(),
            user.email,
            user.ka as _,
            user.wrapwrap_kb as _,
            user.verify_hash as _,
            user.display_name,
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(id)
    }

    pub(crate) async fn get_user(&self, email: &str) -> sqlx::Result<(UserID, User)> {
        query!(
            r#"select user_id as "id: UserID", auth_salt as "auth_salt: String", email,
                      ka as "ka: SecretKey", wrapwrap_kb as "wrapwrap_kb: SecretKey",
                      verify_hash as "verify_hash: VerifyHash", display_name, verified
               from users
               where email = lower($1)"#,
            email
        )
        .try_map(|r| {
            Ok((
                r.id,
                User {
                    auth_salt: SaltString::new(&r.auth_salt).map_err(decode_err("auth_salt"))?,
                    email: r.email,
                    ka: r.ka,
                    wrapwrap_kb: r.wrapwrap_kb,
                    verify_hash: r.verify_hash,
                    display_name: r.display_name,
                    verified: r.verified,
                },
            ))
        })
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn get_user_by_id(&self, id: &UserID) -> sqlx::Result<User> {
        query!(
            r#"select auth_salt as "auth_salt: String", email,
                      ka as "ka: SecretKey", wrapwrap_kb as "wrapwrap_kb: SecretKey",
                      verify_hash as "verify_hash: VerifyHash", display_name, verified
               from users
               where user_id = $1"#,
            id as _
        )
        .try_map(|r| {
            Ok(User {
                auth_salt: SaltString::new(&r.auth_salt).map_err(decode_err("auth_salt"))?,
                email: r.email,
                ka: r.ka,
                wrapwrap_kb: r.wrapwrap_kb,
                verify_hash: r.verify_hash,
                display_name: r.display_name,
                verified: r.verified,
            })
        })
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn set_user_name(&self, id: &UserID, name: &str) -> sqlx::Result<()> {
        query!(
            "update users
             set display_name = $2
             where user_id = $1",
            id as _,
            name,
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn set_user_verified(&self, id: &UserID) -> sqlx::Result<()> {
        query_scalar!("update users set verified = true where user_id = $1 returning 1", id as _)
            .fetch_one(&mut self.get().await?.tx)
            .await?;
        Ok(())
    }

    pub(crate) async fn delete_user(&self, email: &str) -> sqlx::Result<()> {
        query_scalar!(r#"delete from users where email = lower($1)"#, email)
            .execute(&mut self.get().await?.tx)
            .await?;
        Ok(())
    }

    pub(crate) async fn change_user_auth(
        &self,
        uid: &UserID,
        salt: SaltString,
        wwkb: SecretKey,
        verify_hash: VerifyHash,
    ) -> sqlx::Result<()> {
        query!(
            r#"update users
               set auth_salt = $2, wrapwrap_kb = $3, verify_hash = $4
               where user_id = $1"#,
            uid as _,
            salt.to_string(),
            wwkb as _,
            verify_hash as _,
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn reset_user_auth(
        &self,
        uid: &UserID,
        salt: SaltString,
        wwkb: SecretKey,
        verify_hash: VerifyHash,
    ) -> sqlx::Result<()> {
        query!(
            r#"call reset_user_auth($1, $2, $3, $4)"#,
            uid as _,
            salt.to_string(),
            wwkb as _,
            verify_hash as _,
        )
        .execute(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    //
    //
    //

    pub(crate) async fn get_attached_clients(
        &self,
        id: &UserID,
    ) -> sqlx::Result<Vec<AttachedClient>> {
        query_as!(
            AttachedClient,
            r#"select
                   ot.client_id as "client_id?",
                   d.device_id as "device_id?: DeviceID",
                   us.session_id as "session_token_id?: SessionID",
                   ot.id as "refresh_token_id?: OauthTokenID",
                   d.type as "device_type?",
                   d.name as "name?",
                   coalesce(d.created_at, us.created_at, ot.created_at) as "created_time?",
                   us.last_active as "last_access_time?",
                   ot.scope as "scope?"
               from device d
                    full outer join user_session us on (d.device_id = us.device_id)
                    full outer join oauth_token ot on (us.session_id = ot.session_id)
               where
                   (ot.kind is null or ot.kind = 'refresh')
                   and $1 in (d.user_id, us.user_id, ot.user_id)
               order by d.device_id"#,
            id as _,
        )
        .fetch_all(&mut self.get().await?.tx)
        .await
    }

    //
    //
    //

    pub(crate) async fn get_user_avatar_id(&self, id: &UserID) -> sqlx::Result<Option<AvatarID>> {
        query!(r#"select id as "id: AvatarID" from user_avatars where user_id = $1"#, id as _,)
            .map(|r| r.id)
            .fetch_optional(&mut *self.get().await?.tx)
            .await
    }

    pub(crate) async fn get_user_avatar(&self, id: &AvatarID) -> sqlx::Result<Option<Avatar>> {
        query_as!(
            Avatar,
            r#"select id as "id: AvatarID", data, content_type
               from user_avatars
               where id = $1"#,
            id as _,
        )
        .fetch_optional(&mut *self.get().await?.tx)
        .await
    }

    pub(crate) async fn set_user_avatar(&self, id: &UserID, avatar: Avatar) -> sqlx::Result<()> {
        query!(
            r#"insert into user_avatars (user_id, id, data, content_type)
               values ($1, $2, $3, $4)
               on conflict (user_id) do update set
                 id = $2, data = $3, content_type = $4"#,
            id as _,
            avatar.id as _,
            avatar.data,
            avatar.content_type,
        )
        .execute(&mut *self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn delete_user_avatar(
        &self,
        user: &UserID,
        id: &AvatarID,
    ) -> sqlx::Result<()> {
        query!(r#"delete from user_avatars where user_id = $1 and id = $2"#, user as _, id as _,)
            .execute(&mut *self.get().await?.tx)
            .await?;
        Ok(())
    }

    //
    //
    //

    pub(crate) async fn add_verify_code(
        &self,
        user: &UserID,
        session: &SessionID,
        code: &str,
    ) -> sqlx::Result<()> {
        query!(
            r#"insert into verify_codes (user_id, session_id, code)
               values ($1, $2, $3)"#,
            user as _,
            session as _,
            code,
        )
        .execute(&mut *self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn get_verify_code(
        &self,
        user: &UserID,
    ) -> sqlx::Result<(String, VerifyCode)> {
        query!(
            r#"select user_id as "user_id: UserID", session_id as "session_id: SessionID", code,
                      email
               from verify_codes join users using (user_id)
               where user_id = $1"#,
            user as _,
        )
        .map(|r| {
            (r.email, VerifyCode { user_id: r.user_id, session_id: r.session_id, code: r.code })
        })
        .fetch_one(&mut *self.get().await?.tx)
        .await
    }

    pub(crate) async fn try_use_verify_code(
        &self,
        user: &UserID,
        code: &str,
    ) -> sqlx::Result<Option<VerifyCode>> {
        query_as!(
            VerifyCode,
            r#"delete from verify_codes
               where user_id = $1 and code = $2
               returning user_id as "user_id: UserID", session_id as "session_id: SessionID",
                         code"#,
            user as _,
            code,
        )
        .fetch_optional(&mut *self.get().await?.tx)
        .await
    }

    //
    //
    //

    pub(crate) async fn add_password_change(
        &self,
        user: &UserID,
        id: &PasswordChangeID,
        key: &HawkKey,
        forgot_code: Option<&str>,
    ) -> sqlx::Result<()> {
        query!(
            r#"insert into password_change_tokens (id, user_id, hmac_key, forgot_code)
               values ($1, $2, $3, $4)
               on conflict (user_id) do update set id = $1, hmac_key = $3, forgot_code = $4,
                   expires_at = default"#,
            id as _,
            user as _,
            key as _,
            forgot_code,
        )
        .execute(&mut *self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn finish_password_change(
        &self,
        id: &PasswordChangeID,
        is_forgot: bool,
    ) -> sqlx::Result<(HawkKey, (UserID, Option<String>))> {
        query!(
            r#"delete from password_change_tokens
               where id = $1 and expires_at > now() and (forgot_code is not null) = $2
               returning hmac_key as "hmac_key: HawkKey", user_id as "user_id: UserID",
                         forgot_code"#,
            id as _,
            is_forgot,
        )
        .map(|r| (r.hmac_key, (r.user_id, r.forgot_code)))
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    pub(crate) async fn add_account_reset(
        &self,
        user: &UserID,
        id: &AccountResetID,
        key: &HawkKey,
    ) -> sqlx::Result<()> {
        query!(
            r#"insert into account_reset_tokens (id, user_id, hmac_key)
               values ($1, $2, $3)
               on conflict (user_id) do update set id = $1, hmac_key = $3, expires_at = default"#,
            id as _,
            user as _,
            key as _,
        )
        .execute(&mut *self.get().await?.tx)
        .await?;
        Ok(())
    }

    pub(crate) async fn finish_account_reset(
        &self,
        id: &AccountResetID,
    ) -> sqlx::Result<(HawkKey, UserID)> {
        query!(
            r#"delete from account_reset_tokens
               where id = $1 and expires_at > now()
               returning hmac_key as "hmac_key: HawkKey", user_id as "user_id: UserID""#,
            id as _,
        )
        .map(|r| (r.hmac_key, r.user_id))
        .fetch_one(&mut self.get().await?.tx)
        .await
    }

    //
    //
    //

    pub async fn add_invite_code(&self, code: &str, expires: DateTime<Utc>) -> sqlx::Result<()> {
        query!(r#"insert into invite_codes (code, expires_at) values ($1, $2)"#, code, expires,)
            .execute(&mut self.get().await?.tx)
            .await?;
        Ok(())
    }

    pub(crate) async fn use_invite_code(&self, code: &str) -> sqlx::Result<()> {
        query_scalar!(
            r#"delete from invite_codes where code = $1 and expires_at > now() returning 1"#,
            code,
        )
        .fetch_one(&mut self.get().await?.tx)
        .await?;
        Ok(())
    }

    //
    //
    //

    pub(crate) async fn prune_expired_tokens(&self) -> sqlx::Result<()> {
        query!("call prune_expired_tokens()").execute(&mut self.get().await?.tx).await?;
        Ok(())
    }

    pub(crate) async fn prune_expired_verify_codes(&self) -> sqlx::Result<()> {
        query!("call prune_expired_verify_codes()").execute(&mut self.get().await?.tx).await?;
        Ok(())
    }
}

fn decode_err<E: Error + Send + Sync + 'static>(c: &str) -> impl FnOnce(E) -> sqlx::Error {
    let index = c.to_string();
    move |e| sqlx::Error::ColumnDecode { index, source: Box::new(e) }
}