mod access_token;
mod contact;
mod project;
mod project_collaborator;
mod room;
mod room_participant;
mod signup;
#[cfg(test)]
mod tests;
mod user;
mod worktree;

use crate::{Error, Result};
use anyhow::anyhow;
use collections::HashMap;
pub use contact::Contact;
use dashmap::DashMap;
use futures::StreamExt;
use hyper::StatusCode;
use rpc::{proto, ConnectionId};
pub use sea_orm::ConnectOptions;
use sea_orm::{
    entity::prelude::*, ActiveValue, ConnectionTrait, DatabaseBackend, DatabaseConnection,
    DatabaseTransaction, DbErr, FromQueryResult, IntoActiveModel, JoinType, QueryOrder,
    QuerySelect, Statement, TransactionTrait,
};
use sea_query::{Alias, Expr, OnConflict, Query};
use serde::{Deserialize, Serialize};
pub use signup::{Invite, NewSignup, WaitlistSummary};
use sqlx::migrate::{Migrate, Migration, MigrationSource};
use sqlx::Connection;
use std::ops::{Deref, DerefMut};
use std::path::Path;
use std::time::Duration;
use std::{future::Future, marker::PhantomData, rc::Rc, sync::Arc};
use tokio::sync::{Mutex, OwnedMutexGuard};
pub use user::Model as User;

pub struct Database {
    options: ConnectOptions,
    pool: DatabaseConnection,
    rooms: DashMap<RoomId, Arc<Mutex<()>>>,
    #[cfg(test)]
    background: Option<std::sync::Arc<gpui::executor::Background>>,
    #[cfg(test)]
    runtime: Option<tokio::runtime::Runtime>,
}

impl Database {
    pub async fn new(options: ConnectOptions) -> Result<Self> {
        Ok(Self {
            options: options.clone(),
            pool: sea_orm::Database::connect(options).await?,
            rooms: DashMap::with_capacity(16384),
            #[cfg(test)]
            background: None,
            #[cfg(test)]
            runtime: None,
        })
    }

    pub async fn migrate(
        &self,
        migrations_path: &Path,
        ignore_checksum_mismatch: bool,
    ) -> anyhow::Result<Vec<(Migration, Duration)>> {
        let migrations = MigrationSource::resolve(migrations_path)
            .await
            .map_err(|err| anyhow!("failed to load migrations: {err:?}"))?;

        let mut connection = sqlx::AnyConnection::connect(self.options.get_url()).await?;

        connection.ensure_migrations_table().await?;
        let applied_migrations: HashMap<_, _> = connection
            .list_applied_migrations()
            .await?
            .into_iter()
            .map(|m| (m.version, m))
            .collect();

        let mut new_migrations = Vec::new();
        for migration in migrations {
            match applied_migrations.get(&migration.version) {
                Some(applied_migration) => {
                    if migration.checksum != applied_migration.checksum && !ignore_checksum_mismatch
                    {
                        Err(anyhow!(
                            "checksum mismatch for applied migration {}",
                            migration.description
                        ))?;
                    }
                }
                None => {
                    let elapsed = connection.apply(&migration).await?;
                    new_migrations.push((migration, elapsed));
                }
            }
        }

        Ok(new_migrations)
    }

    // users

    pub async fn create_user(
        &self,
        email_address: &str,
        admin: bool,
        params: NewUserParams,
    ) -> Result<NewUserResult> {
        self.transact(|tx| async {
            let user = user::Entity::insert(user::ActiveModel {
                email_address: ActiveValue::set(Some(email_address.into())),
                github_login: ActiveValue::set(params.github_login.clone()),
                github_user_id: ActiveValue::set(Some(params.github_user_id)),
                admin: ActiveValue::set(admin),
                metrics_id: ActiveValue::set(Uuid::new_v4()),
                ..Default::default()
            })
            .on_conflict(
                OnConflict::column(user::Column::GithubLogin)
                    .update_column(user::Column::GithubLogin)
                    .to_owned(),
            )
            .exec_with_returning(&tx)
            .await?;

            tx.commit().await?;

            Ok(NewUserResult {
                user_id: user.id,
                metrics_id: user.metrics_id.to_string(),
                signup_device_id: None,
                inviting_user_id: None,
            })
        })
        .await
    }

    pub async fn get_user_by_id(&self, id: UserId) -> Result<Option<user::Model>> {
        self.transact(|tx| async move { Ok(user::Entity::find_by_id(id).one(&tx).await?) })
            .await
    }

    pub async fn get_users_by_ids(&self, ids: Vec<UserId>) -> Result<Vec<user::Model>> {
        self.transact(|tx| async {
            let tx = tx;
            Ok(user::Entity::find()
                .filter(user::Column::Id.is_in(ids.iter().copied()))
                .all(&tx)
                .await?)
        })
        .await
    }

    pub async fn get_user_by_github_account(
        &self,
        github_login: &str,
        github_user_id: Option<i32>,
    ) -> Result<Option<User>> {
        self.transact(|tx| async {
            let tx = tx;
            if let Some(github_user_id) = github_user_id {
                if let Some(user_by_github_user_id) = user::Entity::find()
                    .filter(user::Column::GithubUserId.eq(github_user_id))
                    .one(&tx)
                    .await?
                {
                    let mut user_by_github_user_id = user_by_github_user_id.into_active_model();
                    user_by_github_user_id.github_login = ActiveValue::set(github_login.into());
                    Ok(Some(user_by_github_user_id.update(&tx).await?))
                } else if let Some(user_by_github_login) = user::Entity::find()
                    .filter(user::Column::GithubLogin.eq(github_login))
                    .one(&tx)
                    .await?
                {
                    let mut user_by_github_login = user_by_github_login.into_active_model();
                    user_by_github_login.github_user_id = ActiveValue::set(Some(github_user_id));
                    Ok(Some(user_by_github_login.update(&tx).await?))
                } else {
                    Ok(None)
                }
            } else {
                Ok(user::Entity::find()
                    .filter(user::Column::GithubLogin.eq(github_login))
                    .one(&tx)
                    .await?)
            }
        })
        .await
    }

    pub async fn get_all_users(&self, page: u32, limit: u32) -> Result<Vec<User>> {
        self.transact(|tx| async move {
            Ok(user::Entity::find()
                .order_by_asc(user::Column::GithubLogin)
                .limit(limit as u64)
                .offset(page as u64 * limit as u64)
                .all(&tx)
                .await?)
        })
        .await
    }

    pub async fn get_users_with_no_invites(
        &self,
        invited_by_another_user: bool,
    ) -> Result<Vec<User>> {
        self.transact(|tx| async move {
            Ok(user::Entity::find()
                .filter(
                    user::Column::InviteCount
                        .eq(0)
                        .and(if invited_by_another_user {
                            user::Column::InviterId.is_not_null()
                        } else {
                            user::Column::InviterId.is_null()
                        }),
                )
                .all(&tx)
                .await?)
        })
        .await
    }

    pub async fn get_user_metrics_id(&self, id: UserId) -> Result<String> {
        #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
        enum QueryAs {
            MetricsId,
        }

        self.transact(|tx| async move {
            let metrics_id: Uuid = user::Entity::find_by_id(id)
                .select_only()
                .column(user::Column::MetricsId)
                .into_values::<_, QueryAs>()
                .one(&tx)
                .await?
                .ok_or_else(|| anyhow!("could not find user"))?;
            Ok(metrics_id.to_string())
        })
        .await
    }

    pub async fn set_user_is_admin(&self, id: UserId, is_admin: bool) -> Result<()> {
        self.transact(|tx| async move {
            user::Entity::update_many()
                .filter(user::Column::Id.eq(id))
                .col_expr(user::Column::Admin, is_admin.into())
                .exec(&tx)
                .await?;
            tx.commit().await?;
            Ok(())
        })
        .await
    }

    pub async fn destroy_user(&self, id: UserId) -> Result<()> {
        self.transact(|tx| async move {
            access_token::Entity::delete_many()
                .filter(access_token::Column::UserId.eq(id))
                .exec(&tx)
                .await?;
            user::Entity::delete_by_id(id).exec(&tx).await?;
            tx.commit().await?;
            Ok(())
        })
        .await
    }

    // contacts

    pub async fn get_contacts(&self, user_id: UserId) -> Result<Vec<Contact>> {
        #[derive(Debug, FromQueryResult)]
        struct ContactWithUserBusyStatuses {
            user_id_a: UserId,
            user_id_b: UserId,
            a_to_b: bool,
            accepted: bool,
            should_notify: bool,
            user_a_busy: bool,
            user_b_busy: bool,
        }

        self.transact(|tx| async move {
            let user_a_participant = Alias::new("user_a_participant");
            let user_b_participant = Alias::new("user_b_participant");
            let mut db_contacts = contact::Entity::find()
                .column_as(
                    Expr::tbl(user_a_participant.clone(), room_participant::Column::Id)
                        .is_not_null(),
                    "user_a_busy",
                )
                .column_as(
                    Expr::tbl(user_b_participant.clone(), room_participant::Column::Id)
                        .is_not_null(),
                    "user_b_busy",
                )
                .filter(
                    contact::Column::UserIdA
                        .eq(user_id)
                        .or(contact::Column::UserIdB.eq(user_id)),
                )
                .join_as(
                    JoinType::LeftJoin,
                    contact::Relation::UserARoomParticipant.def(),
                    user_a_participant,
                )
                .join_as(
                    JoinType::LeftJoin,
                    contact::Relation::UserBRoomParticipant.def(),
                    user_b_participant,
                )
                .into_model::<ContactWithUserBusyStatuses>()
                .stream(&tx)
                .await?;

            let mut contacts = Vec::new();
            while let Some(db_contact) = db_contacts.next().await {
                let db_contact = db_contact?;
                if db_contact.user_id_a == user_id {
                    if db_contact.accepted {
                        contacts.push(Contact::Accepted {
                            user_id: db_contact.user_id_b,
                            should_notify: db_contact.should_notify && db_contact.a_to_b,
                            busy: db_contact.user_b_busy,
                        });
                    } else if db_contact.a_to_b {
                        contacts.push(Contact::Outgoing {
                            user_id: db_contact.user_id_b,
                        })
                    } else {
                        contacts.push(Contact::Incoming {
                            user_id: db_contact.user_id_b,
                            should_notify: db_contact.should_notify,
                        });
                    }
                } else if db_contact.accepted {
                    contacts.push(Contact::Accepted {
                        user_id: db_contact.user_id_a,
                        should_notify: db_contact.should_notify && !db_contact.a_to_b,
                        busy: db_contact.user_a_busy,
                    });
                } else if db_contact.a_to_b {
                    contacts.push(Contact::Incoming {
                        user_id: db_contact.user_id_a,
                        should_notify: db_contact.should_notify,
                    });
                } else {
                    contacts.push(Contact::Outgoing {
                        user_id: db_contact.user_id_a,
                    });
                }
            }

            contacts.sort_unstable_by_key(|contact| contact.user_id());

            Ok(contacts)
        })
        .await
    }

    pub async fn has_contact(&self, user_id_1: UserId, user_id_2: UserId) -> Result<bool> {
        self.transact(|tx| async move {
            let (id_a, id_b) = if user_id_1 < user_id_2 {
                (user_id_1, user_id_2)
            } else {
                (user_id_2, user_id_1)
            };

            Ok(contact::Entity::find()
                .filter(
                    contact::Column::UserIdA
                        .eq(id_a)
                        .and(contact::Column::UserIdB.eq(id_b))
                        .and(contact::Column::Accepted.eq(true)),
                )
                .one(&tx)
                .await?
                .is_some())
        })
        .await
    }

    pub async fn send_contact_request(&self, sender_id: UserId, receiver_id: UserId) -> Result<()> {
        self.transact(|tx| async move {
            let (id_a, id_b, a_to_b) = if sender_id < receiver_id {
                (sender_id, receiver_id, true)
            } else {
                (receiver_id, sender_id, false)
            };

            let rows_affected = contact::Entity::insert(contact::ActiveModel {
                user_id_a: ActiveValue::set(id_a),
                user_id_b: ActiveValue::set(id_b),
                a_to_b: ActiveValue::set(a_to_b),
                accepted: ActiveValue::set(false),
                should_notify: ActiveValue::set(true),
                ..Default::default()
            })
            .on_conflict(
                OnConflict::columns([contact::Column::UserIdA, contact::Column::UserIdB])
                    .values([
                        (contact::Column::Accepted, true.into()),
                        (contact::Column::ShouldNotify, false.into()),
                    ])
                    .action_and_where(
                        contact::Column::Accepted.eq(false).and(
                            contact::Column::AToB
                                .eq(a_to_b)
                                .and(contact::Column::UserIdA.eq(id_b))
                                .or(contact::Column::AToB
                                    .ne(a_to_b)
                                    .and(contact::Column::UserIdA.eq(id_a))),
                        ),
                    )
                    .to_owned(),
            )
            .exec_without_returning(&tx)
            .await?;

            if rows_affected == 1 {
                tx.commit().await?;
                Ok(())
            } else {
                Err(anyhow!("contact already requested"))?
            }
        })
        .await
    }

    pub async fn remove_contact(&self, requester_id: UserId, responder_id: UserId) -> Result<()> {
        self.transact(|tx| async move {
            let (id_a, id_b) = if responder_id < requester_id {
                (responder_id, requester_id)
            } else {
                (requester_id, responder_id)
            };

            let result = contact::Entity::delete_many()
                .filter(
                    contact::Column::UserIdA
                        .eq(id_a)
                        .and(contact::Column::UserIdB.eq(id_b)),
                )
                .exec(&tx)
                .await?;

            if result.rows_affected == 1 {
                tx.commit().await?;
                Ok(())
            } else {
                Err(anyhow!("no such contact"))?
            }
        })
        .await
    }

    pub async fn dismiss_contact_notification(
        &self,
        user_id: UserId,
        contact_user_id: UserId,
    ) -> Result<()> {
        self.transact(|tx| async move {
            let (id_a, id_b, a_to_b) = if user_id < contact_user_id {
                (user_id, contact_user_id, true)
            } else {
                (contact_user_id, user_id, false)
            };

            let result = contact::Entity::update_many()
                .set(contact::ActiveModel {
                    should_notify: ActiveValue::set(false),
                    ..Default::default()
                })
                .filter(
                    contact::Column::UserIdA
                        .eq(id_a)
                        .and(contact::Column::UserIdB.eq(id_b))
                        .and(
                            contact::Column::AToB
                                .eq(a_to_b)
                                .and(contact::Column::Accepted.eq(true))
                                .or(contact::Column::AToB
                                    .ne(a_to_b)
                                    .and(contact::Column::Accepted.eq(false))),
                        ),
                )
                .exec(&tx)
                .await?;
            if result.rows_affected == 0 {
                Err(anyhow!("no such contact request"))?
            } else {
                tx.commit().await?;
                Ok(())
            }
        })
        .await
    }

    pub async fn respond_to_contact_request(
        &self,
        responder_id: UserId,
        requester_id: UserId,
        accept: bool,
    ) -> Result<()> {
        self.transact(|tx| async move {
            let (id_a, id_b, a_to_b) = if responder_id < requester_id {
                (responder_id, requester_id, false)
            } else {
                (requester_id, responder_id, true)
            };
            let rows_affected = if accept {
                let result = contact::Entity::update_many()
                    .set(contact::ActiveModel {
                        accepted: ActiveValue::set(true),
                        should_notify: ActiveValue::set(true),
                        ..Default::default()
                    })
                    .filter(
                        contact::Column::UserIdA
                            .eq(id_a)
                            .and(contact::Column::UserIdB.eq(id_b))
                            .and(contact::Column::AToB.eq(a_to_b)),
                    )
                    .exec(&tx)
                    .await?;
                result.rows_affected
            } else {
                let result = contact::Entity::delete_many()
                    .filter(
                        contact::Column::UserIdA
                            .eq(id_a)
                            .and(contact::Column::UserIdB.eq(id_b))
                            .and(contact::Column::AToB.eq(a_to_b))
                            .and(contact::Column::Accepted.eq(false)),
                    )
                    .exec(&tx)
                    .await?;

                result.rows_affected
            };

            if rows_affected == 1 {
                tx.commit().await?;
                Ok(())
            } else {
                Err(anyhow!("no such contact request"))?
            }
        })
        .await
    }

    pub fn fuzzy_like_string(string: &str) -> String {
        let mut result = String::with_capacity(string.len() * 2 + 1);
        for c in string.chars() {
            if c.is_alphanumeric() {
                result.push('%');
                result.push(c);
            }
        }
        result.push('%');
        result
    }

    pub async fn fuzzy_search_users(&self, name_query: &str, limit: u32) -> Result<Vec<User>> {
        self.transact(|tx| async {
            let tx = tx;
            let like_string = Self::fuzzy_like_string(name_query);
            let query = "
                SELECT users.*
                FROM users
                WHERE github_login ILIKE $1
                ORDER BY github_login <-> $2
                LIMIT $3
            ";

            Ok(user::Entity::find()
                .from_raw_sql(Statement::from_sql_and_values(
                    self.pool.get_database_backend(),
                    query.into(),
                    vec![like_string.into(), name_query.into(), limit.into()],
                ))
                .all(&tx)
                .await?)
        })
        .await
    }

    // signups

    pub async fn create_signup(&self, signup: NewSignup) -> Result<()> {
        self.transact(|tx| async {
            signup::ActiveModel {
                email_address: ActiveValue::set(signup.email_address.clone()),
                email_confirmation_code: ActiveValue::set(random_email_confirmation_code()),
                email_confirmation_sent: ActiveValue::set(false),
                platform_mac: ActiveValue::set(signup.platform_mac),
                platform_windows: ActiveValue::set(signup.platform_windows),
                platform_linux: ActiveValue::set(signup.platform_linux),
                platform_unknown: ActiveValue::set(false),
                editor_features: ActiveValue::set(Some(signup.editor_features.clone())),
                programming_languages: ActiveValue::set(Some(signup.programming_languages.clone())),
                device_id: ActiveValue::set(signup.device_id.clone()),
                ..Default::default()
            }
            .insert(&tx)
            .await?;
            tx.commit().await?;
            Ok(())
        })
        .await
    }

    pub async fn get_waitlist_summary(&self) -> Result<WaitlistSummary> {
        self.transact(|tx| async move {
            let query = "
                SELECT
                    COUNT(*) as count,
                    COALESCE(SUM(CASE WHEN platform_linux THEN 1 ELSE 0 END), 0) as linux_count,
                    COALESCE(SUM(CASE WHEN platform_mac THEN 1 ELSE 0 END), 0) as mac_count,
                    COALESCE(SUM(CASE WHEN platform_windows THEN 1 ELSE 0 END), 0) as windows_count,
                    COALESCE(SUM(CASE WHEN platform_unknown THEN 1 ELSE 0 END), 0) as unknown_count
                FROM (
                    SELECT *
                    FROM signups
                    WHERE
                        NOT email_confirmation_sent
                ) AS unsent
            ";
            Ok(
                WaitlistSummary::find_by_statement(Statement::from_sql_and_values(
                    self.pool.get_database_backend(),
                    query.into(),
                    vec![],
                ))
                .one(&tx)
                .await?
                .ok_or_else(|| anyhow!("invalid result"))?,
            )
        })
        .await
    }

    pub async fn record_sent_invites(&self, invites: &[Invite]) -> Result<()> {
        let emails = invites
            .iter()
            .map(|s| s.email_address.as_str())
            .collect::<Vec<_>>();
        self.transact(|tx| async {
            signup::Entity::update_many()
                .filter(signup::Column::EmailAddress.is_in(emails.iter().copied()))
                .col_expr(signup::Column::EmailConfirmationSent, true.into())
                .exec(&tx)
                .await?;
            tx.commit().await?;
            Ok(())
        })
        .await
    }

    pub async fn get_unsent_invites(&self, count: usize) -> Result<Vec<Invite>> {
        self.transact(|tx| async move {
            Ok(signup::Entity::find()
                .select_only()
                .column(signup::Column::EmailAddress)
                .column(signup::Column::EmailConfirmationCode)
                .filter(
                    signup::Column::EmailConfirmationSent.eq(false).and(
                        signup::Column::PlatformMac
                            .eq(true)
                            .or(signup::Column::PlatformUnknown.eq(true)),
                    ),
                )
                .limit(count as u64)
                .into_model()
                .all(&tx)
                .await?)
        })
        .await
    }

    // invite codes

    pub async fn create_invite_from_code(
        &self,
        code: &str,
        email_address: &str,
        device_id: Option<&str>,
    ) -> Result<Invite> {
        self.transact(|tx| async move {
            let existing_user = user::Entity::find()
                .filter(user::Column::EmailAddress.eq(email_address))
                .one(&tx)
                .await?;

            if existing_user.is_some() {
                Err(anyhow!("email address is already in use"))?;
            }

            let inviter = match user::Entity::find()
                .filter(user::Column::InviteCode.eq(code))
                .one(&tx)
                .await?
            {
                Some(inviter) => inviter,
                None => {
                    return Err(Error::Http(
                        StatusCode::NOT_FOUND,
                        "invite code not found".to_string(),
                    ))?
                }
            };

            if inviter.invite_count == 0 {
                Err(Error::Http(
                    StatusCode::UNAUTHORIZED,
                    "no invites remaining".to_string(),
                ))?;
            }

            let signup = signup::Entity::insert(signup::ActiveModel {
                email_address: ActiveValue::set(email_address.into()),
                email_confirmation_code: ActiveValue::set(random_email_confirmation_code()),
                email_confirmation_sent: ActiveValue::set(false),
                inviting_user_id: ActiveValue::set(Some(inviter.id)),
                platform_linux: ActiveValue::set(false),
                platform_mac: ActiveValue::set(false),
                platform_windows: ActiveValue::set(false),
                platform_unknown: ActiveValue::set(true),
                device_id: ActiveValue::set(device_id.map(|device_id| device_id.into())),
                ..Default::default()
            })
            .on_conflict(
                OnConflict::column(signup::Column::EmailAddress)
                    .update_column(signup::Column::InvitingUserId)
                    .to_owned(),
            )
            .exec_with_returning(&tx)
            .await?;
            tx.commit().await?;

            Ok(Invite {
                email_address: signup.email_address,
                email_confirmation_code: signup.email_confirmation_code,
            })
        })
        .await
    }

    pub async fn create_user_from_invite(
        &self,
        invite: &Invite,
        user: NewUserParams,
    ) -> Result<Option<NewUserResult>> {
        self.transact(|tx| async {
            let tx = tx;
            let signup = signup::Entity::find()
                .filter(
                    signup::Column::EmailAddress
                        .eq(invite.email_address.as_str())
                        .and(
                            signup::Column::EmailConfirmationCode
                                .eq(invite.email_confirmation_code.as_str()),
                        ),
                )
                .one(&tx)
                .await?
                .ok_or_else(|| Error::Http(StatusCode::NOT_FOUND, "no such invite".to_string()))?;

            if signup.user_id.is_some() {
                return Ok(None);
            }

            let user = user::Entity::insert(user::ActiveModel {
                email_address: ActiveValue::set(Some(invite.email_address.clone())),
                github_login: ActiveValue::set(user.github_login.clone()),
                github_user_id: ActiveValue::set(Some(user.github_user_id)),
                admin: ActiveValue::set(false),
                invite_count: ActiveValue::set(user.invite_count),
                invite_code: ActiveValue::set(Some(random_invite_code())),
                metrics_id: ActiveValue::set(Uuid::new_v4()),
                ..Default::default()
            })
            .on_conflict(
                OnConflict::column(user::Column::GithubLogin)
                    .update_columns([
                        user::Column::EmailAddress,
                        user::Column::GithubUserId,
                        user::Column::Admin,
                    ])
                    .to_owned(),
            )
            .exec_with_returning(&tx)
            .await?;

            let mut signup = signup.into_active_model();
            signup.user_id = ActiveValue::set(Some(user.id));
            let signup = signup.update(&tx).await?;

            if let Some(inviting_user_id) = signup.inviting_user_id {
                let result = user::Entity::update_many()
                    .filter(
                        user::Column::Id
                            .eq(inviting_user_id)
                            .and(user::Column::InviteCount.gt(0)),
                    )
                    .col_expr(
                        user::Column::InviteCount,
                        Expr::col(user::Column::InviteCount).sub(1),
                    )
                    .exec(&tx)
                    .await?;

                if result.rows_affected == 0 {
                    Err(Error::Http(
                        StatusCode::UNAUTHORIZED,
                        "no invites remaining".to_string(),
                    ))?;
                }

                contact::Entity::insert(contact::ActiveModel {
                    user_id_a: ActiveValue::set(inviting_user_id),
                    user_id_b: ActiveValue::set(user.id),
                    a_to_b: ActiveValue::set(true),
                    should_notify: ActiveValue::set(true),
                    accepted: ActiveValue::set(true),
                    ..Default::default()
                })
                .on_conflict(OnConflict::new().do_nothing().to_owned())
                .exec_without_returning(&tx)
                .await?;
            }

            tx.commit().await?;
            Ok(Some(NewUserResult {
                user_id: user.id,
                metrics_id: user.metrics_id.to_string(),
                inviting_user_id: signup.inviting_user_id,
                signup_device_id: signup.device_id,
            }))
        })
        .await
    }

    pub async fn set_invite_count_for_user(&self, id: UserId, count: u32) -> Result<()> {
        self.transact(|tx| async move {
            if count > 0 {
                user::Entity::update_many()
                    .filter(
                        user::Column::Id
                            .eq(id)
                            .and(user::Column::InviteCode.is_null()),
                    )
                    .col_expr(user::Column::InviteCode, random_invite_code().into())
                    .exec(&tx)
                    .await?;
            }

            user::Entity::update_many()
                .filter(user::Column::Id.eq(id))
                .col_expr(user::Column::InviteCount, count.into())
                .exec(&tx)
                .await?;
            tx.commit().await?;
            Ok(())
        })
        .await
    }

    pub async fn get_invite_code_for_user(&self, id: UserId) -> Result<Option<(String, u32)>> {
        self.transact(|tx| async move {
            match user::Entity::find_by_id(id).one(&tx).await? {
                Some(user) if user.invite_code.is_some() => {
                    Ok(Some((user.invite_code.unwrap(), user.invite_count as u32)))
                }
                _ => Ok(None),
            }
        })
        .await
    }

    pub async fn get_user_for_invite_code(&self, code: &str) -> Result<User> {
        self.transact(|tx| async move {
            user::Entity::find()
                .filter(user::Column::InviteCode.eq(code))
                .one(&tx)
                .await?
                .ok_or_else(|| {
                    Error::Http(
                        StatusCode::NOT_FOUND,
                        "that invite code does not exist".to_string(),
                    )
                })
        })
        .await
    }

    // projects

    pub async fn share_project(
        &self,
        room_id: RoomId,
        connection_id: ConnectionId,
        worktrees: &[proto::WorktreeMetadata],
    ) -> Result<RoomGuard<(ProjectId, proto::Room)>> {
        self.transact(|tx| async move {
            let participant = room_participant::Entity::find()
                .filter(room_participant::Column::AnsweringConnectionId.eq(connection_id.0))
                .one(&tx)
                .await?
                .ok_or_else(|| anyhow!("could not find participant"))?;
            if participant.room_id != room_id {
                return Err(anyhow!("shared project on unexpected room"))?;
            }

            let project = project::ActiveModel {
                room_id: ActiveValue::set(participant.room_id),
                host_user_id: ActiveValue::set(participant.user_id),
                host_connection_id: ActiveValue::set(connection_id.0 as i32),
                ..Default::default()
            }
            .insert(&tx)
            .await?;

            worktree::Entity::insert_many(worktrees.iter().map(|worktree| worktree::ActiveModel {
                id: ActiveValue::set(worktree.id as i32),
                project_id: ActiveValue::set(project.id),
                abs_path: ActiveValue::set(worktree.abs_path.clone()),
                root_name: ActiveValue::set(worktree.root_name.clone()),
                visible: ActiveValue::set(worktree.visible),
                scan_id: ActiveValue::set(0),
                is_complete: ActiveValue::set(false),
            }))
            .exec(&tx)
            .await?;

            project_collaborator::ActiveModel {
                project_id: ActiveValue::set(project.id),
                connection_id: ActiveValue::set(connection_id.0 as i32),
                user_id: ActiveValue::set(participant.user_id),
                replica_id: ActiveValue::set(0),
                is_host: ActiveValue::set(true),
                ..Default::default()
            }
            .insert(&tx)
            .await?;

            let room = self.get_room(room_id, &tx).await?;
            self.commit_room_transaction(room_id, tx, (project.id, room))
                .await
        })
        .await
    }

    async fn get_room(&self, room_id: RoomId, tx: &DatabaseTransaction) -> Result<proto::Room> {
        let db_room = room::Entity::find_by_id(room_id)
            .one(tx)
            .await?
            .ok_or_else(|| anyhow!("could not find room"))?;

        let mut db_participants = db_room
            .find_related(room_participant::Entity)
            .stream(tx)
            .await?;
        let mut participants = HashMap::default();
        let mut pending_participants = Vec::new();
        while let Some(db_participant) = db_participants.next().await {
            let db_participant = db_participant?;
            if let Some(answering_connection_id) = db_participant.answering_connection_id {
                let location = match (
                    db_participant.location_kind,
                    db_participant.location_project_id,
                ) {
                    (Some(0), Some(project_id)) => {
                        Some(proto::participant_location::Variant::SharedProject(
                            proto::participant_location::SharedProject {
                                id: project_id.to_proto(),
                            },
                        ))
                    }
                    (Some(1), _) => Some(proto::participant_location::Variant::UnsharedProject(
                        Default::default(),
                    )),
                    _ => Some(proto::participant_location::Variant::External(
                        Default::default(),
                    )),
                };
                participants.insert(
                    answering_connection_id,
                    proto::Participant {
                        user_id: db_participant.user_id.to_proto(),
                        peer_id: answering_connection_id as u32,
                        projects: Default::default(),
                        location: Some(proto::ParticipantLocation { variant: location }),
                    },
                );
            } else {
                pending_participants.push(proto::PendingParticipant {
                    user_id: db_participant.user_id.to_proto(),
                    calling_user_id: db_participant.calling_user_id.to_proto(),
                    initial_project_id: db_participant.initial_project_id.map(|id| id.to_proto()),
                });
            }
        }

        let mut db_projects = db_room
            .find_related(project::Entity)
            .find_with_related(worktree::Entity)
            .stream(tx)
            .await?;

        while let Some(row) = db_projects.next().await {
            let (db_project, db_worktree) = row?;
            if let Some(participant) = participants.get_mut(&db_project.host_connection_id) {
                let project = if let Some(project) = participant
                    .projects
                    .iter_mut()
                    .find(|project| project.id == db_project.id.to_proto())
                {
                    project
                } else {
                    participant.projects.push(proto::ParticipantProject {
                        id: db_project.id.to_proto(),
                        worktree_root_names: Default::default(),
                    });
                    participant.projects.last_mut().unwrap()
                };

                if let Some(db_worktree) = db_worktree {
                    project.worktree_root_names.push(db_worktree.root_name);
                }
            }
        }

        Ok(proto::Room {
            id: db_room.id.to_proto(),
            live_kit_room: db_room.live_kit_room,
            participants: participants.into_values().collect(),
            pending_participants,
        })
    }

    async fn commit_room_transaction<T>(
        &self,
        room_id: RoomId,
        tx: DatabaseTransaction,
        data: T,
    ) -> Result<RoomGuard<T>> {
        let lock = self.rooms.entry(room_id).or_default().clone();
        let _guard = lock.lock_owned().await;
        tx.commit().await?;
        Ok(RoomGuard {
            data,
            _guard,
            _not_send: PhantomData,
        })
    }

    pub async fn create_access_token_hash(
        &self,
        user_id: UserId,
        access_token_hash: &str,
        max_access_token_count: usize,
    ) -> Result<()> {
        self.transact(|tx| async {
            let tx = tx;

            access_token::ActiveModel {
                user_id: ActiveValue::set(user_id),
                hash: ActiveValue::set(access_token_hash.into()),
                ..Default::default()
            }
            .insert(&tx)
            .await?;

            access_token::Entity::delete_many()
                .filter(
                    access_token::Column::Id.in_subquery(
                        Query::select()
                            .column(access_token::Column::Id)
                            .from(access_token::Entity)
                            .and_where(access_token::Column::UserId.eq(user_id))
                            .order_by(access_token::Column::Id, sea_orm::Order::Desc)
                            .limit(10000)
                            .offset(max_access_token_count as u64)
                            .to_owned(),
                    ),
                )
                .exec(&tx)
                .await?;
            tx.commit().await?;
            Ok(())
        })
        .await
    }

    pub async fn get_access_token_hashes(&self, user_id: UserId) -> Result<Vec<String>> {
        #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
        enum QueryAs {
            Hash,
        }

        self.transact(|tx| async move {
            Ok(access_token::Entity::find()
                .select_only()
                .column(access_token::Column::Hash)
                .filter(access_token::Column::UserId.eq(user_id))
                .order_by_desc(access_token::Column::Id)
                .into_values::<_, QueryAs>()
                .all(&tx)
                .await?)
        })
        .await
    }

    async fn transact<F, Fut, T>(&self, f: F) -> Result<T>
    where
        F: Send + Fn(DatabaseTransaction) -> Fut,
        Fut: Send + Future<Output = Result<T>>,
    {
        let body = async {
            loop {
                let tx = self.pool.begin().await?;

                // In Postgres, serializable transactions are opt-in
                if let DatabaseBackend::Postgres = self.pool.get_database_backend() {
                    tx.execute(Statement::from_string(
                        DatabaseBackend::Postgres,
                        "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;".into(),
                    ))
                    .await?;
                }

                match f(tx).await {
                    Ok(result) => return Ok(result),
                    Err(error) => match error {
                        Error::Database2(
                            DbErr::Exec(sea_orm::RuntimeErr::SqlxError(error))
                            | DbErr::Query(sea_orm::RuntimeErr::SqlxError(error)),
                        ) if error
                            .as_database_error()
                            .and_then(|error| error.code())
                            .as_deref()
                            == Some("40001") =>
                        {
                            // Retry (don't break the loop)
                        }
                        error @ _ => return Err(error),
                    },
                }
            }
        };

        #[cfg(test)]
        {
            if let Some(background) = self.background.as_ref() {
                background.simulate_random_delay().await;
            }

            self.runtime.as_ref().unwrap().block_on(body)
        }

        #[cfg(not(test))]
        {
            body.await
        }
    }
}

pub struct RoomGuard<T> {
    data: T,
    _guard: OwnedMutexGuard<()>,
    _not_send: PhantomData<Rc<()>>,
}

impl<T> Deref for RoomGuard<T> {
    type Target = T;

    fn deref(&self) -> &T {
        &self.data
    }
}

impl<T> DerefMut for RoomGuard<T> {
    fn deref_mut(&mut self) -> &mut T {
        &mut self.data
    }
}

#[derive(Debug, Serialize, Deserialize)]
pub struct NewUserParams {
    pub github_login: String,
    pub github_user_id: i32,
    pub invite_count: i32,
}

#[derive(Debug)]
pub struct NewUserResult {
    pub user_id: UserId,
    pub metrics_id: String,
    pub inviting_user_id: Option<UserId>,
    pub signup_device_id: Option<String>,
}

fn random_invite_code() -> String {
    nanoid::nanoid!(16)
}

fn random_email_confirmation_code() -> String {
    nanoid::nanoid!(64)
}

macro_rules! id_type {
    ($name:ident) => {
        #[derive(
            Clone,
            Copy,
            Debug,
            Default,
            PartialEq,
            Eq,
            PartialOrd,
            Ord,
            Hash,
            sqlx::Type,
            Serialize,
            Deserialize,
        )]
        #[sqlx(transparent)]
        #[serde(transparent)]
        pub struct $name(pub i32);

        impl $name {
            #[allow(unused)]
            pub const MAX: Self = Self(i32::MAX);

            #[allow(unused)]
            pub fn from_proto(value: u64) -> Self {
                Self(value as i32)
            }

            #[allow(unused)]
            pub fn to_proto(self) -> u64 {
                self.0 as u64
            }
        }

        impl std::fmt::Display for $name {
            fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
                self.0.fmt(f)
            }
        }

        impl From<$name> for sea_query::Value {
            fn from(value: $name) -> Self {
                sea_query::Value::Int(Some(value.0))
            }
        }

        impl sea_orm::TryGetable for $name {
            fn try_get(
                res: &sea_orm::QueryResult,
                pre: &str,
                col: &str,
            ) -> Result<Self, sea_orm::TryGetError> {
                Ok(Self(i32::try_get(res, pre, col)?))
            }
        }

        impl sea_query::ValueType for $name {
            fn try_from(v: Value) -> Result<Self, sea_query::ValueTypeErr> {
                match v {
                    Value::TinyInt(Some(int)) => {
                        Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
                    }
                    Value::SmallInt(Some(int)) => {
                        Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
                    }
                    Value::Int(Some(int)) => {
                        Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
                    }
                    Value::BigInt(Some(int)) => {
                        Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
                    }
                    Value::TinyUnsigned(Some(int)) => {
                        Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
                    }
                    Value::SmallUnsigned(Some(int)) => {
                        Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
                    }
                    Value::Unsigned(Some(int)) => {
                        Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
                    }
                    Value::BigUnsigned(Some(int)) => {
                        Ok(Self(int.try_into().map_err(|_| sea_query::ValueTypeErr)?))
                    }
                    _ => Err(sea_query::ValueTypeErr),
                }
            }

            fn type_name() -> String {
                stringify!($name).into()
            }

            fn array_type() -> sea_query::ArrayType {
                sea_query::ArrayType::Int
            }

            fn column_type() -> sea_query::ColumnType {
                sea_query::ColumnType::Integer(None)
            }
        }

        impl sea_orm::TryFromU64 for $name {
            fn try_from_u64(n: u64) -> Result<Self, DbErr> {
                Ok(Self(n.try_into().map_err(|_| {
                    DbErr::ConvertFromU64(concat!(
                        "error converting ",
                        stringify!($name),
                        " to u64"
                    ))
                })?))
            }
        }

        impl sea_query::Nullable for $name {
            fn null() -> Value {
                Value::Int(None)
            }
        }
    };
}

id_type!(AccessTokenId);
id_type!(ContactId);
id_type!(UserId);
id_type!(RoomId);
id_type!(RoomParticipantId);
id_type!(ProjectId);
id_type!(ProjectCollaboratorId);
id_type!(SignupId);
id_type!(WorktreeId);

#[cfg(test)]
pub use test::*;

#[cfg(test)]
mod test {
    use super::*;
    use gpui::executor::Background;
    use lazy_static::lazy_static;
    use parking_lot::Mutex;
    use rand::prelude::*;
    use sea_orm::ConnectionTrait;
    use sqlx::migrate::MigrateDatabase;
    use std::sync::Arc;

    pub struct TestDb {
        pub db: Option<Arc<Database>>,
        pub connection: Option<sqlx::AnyConnection>,
    }

    impl TestDb {
        pub fn sqlite(background: Arc<Background>) -> Self {
            let url = format!("sqlite::memory:");
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_io()
                .enable_time()
                .build()
                .unwrap();

            let mut db = runtime.block_on(async {
                let mut options = ConnectOptions::new(url);
                options.max_connections(5);
                let db = Database::new(options).await.unwrap();
                let sql = include_str!(concat!(
                    env!("CARGO_MANIFEST_DIR"),
                    "/migrations.sqlite/20221109000000_test_schema.sql"
                ));
                db.pool
                    .execute(sea_orm::Statement::from_string(
                        db.pool.get_database_backend(),
                        sql.into(),
                    ))
                    .await
                    .unwrap();
                db
            });

            db.background = Some(background);
            db.runtime = Some(runtime);

            Self {
                db: Some(Arc::new(db)),
                connection: None,
            }
        }

        pub fn postgres(background: Arc<Background>) -> Self {
            lazy_static! {
                static ref LOCK: Mutex<()> = Mutex::new(());
            }

            let _guard = LOCK.lock();
            let mut rng = StdRng::from_entropy();
            let url = format!(
                "postgres://postgres@localhost/zed-test-{}",
                rng.gen::<u128>()
            );
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_io()
                .enable_time()
                .build()
                .unwrap();

            let mut db = runtime.block_on(async {
                sqlx::Postgres::create_database(&url)
                    .await
                    .expect("failed to create test db");
                let mut options = ConnectOptions::new(url);
                options
                    .max_connections(5)
                    .idle_timeout(Duration::from_secs(0));
                let db = Database::new(options).await.unwrap();
                let migrations_path = concat!(env!("CARGO_MANIFEST_DIR"), "/migrations");
                db.migrate(Path::new(migrations_path), false).await.unwrap();
                db
            });

            db.background = Some(background);
            db.runtime = Some(runtime);

            Self {
                db: Some(Arc::new(db)),
                connection: None,
            }
        }

        pub fn db(&self) -> &Arc<Database> {
            self.db.as_ref().unwrap()
        }
    }

    impl Drop for TestDb {
        fn drop(&mut self) {
            let db = self.db.take().unwrap();
            if let DatabaseBackend::Postgres = db.pool.get_database_backend() {
                db.runtime.as_ref().unwrap().block_on(async {
                    use util::ResultExt;
                    let query = "
                        SELECT pg_terminate_backend(pg_stat_activity.pid)
                        FROM pg_stat_activity
                        WHERE
                            pg_stat_activity.datname = current_database() AND
                            pid <> pg_backend_pid();
                    ";
                    db.pool
                        .execute(sea_orm::Statement::from_string(
                            db.pool.get_database_backend(),
                            query.into(),
                        ))
                        .await
                        .log_err();
                    sqlx::Postgres::drop_database(db.options.get_url())
                        .await
                        .log_err();
                })
            }
        }
    }
}