gitr/src/db.rs

use std::{
    path::Path,
    sync::{Arc, Mutex},
    time::{SystemTime, UNIX_EPOCH},
};

use rusqlite::{Connection, OptionalExtension, Transaction, params};

use crate::{
    error::{AppError, AppResult},
    models::{
        AccessMode, AccessToken, Collaboration, CollaboratorResponse, PullRequest,
        PullRequestStatus, Repository, RepositoryWithOwner, User,
    },
};

#[derive(Clone)]
pub struct Database {
    conn: Arc<Mutex<Connection>>,
}

impl Database {
    pub fn open(path: &Path) -> AppResult<Self> {
        let conn = Connection::open(path)?;
        conn.execute_batch(
            r#"
            PRAGMA journal_mode = WAL;
            PRAGMA foreign_keys = ON;
            PRAGMA synchronous = NORMAL;
            PRAGMA temp_store = MEMORY;
            "#,
        )?;
        Ok(Self {
            conn: Arc::new(Mutex::new(conn)),
        })
    }

    pub fn init_schema(&self) -> AppResult<()> {
        let conn = self.lock()?;
        conn.execute_batch(
            r#"
            CREATE TABLE IF NOT EXISTS user (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                lower_name TEXT NOT NULL UNIQUE,
                name TEXT NOT NULL UNIQUE,
                full_name TEXT NOT NULL DEFAULT '',
                email TEXT NOT NULL UNIQUE,
                password_hash TEXT NOT NULL,
                is_active INTEGER NOT NULL DEFAULT 1,
                is_admin INTEGER NOT NULL DEFAULT 0,
                created_unix INTEGER NOT NULL,
                updated_unix INTEGER NOT NULL
            );

            CREATE TABLE IF NOT EXISTS repository (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                owner_id INTEGER NOT NULL,
                lower_name TEXT NOT NULL,
                name TEXT NOT NULL,
                description TEXT NOT NULL DEFAULT '',
                default_branch TEXT NOT NULL,
                is_private INTEGER NOT NULL DEFAULT 0,
                is_bare INTEGER NOT NULL DEFAULT 1,
                is_fork INTEGER NOT NULL DEFAULT 0,
                fork_id INTEGER NOT NULL DEFAULT 0,
                created_unix INTEGER NOT NULL,
                updated_unix INTEGER NOT NULL,
                UNIQUE(owner_id, lower_name),
                FOREIGN KEY(owner_id) REFERENCES user(id) ON DELETE CASCADE
            );

            CREATE TABLE IF NOT EXISTS access_token (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                name TEXT NOT NULL,
                token_hash TEXT NOT NULL UNIQUE,
                created_unix INTEGER NOT NULL,
                updated_unix INTEGER NOT NULL DEFAULT 0,
                FOREIGN KEY(user_id) REFERENCES user(id) ON DELETE CASCADE
            );

            CREATE UNIQUE INDEX IF NOT EXISTS idx_access_token_user_name
            ON access_token (user_id, name);

            CREATE TABLE IF NOT EXISTS access (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                repo_id INTEGER NOT NULL,
                mode INTEGER NOT NULL,
                UNIQUE(user_id, repo_id),
                FOREIGN KEY(user_id) REFERENCES user(id) ON DELETE CASCADE,
                FOREIGN KEY(repo_id) REFERENCES repository(id) ON DELETE CASCADE
            );

            CREATE TABLE IF NOT EXISTS collaboration (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                repo_id INTEGER NOT NULL,
                mode INTEGER NOT NULL,
                UNIQUE(user_id, repo_id),
                FOREIGN KEY(user_id) REFERENCES user(id) ON DELETE CASCADE,
                FOREIGN KEY(repo_id) REFERENCES repository(id) ON DELETE CASCADE
            );

            CREATE TABLE IF NOT EXISTS pull_request (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                index_in_repo INTEGER NOT NULL,
                title TEXT NOT NULL,
                body TEXT NOT NULL DEFAULT '',
                status INTEGER NOT NULL,
                head_repo_id INTEGER NOT NULL,
                base_repo_id INTEGER NOT NULL,
                head_user_name TEXT NOT NULL,
                head_branch TEXT NOT NULL,
                base_branch TEXT NOT NULL,
                merge_base TEXT NOT NULL,
                merged_commit_id TEXT NOT NULL DEFAULT '',
                poster_id INTEGER NOT NULL,
                has_merged INTEGER NOT NULL DEFAULT 0,
                is_closed INTEGER NOT NULL DEFAULT 0,
                created_unix INTEGER NOT NULL,
                updated_unix INTEGER NOT NULL,
                FOREIGN KEY(head_repo_id) REFERENCES repository(id) ON DELETE CASCADE,
                FOREIGN KEY(base_repo_id) REFERENCES repository(id) ON DELETE CASCADE,
                FOREIGN KEY(poster_id) REFERENCES user(id) ON DELETE CASCADE
            );

            CREATE UNIQUE INDEX IF NOT EXISTS idx_pull_request_base_repo_index
            ON pull_request (base_repo_id, index_in_repo);
            "#,
        )?;
        ensure_column_exists(
            &conn,
            "pull_request",
            "merged_commit_id",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        ensure_column_exists(
            &conn,
            "access_token",
            "updated_unix",
            "INTEGER NOT NULL DEFAULT 0",
        )?;
        Ok(())
    }

    pub fn create_user(&self, new_user: NewUser<'_>) -> AppResult<User> {
        let conn = self.lock()?;
        let tx = conn.unchecked_transaction()?;

        let lower_name = new_user.username.to_ascii_lowercase();
        let email = new_user.email.trim().to_ascii_lowercase();

        if self.user_exists_by_lower_name(&tx, &lower_name)? {
            return Err(AppError::Conflict(format!(
                "user already exists: {}",
                new_user.username
            )));
        }

        if self.user_exists_by_email(&tx, &email)? {
            return Err(AppError::Conflict(format!("email already used: {email}")));
        }

        let now = now_unix();
        tx.execute(
            r#"
            INSERT INTO user (
                lower_name, name, full_name, email, password_hash,
                is_active, is_admin, created_unix, updated_unix
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
            "#,
            params![
                lower_name,
                new_user.username,
                new_user.full_name,
                email,
                new_user.password_hash,
                new_user.is_active,
                new_user.is_admin,
                now,
                now
            ],
        )?;

        let id = tx.last_insert_rowid();
        tx.commit()?;
        drop(conn);
        self.get_user_by_id(id)?
            .ok_or_else(|| AppError::NotFound(format!("user disappeared after create: {id}")))
    }

    pub fn user_count(&self) -> AppResult<i64> {
        let conn = self.lock()?;
        conn.query_row("SELECT COUNT(*) FROM user", [], |row| row.get::<_, i64>(0))
            .map_err(Into::into)
    }

    pub fn get_user_by_id(&self, id: i64) -> AppResult<Option<User>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, lower_name, name, full_name, email, password_hash,
                   is_active, is_admin, created_unix, updated_unix
            FROM user WHERE id = ?1
            "#,
        )?;
        stmt.query_row(params![id], row_to_user)
            .optional()
            .map_err(Into::into)
    }

    pub fn get_user_by_username(&self, username: &str) -> AppResult<Option<User>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, lower_name, name, full_name, email, password_hash,
                   is_active, is_admin, created_unix, updated_unix
            FROM user WHERE lower_name = ?1
            "#,
        )?;
        stmt.query_row(params![username.to_ascii_lowercase()], row_to_user)
            .optional()
            .map_err(Into::into)
    }

    pub fn get_user_by_email(&self, email: &str) -> AppResult<Option<User>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, lower_name, name, full_name, email, password_hash,
                   is_active, is_admin, created_unix, updated_unix
            FROM user WHERE email = ?1
            "#,
        )?;
        stmt.query_row(params![email.trim().to_ascii_lowercase()], row_to_user)
            .optional()
            .map_err(Into::into)
    }

    pub fn create_repository(&self, new_repo: NewRepository<'_>) -> AppResult<Repository> {
        let conn = self.lock()?;
        let tx = conn.unchecked_transaction()?;

        let lower_name = new_repo.name.to_ascii_lowercase();
        if self.repo_exists_by_name(&tx, new_repo.owner_id, &lower_name)? {
            return Err(AppError::Conflict(format!(
                "repository already exists: {}/{}",
                new_repo.owner_name, new_repo.name
            )));
        }

        let now = now_unix();
        tx.execute(
            r#"
            INSERT INTO repository (
                owner_id, lower_name, name, description, default_branch,
                is_private, is_bare, is_fork, fork_id, created_unix, updated_unix
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
            "#,
            params![
                new_repo.owner_id,
                lower_name,
                new_repo.name,
                new_repo.description,
                new_repo.default_branch,
                new_repo.is_private,
                new_repo.is_bare,
                new_repo.is_fork,
                new_repo.fork_id,
                now,
                now
            ],
        )?;

        let id = tx.last_insert_rowid();
        tx.commit()?;
        drop(conn);
        self.get_repository_by_id(id)?
            .ok_or_else(|| AppError::NotFound(format!("repository disappeared after create: {id}")))
    }

    pub fn get_repository_by_id(&self, id: i64) -> AppResult<Option<Repository>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, owner_id, lower_name, name, description, default_branch,
                   is_private, is_bare, is_fork, fork_id, created_unix, updated_unix
            FROM repository WHERE id = ?1
            "#,
        )?;
        stmt.query_row(params![id], row_to_repo)
            .optional()
            .map_err(Into::into)
    }

    pub fn get_repository_by_name(
        &self,
        owner_id: i64,
        name: &str,
    ) -> AppResult<Option<Repository>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, owner_id, lower_name, name, description, default_branch,
                   is_private, is_bare, is_fork, fork_id, created_unix, updated_unix
            FROM repository
            WHERE owner_id = ?1 AND lower_name = ?2
            "#,
        )?;
        stmt.query_row(params![owner_id, name.to_ascii_lowercase()], row_to_repo)
            .optional()
            .map_err(Into::into)
    }

    pub fn list_repositories_with_owners(&self) -> AppResult<Vec<RepositoryWithOwner>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT
                r.id, r.owner_id, r.lower_name, r.name, r.description, r.default_branch,
                r.is_private, r.is_bare, r.is_fork, r.fork_id, r.created_unix, r.updated_unix,
                u.id, u.lower_name, u.name, u.full_name, u.email, u.password_hash,
                u.is_active, u.is_admin, u.created_unix, u.updated_unix
            FROM repository r
            JOIN user u ON u.id = r.owner_id
            ORDER BY u.lower_name ASC, r.lower_name ASC
            "#,
        )?;
        let rows = stmt.query_map([], row_to_repository_with_owner)?;
        let mut repos = Vec::new();
        for row in rows {
            repos.push(row?);
        }
        Ok(repos)
    }

    pub fn list_repositories_with_owners_by_owner(
        &self,
        owner_id: i64,
    ) -> AppResult<Vec<RepositoryWithOwner>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT
                r.id, r.owner_id, r.lower_name, r.name, r.description, r.default_branch,
                r.is_private, r.is_bare, r.is_fork, r.fork_id, r.created_unix, r.updated_unix,
                u.id, u.lower_name, u.name, u.full_name, u.email, u.password_hash,
                u.is_active, u.is_admin, u.created_unix, u.updated_unix
            FROM repository r
            JOIN user u ON u.id = r.owner_id
            WHERE r.owner_id = ?1
            ORDER BY r.lower_name ASC
            "#,
        )?;
        let rows = stmt.query_map(params![owner_id], row_to_repository_with_owner)?;
        let mut repos = Vec::new();
        for row in rows {
            repos.push(row?);
        }
        Ok(repos)
    }

    pub fn has_forked_by(&self, repo_id: i64, user_id: i64) -> AppResult<bool> {
        let conn = self.lock()?;
        let mut stmt =
            conn.prepare("SELECT 1 FROM repository WHERE owner_id = ?1 AND fork_id = ?2 LIMIT 1")?;
        let found = stmt
            .query_row(params![user_id, repo_id], |row| row.get::<_, i64>(0))
            .optional()?;
        Ok(found.is_some())
    }

    pub fn delete_repository_by_id(&self, id: i64) -> AppResult<()> {
        let conn = self.lock()?;
        conn.execute("DELETE FROM repository WHERE id = ?1", params![id])?;
        Ok(())
    }

    pub fn access_mode(
        &self,
        user_id: i64,
        repo_id: i64,
        owner_id: i64,
        private: bool,
    ) -> AppResult<AccessMode> {
        if repo_id <= 0 {
            return Ok(AccessMode::None);
        }

        let mut mode = if private {
            AccessMode::None
        } else {
            AccessMode::Read
        };

        if user_id <= 0 {
            return Ok(mode);
        }
        if user_id == owner_id {
            return Ok(AccessMode::Owner);
        }

        let conn = self.lock()?;
        let mut stmt =
            conn.prepare("SELECT mode FROM access WHERE user_id = ?1 AND repo_id = ?2")?;
        let found = stmt
            .query_row(params![user_id, repo_id], |row| row.get::<_, i64>(0))
            .optional()?;
        if let Some(value) = found {
            mode = access_mode_from_i64(value);
        }
        Ok(mode)
    }

    pub fn authorize(
        &self,
        user_id: i64,
        repo_id: i64,
        desired: AccessMode,
        owner_id: i64,
        private: bool,
    ) -> AppResult<bool> {
        Ok((desired as i64) <= (self.access_mode(user_id, repo_id, owner_id, private)? as i64))
    }

    pub fn set_repo_perms(&self, repo_id: i64, access_map: &[(i64, AccessMode)]) -> AppResult<()> {
        let conn = self.lock()?;
        let tx = conn.unchecked_transaction()?;
        tx.execute("DELETE FROM access WHERE repo_id = ?1", params![repo_id])?;
        for (user_id, mode) in access_map {
            tx.execute(
                "INSERT INTO access (user_id, repo_id, mode) VALUES (?1, ?2, ?3)",
                params![user_id, repo_id, *mode as i64],
            )?;
        }
        tx.commit()?;
        Ok(())
    }

    pub fn upsert_collaboration(
        &self,
        repo_id: i64,
        user_id: i64,
        mode: AccessMode,
    ) -> AppResult<Collaboration> {
        let conn = self.lock()?;
        let tx = conn.unchecked_transaction()?;
        tx.execute(
            r#"
            INSERT INTO collaboration (user_id, repo_id, mode)
            VALUES (?1, ?2, ?3)
            ON CONFLICT(user_id, repo_id) DO UPDATE SET mode = excluded.mode
            "#,
            params![user_id, repo_id, mode as i64],
        )?;
        tx.execute(
            r#"
            INSERT INTO access (user_id, repo_id, mode)
            VALUES (?1, ?2, ?3)
            ON CONFLICT(user_id, repo_id) DO UPDATE SET mode = excluded.mode
            "#,
            params![user_id, repo_id, mode as i64],
        )?;
        let id = tx.query_row(
            "SELECT id FROM collaboration WHERE user_id = ?1 AND repo_id = ?2",
            params![user_id, repo_id],
            |row| row.get::<_, i64>(0),
        )?;
        tx.commit()?;
        drop(conn);
        self.get_collaboration_by_id(id)?.ok_or_else(|| {
            AppError::NotFound(format!("collaboration disappeared after upsert: {id}"))
        })
    }

    pub fn get_collaboration_by_id(&self, id: i64) -> AppResult<Option<Collaboration>> {
        let conn = self.lock()?;
        let mut stmt =
            conn.prepare("SELECT id, user_id, repo_id, mode FROM collaboration WHERE id = ?1")?;
        stmt.query_row(params![id], row_to_collaboration)
            .optional()
            .map_err(Into::into)
    }

    pub fn get_collaborator(
        &self,
        repo_id: i64,
        user_id: i64,
    ) -> AppResult<Option<CollaboratorResponse>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT
                u.id, u.lower_name, u.name, u.full_name, u.email, u.password_hash,
                u.is_active, u.is_admin, u.created_unix, u.updated_unix,
                c.mode
            FROM collaboration c
            JOIN user u ON u.id = c.user_id
            WHERE c.repo_id = ?1 AND c.user_id = ?2
            LIMIT 1
            "#,
        )?;
        stmt.query_row(params![repo_id, user_id], row_to_collaborator_response)
            .optional()
            .map_err(Into::into)
    }

    pub fn list_collaborators(&self, repo_id: i64) -> AppResult<Vec<CollaboratorResponse>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT
                u.id, u.lower_name, u.name, u.full_name, u.email, u.password_hash,
                u.is_active, u.is_admin, u.created_unix, u.updated_unix,
                c.mode
            FROM collaboration c
            JOIN user u ON u.id = c.user_id
            WHERE c.repo_id = ?1
            ORDER BY u.lower_name ASC
            "#,
        )?;
        let rows = stmt.query_map(params![repo_id], row_to_collaborator_response)?;
        let mut collaborators = Vec::new();
        for row in rows {
            collaborators.push(row?);
        }
        Ok(collaborators)
    }

    pub fn delete_collaboration(&self, repo_id: i64, user_id: i64) -> AppResult<()> {
        let conn = self.lock()?;
        let tx = conn.unchecked_transaction()?;
        tx.execute(
            "DELETE FROM collaboration WHERE repo_id = ?1 AND user_id = ?2",
            params![repo_id, user_id],
        )?;
        tx.execute(
            "DELETE FROM access WHERE repo_id = ?1 AND user_id = ?2",
            params![repo_id, user_id],
        )?;
        tx.commit()?;
        Ok(())
    }

    pub fn create_access_token(
        &self,
        user_id: i64,
        name: &str,
        token_hash: &str,
    ) -> AppResult<AccessToken> {
        let conn = self.lock()?;
        let tx = conn.unchecked_transaction()?;
        if self.access_token_exists_by_name(&tx, user_id, name)? {
            return Err(AppError::Conflict(format!(
                "access token already exists: {name}"
            )));
        }
        let now = now_unix();
        tx.execute(
            r#"
            INSERT INTO access_token (user_id, name, token_hash, created_unix, updated_unix)
            VALUES (?1, ?2, ?3, ?4, 0)
            "#,
            params![user_id, name, token_hash, now],
        )?;
        let id = tx.last_insert_rowid();
        tx.commit()?;
        drop(conn);
        self.get_access_token_by_id(id)?.ok_or_else(|| {
            AppError::NotFound(format!("access token disappeared after create: {id}"))
        })
    }

    pub fn get_access_token_by_id(&self, id: i64) -> AppResult<Option<AccessToken>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, user_id, name, token_hash, created_unix, updated_unix
            FROM access_token WHERE id = ?1
            "#,
        )?;
        stmt.query_row(params![id], row_to_access_token)
            .optional()
            .map_err(Into::into)
    }

    pub fn get_access_token_by_hash(&self, token_hash: &str) -> AppResult<Option<AccessToken>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, user_id, name, token_hash, created_unix, updated_unix
            FROM access_token WHERE token_hash = ?1
            "#,
        )?;
        stmt.query_row(params![token_hash], row_to_access_token)
            .optional()
            .map_err(Into::into)
    }

    pub fn list_access_tokens_by_user(&self, user_id: i64) -> AppResult<Vec<AccessToken>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, user_id, name, token_hash, created_unix, updated_unix
            FROM access_token
            WHERE user_id = ?1
            ORDER BY id ASC
            "#,
        )?;
        let rows = stmt.query_map(params![user_id], row_to_access_token)?;
        let mut tokens = Vec::new();
        for row in rows {
            tokens.push(row?);
        }
        Ok(tokens)
    }

    pub fn delete_access_token_by_id(&self, user_id: i64, token_id: i64) -> AppResult<bool> {
        let conn = self.lock()?;
        let affected = conn.execute(
            "DELETE FROM access_token WHERE id = ?1 AND user_id = ?2",
            params![token_id, user_id],
        )?;
        Ok(affected > 0)
    }

    pub fn touch_access_token(&self, token_id: i64) -> AppResult<()> {
        let conn = self.lock()?;
        conn.execute(
            "UPDATE access_token SET updated_unix = ?2 WHERE id = ?1",
            params![token_id, now_unix()],
        )?;
        Ok(())
    }

    pub fn create_pull_request(&self, new_pull: NewPullRequest<'_>) -> AppResult<PullRequest> {
        let conn = self.lock()?;
        let tx = conn.unchecked_transaction()?;
        let now = now_unix();
        let index = tx.query_row(
            "SELECT COALESCE(MAX(index_in_repo), 0) + 1 FROM pull_request WHERE base_repo_id = ?1",
            params![new_pull.base_repo_id],
            |row| row.get::<_, i64>(0),
        )?;
        tx.execute(
            r#"
            INSERT INTO pull_request (
                index_in_repo, title, body, status, head_repo_id, base_repo_id,
                head_user_name, head_branch, base_branch, merge_base, merged_commit_id, poster_id,
                has_merged, is_closed, created_unix, updated_unix
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, '', ?11, 0, 0, ?12, ?13)
            "#,
            params![
                index,
                new_pull.title,
                new_pull.body,
                new_pull.status as i64,
                new_pull.head_repo_id,
                new_pull.base_repo_id,
                new_pull.head_user_name,
                new_pull.head_branch,
                new_pull.base_branch,
                new_pull.merge_base,
                new_pull.poster_id,
                now,
                now
            ],
        )?;
        let id = tx.last_insert_rowid();
        tx.commit()?;
        drop(conn);
        self.get_pull_request_by_id(id)?.ok_or_else(|| {
            AppError::NotFound(format!("pull request disappeared after create: {id}"))
        })
    }

    pub fn get_pull_request_by_id(&self, id: i64) -> AppResult<Option<PullRequest>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, index_in_repo, title, body, status, head_repo_id, base_repo_id,
                   head_user_name, head_branch, base_branch, merge_base, merged_commit_id, poster_id,
                   has_merged, is_closed, created_unix, updated_unix
            FROM pull_request WHERE id = ?1
            "#,
        )?;
        stmt.query_row(params![id], row_to_pull_request)
            .optional()
            .map_err(Into::into)
    }

    pub fn get_unmerged_pull_request(
        &self,
        head_repo_id: i64,
        base_repo_id: i64,
        head_branch: &str,
        base_branch: &str,
    ) -> AppResult<Option<PullRequest>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, index_in_repo, title, body, status, head_repo_id, base_repo_id,
                   head_user_name, head_branch, base_branch, merge_base, merged_commit_id, poster_id,
                   has_merged, is_closed, created_unix, updated_unix
            FROM pull_request
            WHERE head_repo_id = ?1 AND base_repo_id = ?2
              AND head_branch = ?3 AND base_branch = ?4
              AND has_merged = 0 AND is_closed = 0
            LIMIT 1
            "#,
        )?;
        stmt.query_row(
            params![head_repo_id, base_repo_id, head_branch, base_branch],
            row_to_pull_request,
        )
        .optional()
        .map_err(Into::into)
    }

    pub fn get_pull_request_by_base_repo_and_index(
        &self,
        base_repo_id: i64,
        index: i64,
    ) -> AppResult<Option<PullRequest>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, index_in_repo, title, body, status, head_repo_id, base_repo_id,
                   head_user_name, head_branch, base_branch, merge_base, merged_commit_id, poster_id,
                   has_merged, is_closed, created_unix, updated_unix
            FROM pull_request
            WHERE base_repo_id = ?1 AND index_in_repo = ?2
            LIMIT 1
            "#,
        )?;
        stmt.query_row(params![base_repo_id, index], row_to_pull_request)
            .optional()
            .map_err(Into::into)
    }

    pub fn list_pull_requests_by_base_repo(
        &self,
        base_repo_id: i64,
    ) -> AppResult<Vec<PullRequest>> {
        let conn = self.lock()?;
        let mut stmt = conn.prepare(
            r#"
            SELECT id, index_in_repo, title, body, status, head_repo_id, base_repo_id,
                   head_user_name, head_branch, base_branch, merge_base, merged_commit_id, poster_id,
                   has_merged, is_closed, created_unix, updated_unix
            FROM pull_request
            WHERE base_repo_id = ?1
            ORDER BY index_in_repo ASC
            "#,
        )?;
        let rows = stmt.query_map(params![base_repo_id], row_to_pull_request)?;
        let mut pulls = Vec::new();
        for row in rows {
            pulls.push(row?);
        }
        Ok(pulls)
    }

    pub fn mark_pull_request_merged(
        &self,
        id: i64,
        merged_commit_id: &str,
    ) -> AppResult<PullRequest> {
        let conn = self.lock()?;
        let now = now_unix();
        conn.execute(
            r#"
            UPDATE pull_request
            SET has_merged = 1, is_closed = 1, merged_commit_id = ?2, updated_unix = ?3
            WHERE id = ?1
            "#,
            params![id, merged_commit_id, now],
        )?;
        drop(conn);
        self.get_pull_request_by_id(id)?.ok_or_else(|| {
            AppError::NotFound(format!("pull request disappeared after merge: {id}"))
        })
    }

    pub fn update_pull_request_open_state(
        &self,
        id: i64,
        is_closed: bool,
        status: PullRequestStatus,
    ) -> AppResult<PullRequest> {
        let conn = self.lock()?;
        let now = now_unix();
        conn.execute(
            r#"
            UPDATE pull_request
            SET is_closed = ?2, status = ?3, updated_unix = ?4
            WHERE id = ?1
            "#,
            params![id, is_closed, status as i64, now],
        )?;
        drop(conn);
        self.get_pull_request_by_id(id)?.ok_or_else(|| {
            AppError::NotFound(format!("pull request disappeared after state update: {id}"))
        })
    }

    fn user_exists_by_lower_name(&self, tx: &Transaction<'_>, lower_name: &str) -> AppResult<bool> {
        let found = tx
            .query_row(
                "SELECT 1 FROM user WHERE lower_name = ?1 LIMIT 1",
                params![lower_name],
                |row| row.get::<_, i64>(0),
            )
            .optional()?;
        Ok(found.is_some())
    }

    fn user_exists_by_email(&self, tx: &Transaction<'_>, email: &str) -> AppResult<bool> {
        let found = tx
            .query_row(
                "SELECT 1 FROM user WHERE email = ?1 LIMIT 1",
                params![email],
                |row| row.get::<_, i64>(0),
            )
            .optional()?;
        Ok(found.is_some())
    }

    fn repo_exists_by_name(
        &self,
        tx: &Transaction<'_>,
        owner_id: i64,
        lower_name: &str,
    ) -> AppResult<bool> {
        let found = tx
            .query_row(
                "SELECT 1 FROM repository WHERE owner_id = ?1 AND lower_name = ?2 LIMIT 1",
                params![owner_id, lower_name],
                |row| row.get::<_, i64>(0),
            )
            .optional()?;
        Ok(found.is_some())
    }

    fn access_token_exists_by_name(
        &self,
        tx: &Transaction<'_>,
        user_id: i64,
        name: &str,
    ) -> AppResult<bool> {
        let found = tx
            .query_row(
                "SELECT 1 FROM access_token WHERE user_id = ?1 AND name = ?2 LIMIT 1",
                params![user_id, name],
                |row| row.get::<_, i64>(0),
            )
            .optional()?;
        Ok(found.is_some())
    }

    fn lock(&self) -> AppResult<std::sync::MutexGuard<'_, Connection>> {
        self.conn
            .lock()
            .map_err(|_| AppError::Db(rusqlite::Error::InvalidQuery))
    }
}

pub struct NewUser<'a> {
    pub username: &'a str,
    pub full_name: &'a str,
    pub email: &'a str,
    pub password_hash: &'a str,
    pub is_active: bool,
    pub is_admin: bool,
}

pub struct NewRepository<'a> {
    pub owner_id: i64,
    pub owner_name: &'a str,
    pub name: &'a str,
    pub description: &'a str,
    pub default_branch: &'a str,
    pub is_private: bool,
    pub is_bare: bool,
    pub is_fork: bool,
    pub fork_id: i64,
}

pub struct NewPullRequest<'a> {
    pub title: &'a str,
    pub body: &'a str,
    pub status: PullRequestStatus,
    pub head_repo_id: i64,
    pub base_repo_id: i64,
    pub head_user_name: &'a str,
    pub head_branch: &'a str,
    pub base_branch: &'a str,
    pub merge_base: &'a str,
    pub poster_id: i64,
}

fn row_to_user(row: &rusqlite::Row<'_>) -> rusqlite::Result<User> {
    Ok(User {
        id: row.get(0)?,
        lower_name: row.get(1)?,
        name: row.get(2)?,
        full_name: row.get(3)?,
        email: row.get(4)?,
        password_hash: row.get(5)?,
        is_active: row.get(6)?,
        is_admin: row.get(7)?,
        created_unix: row.get(8)?,
        updated_unix: row.get(9)?,
    })
}

fn row_to_user_at(row: &rusqlite::Row<'_>, offset: usize) -> rusqlite::Result<User> {
    Ok(User {
        id: row.get(offset)?,
        lower_name: row.get(offset + 1)?,
        name: row.get(offset + 2)?,
        full_name: row.get(offset + 3)?,
        email: row.get(offset + 4)?,
        password_hash: row.get(offset + 5)?,
        is_active: row.get(offset + 6)?,
        is_admin: row.get(offset + 7)?,
        created_unix: row.get(offset + 8)?,
        updated_unix: row.get(offset + 9)?,
    })
}

fn row_to_repo(row: &rusqlite::Row<'_>) -> rusqlite::Result<Repository> {
    Ok(Repository {
        id: row.get(0)?,
        owner_id: row.get(1)?,
        lower_name: row.get(2)?,
        name: row.get(3)?,
        description: row.get(4)?,
        default_branch: row.get(5)?,
        is_private: row.get(6)?,
        is_bare: row.get(7)?,
        is_fork: row.get(8)?,
        fork_id: row.get(9)?,
        created_unix: row.get(10)?,
        updated_unix: row.get(11)?,
    })
}

fn row_to_repo_at(row: &rusqlite::Row<'_>, offset: usize) -> rusqlite::Result<Repository> {
    Ok(Repository {
        id: row.get(offset)?,
        owner_id: row.get(offset + 1)?,
        lower_name: row.get(offset + 2)?,
        name: row.get(offset + 3)?,
        description: row.get(offset + 4)?,
        default_branch: row.get(offset + 5)?,
        is_private: row.get(offset + 6)?,
        is_bare: row.get(offset + 7)?,
        is_fork: row.get(offset + 8)?,
        fork_id: row.get(offset + 9)?,
        created_unix: row.get(offset + 10)?,
        updated_unix: row.get(offset + 11)?,
    })
}

fn row_to_repository_with_owner(row: &rusqlite::Row<'_>) -> rusqlite::Result<RepositoryWithOwner> {
    Ok(RepositoryWithOwner {
        repo: row_to_repo_at(row, 0)?,
        owner: row_to_user_at(row, 12)?,
    })
}

fn row_to_access_token(row: &rusqlite::Row<'_>) -> rusqlite::Result<AccessToken> {
    Ok(AccessToken {
        id: row.get(0)?,
        user_id: row.get(1)?,
        name: row.get(2)?,
        token_hash: row.get(3)?,
        created_unix: row.get(4)?,
        updated_unix: row.get(5)?,
    })
}

fn row_to_collaborator_response(
    row: &rusqlite::Row<'_>,
) -> rusqlite::Result<CollaboratorResponse> {
    Ok(CollaboratorResponse {
        user: row_to_user_at(row, 0)?,
        mode: access_mode_from_i64(row.get::<_, i64>(10)?),
    })
}

fn row_to_pull_request(row: &rusqlite::Row<'_>) -> rusqlite::Result<PullRequest> {
    Ok(PullRequest {
        id: row.get(0)?,
        index: row.get(1)?,
        title: row.get(2)?,
        body: row.get(3)?,
        status: pull_request_status_from_i64(row.get::<_, i64>(4)?),
        head_repo_id: row.get(5)?,
        base_repo_id: row.get(6)?,
        head_user_name: row.get(7)?,
        head_branch: row.get(8)?,
        base_branch: row.get(9)?,
        merge_base: row.get(10)?,
        merged_commit_id: row.get(11)?,
        poster_id: row.get(12)?,
        has_merged: row.get(13)?,
        is_closed: row.get(14)?,
        created_unix: row.get(15)?,
        updated_unix: row.get(16)?,
    })
}

fn row_to_collaboration(row: &rusqlite::Row<'_>) -> rusqlite::Result<Collaboration> {
    Ok(Collaboration {
        id: row.get(0)?,
        user_id: row.get(1)?,
        repo_id: row.get(2)?,
        mode: access_mode_from_i64(row.get::<_, i64>(3)?),
    })
}

fn access_mode_from_i64(value: i64) -> AccessMode {
    match value {
        1 => AccessMode::Read,
        2 => AccessMode::Write,
        3 => AccessMode::Admin,
        4 => AccessMode::Owner,
        _ => AccessMode::None,
    }
}

fn pull_request_status_from_i64(value: i64) -> PullRequestStatus {
    match value {
        0 => PullRequestStatus::Conflict,
        1 => PullRequestStatus::Checking,
        _ => PullRequestStatus::Mergeable,
    }
}

fn now_unix() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs() as i64
}

fn ensure_column_exists(
    conn: &Connection,
    table_name: &str,
    column_name: &str,
    column_sql: &str,
) -> AppResult<()> {
    let mut stmt = conn.prepare(&format!("PRAGMA table_info({table_name})"))?;
    let rows = stmt.query_map([], |row| row.get::<_, String>(1))?;
    for row in rows {
        if row? == column_name {
            return Ok(());
        }
    }
    conn.execute(
        &format!("ALTER TABLE {table_name} ADD COLUMN {column_name} {column_sql}"),
        [],
    )?;
    Ok(())
}