// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

//! Groups: the nested, user-owned repository hierarchy, with per-group settings
//! (visibility, collaborators) and recursive garbage collection of the auto-
//! created groups that once held a now-deleted repository.

use model::{Group, Visibility};
use sqlx::any::AnyRow;
use sqlx::{AssertSqlSafe, Row};

use crate::repos::Collaborator;
use crate::{Store, StoreError, map_conflict, new_id, now_ms};

/// The columns of `groups` in a fixed order, shared by every `SELECT`.
const GROUP_COLUMNS: &str =
    "id, owner_id, parent_id, name, path, description, manual, visibility, created_at";

/// Rank a permission token so the highest across repo + group collaborators wins.
fn perm_rank(perm: &str) -> i32 {
    match perm {
        "admin" => 3,
        "write" => 2,
        "read" => 1,
        _ => 0,
    }
}

impl Store {
    /// Ensure a group exists at `owner_id`/`path`, creating it and every missing
    /// intermediate group (all auto), and return the leaf group.
    ///
    /// # Errors
    ///
    /// * [`StoreError::Name`] if any segment is an invalid name.
    /// * [`StoreError::GroupTooDeep`] past [`model::MAX_GROUP_DEPTH`].
    /// * [`StoreError::Query`] on a database failure.
    pub async fn ensure_group_path(&self, owner_id: &str, path: &str) -> Result<Group, StoreError> {
        let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
        if segments.is_empty() {
            return Err(StoreError::Name(model::NameError::Empty));
        }
        if segments.len() > model::MAX_GROUP_DEPTH {
            return Err(StoreError::GroupTooDeep {
                max: model::MAX_GROUP_DEPTH,
            });
        }
        for segment in &segments {
            model::validate_name(segment)?;
        }

        let mut parent_id: Option<String> = None;
        let mut prefix = String::new();
        let mut leaf: Option<Group> = None;
        for segment in segments {
            if prefix.is_empty() {
                prefix.push_str(segment);
            } else {
                prefix.push('/');
                prefix.push_str(segment);
            }
            let group = match self.group_by_owner_path(owner_id, &prefix).await? {
                Some(existing) => existing,
                None => {
                    self.insert_group(owner_id, parent_id.as_deref(), segment, &prefix)
                        .await?
                }
            };
            parent_id = Some(group.id.clone());
            leaf = Some(group);
        }
        leaf.ok_or(StoreError::Name(model::NameError::Empty))
    }

    /// Create a group explicitly at `path`, marking its leaf as manual so it is
    /// never garbage-collected. Intermediates are created auto.
    ///
    /// # Errors
    ///
    /// See [`Store::ensure_group_path`].
    pub async fn create_group(&self, owner_id: &str, path: &str) -> Result<Group, StoreError> {
        let leaf = self.ensure_group_path(owner_id, path).await?;
        sqlx::query("UPDATE groups SET manual = $1 WHERE id = $2")
            .bind(true)
            .bind(&leaf.id)
            .execute(&self.pool)
            .await?;
        self.group_by_id(&leaf.id)
            .await?
            .ok_or(StoreError::Name(model::NameError::Empty))
    }

    /// Insert a single (auto) group row.
    async fn insert_group(
        &self,
        owner_id: &str,
        parent_id: Option<&str>,
        name: &str,
        path: &str,
    ) -> Result<Group, StoreError> {
        let group = Group {
            id: new_id(),
            owner_id: owner_id.to_string(),
            parent_id: parent_id.map(str::to_string),
            name: name.to_string(),
            path: path.to_string(),
            description: None,
            manual: false,
            // Auto groups are transparent to the visibility ceiling (see migration
            // 0021); only a deliberately-set group restricts the repos beneath it.
            visibility: Visibility::Public,
            created_at: now_ms(),
        };
        let sql = "INSERT INTO groups \
             (id, owner_id, parent_id, name, path, description, manual, visibility, created_at) \
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)";
        sqlx::query(sql)
            .bind(group.id.clone())
            .bind(group.owner_id.clone())
            .bind(group.parent_id.clone())
            .bind(group.name.clone())
            .bind(group.path.clone())
            .bind(group.description.clone())
            .bind(group.manual)
            .bind(group.visibility.as_str())
            .bind(group.created_at)
            .execute(&self.pool)
            .await
            .map_err(|e| map_conflict(e, "group", &[("path", "path")]))?;
        Ok(group)
    }

    /// Resolve a group by `(owner_id, path)`, or `None`.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn group_by_owner_path(
        &self,
        owner_id: &str,
        path: &str,
    ) -> Result<Option<Group>, StoreError> {
        let sql = format!("SELECT {GROUP_COLUMNS} FROM groups WHERE owner_id = $1 AND path = $2");
        let row = sqlx::query(AssertSqlSafe(sql))
            .bind(owner_id)
            .bind(path)
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.as_ref().map(|r| self.map_group(r)).transpose()?)
    }

    /// Fetch a group by id.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn group_by_id(&self, id: &str) -> Result<Option<Group>, StoreError> {
        let sql = format!("SELECT {GROUP_COLUMNS} FROM groups WHERE id = $1");
        let row = sqlx::query(AssertSqlSafe(sql))
            .bind(id)
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.as_ref().map(|r| self.map_group(r)).transpose()?)
    }

    /// List every group owned by `owner_id`, ordered by path.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn groups_by_owner(&self, owner_id: &str) -> Result<Vec<Group>, StoreError> {
        let sql =
            format!("SELECT {GROUP_COLUMNS} FROM groups WHERE owner_id = $1 ORDER BY path ASC");
        let rows = sqlx::query(AssertSqlSafe(sql))
            .bind(owner_id)
            .fetch_all(&self.pool)
            .await?;
        rows.iter()
            .map(|r| self.map_group(r))
            .collect::<Result<_, _>>()
            .map_err(Into::into)
    }

    /// List every group across all owners, ordered by path. For the admin view.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn list_groups(&self) -> Result<Vec<Group>, StoreError> {
        let sql = format!("SELECT {GROUP_COLUMNS} FROM groups ORDER BY owner_id ASC, path ASC");
        let rows = sqlx::query(AssertSqlSafe(sql))
            .fetch_all(&self.pool)
            .await?;
        rows.iter()
            .map(|r| self.map_group(r))
            .collect::<Result<_, _>>()
            .map_err(Into::into)
    }

    /// One page of groups, ordered by owner then path. For the admin view.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn list_groups_paged(
        &self,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<Group>, StoreError> {
        let sql = format!(
            "SELECT {GROUP_COLUMNS} FROM groups ORDER BY owner_id ASC, path ASC LIMIT $1 OFFSET $2"
        );
        let rows = sqlx::query(AssertSqlSafe(sql))
            .bind(limit)
            .bind(offset)
            .fetch_all(&self.pool)
            .await?;
        rows.iter()
            .map(|r| self.map_group(r))
            .collect::<Result<_, _>>()
            .map_err(Into::into)
    }

    /// Set a group's visibility.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn set_group_visibility(
        &self,
        group_id: &str,
        visibility: Visibility,
    ) -> Result<(), StoreError> {
        sqlx::query("UPDATE groups SET visibility = $1 WHERE id = $2")
            .bind(visibility.as_str())
            .bind(group_id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// Rename a group's leaf name, updating the materialized paths of the group,
    /// its descendant groups, and every repository beneath it.
    ///
    /// # Errors
    ///
    /// * [`StoreError::Name`] if `new_name` is invalid.
    /// * [`StoreError::Conflict`] if the new path is already taken.
    /// * [`StoreError::Query`] for any other database failure.
    pub async fn rename_group(&self, group_id: &str, new_name: &str) -> Result<Group, StoreError> {
        model::validate_name(new_name)?;
        let group = self
            .group_by_id(group_id)
            .await?
            .ok_or(StoreError::Name(model::NameError::Empty))?;
        let old_path = group.path.clone();
        let new_path = match old_path.rsplit_once('/') {
            Some((prefix, _)) => format!("{prefix}/{new_name}"),
            None => new_name.to_string(),
        };
        if new_path != old_path
            && (self
                .group_by_owner_path(&group.owner_id, &new_path)
                .await?
                .is_some()
                || self
                    .repo_by_owner_path(&group.owner_id, &new_path)
                    .await?
                    .is_some())
        {
            return Err(StoreError::Conflict {
                entity: "group",
                field: "path",
            });
        }
        let old_prefix = format!("{old_path}/%");
        let old_len = i64::try_from(old_path.len()).unwrap_or(i64::MAX);
        let mut tx = self.pool.begin().await?;
        // Descendant groups: replace the shared path prefix.
        sqlx::query(
            "UPDATE groups SET path = $1 || SUBSTR(path, $2) \
             WHERE owner_id = $3 AND path LIKE $4",
        )
        .bind(&new_path)
        .bind(old_len + 1)
        .bind(&group.owner_id)
        .bind(&old_prefix)
        .execute(&mut *tx)
        .await?;
        // Repositories beneath the group.
        sqlx::query(
            "UPDATE repos SET path = $1 || SUBSTR(path, $2) \
             WHERE owner_id = $3 AND path LIKE $4",
        )
        .bind(&new_path)
        .bind(old_len + 1)
        .bind(&group.owner_id)
        .bind(&old_prefix)
        .execute(&mut *tx)
        .await?;
        // The group itself.
        sqlx::query("UPDATE groups SET name = $1, path = $2 WHERE id = $3")
            .bind(new_name)
            .bind(&new_path)
            .bind(group_id)
            .execute(&mut *tx)
            .await?;
        tx.commit().await?;
        self.group_by_id(group_id)
            .await?
            .ok_or(StoreError::Name(model::NameError::Empty))
    }

    /// Whether a group holds any repositories directly.
    async fn group_has_repos(&self, group_id: &str) -> Result<bool, StoreError> {
        let row = sqlx::query("SELECT 1 AS n FROM repos WHERE group_id = $1")
            .bind(group_id)
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.is_some())
    }

    /// Whether a group has any child groups.
    async fn group_has_children(&self, group_id: &str) -> Result<bool, StoreError> {
        let row = sqlx::query("SELECT 1 AS n FROM groups WHERE parent_id = $1")
            .bind(group_id)
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.is_some())
    }

    /// The number of repositories anywhere in a group's subtree.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn group_subtree_repo_count(&self, group: &Group) -> Result<i64, StoreError> {
        let prefix = format!("{}/%", group.path);
        let n: i64 =
            sqlx::query("SELECT COUNT(*) AS n FROM repos WHERE owner_id = $1 AND path LIKE $2")
                .bind(&group.owner_id)
                .bind(&prefix)
                .fetch_one(&self.pool)
                .await?
                .try_get("n")?;
        Ok(n)
    }

    /// Garbage-collect empty auto groups upward from `group_id`: while a group is
    /// auto (not manual) and holds no repos and no child groups, delete it and
    /// continue with its parent. Manual groups (and non-empty ones) stop the walk.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn gc_empty_groups(&self, mut group_id: Option<String>) -> Result<(), StoreError> {
        let mut guard = 0;
        while let Some(gid) = group_id {
            guard += 1;
            if guard > 64 {
                break;
            }
            let Some(group) = self.group_by_id(&gid).await? else {
                break;
            };
            if group.manual
                || self.group_has_repos(&gid).await?
                || self.group_has_children(&gid).await?
            {
                break;
            }
            let parent = group.parent_id.clone();
            self.delete_group(&gid).await?;
            group_id = parent;
        }
        Ok(())
    }

    /// Delete a group by id. Child groups cascade; repos have their `group_id`
    /// set null by the schema. Returns `true` if a row was deleted.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn delete_group(&self, group_id: &str) -> Result<bool, StoreError> {
        let deleted = sqlx::query("DELETE FROM groups WHERE id = $1")
            .bind(group_id)
            .execute(&self.pool)
            .await?
            .rows_affected();
        Ok(deleted > 0)
    }

    // ---- Group collaborators ----

    /// Add or update a group collaborator's permission.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn add_group_collaborator(
        &self,
        group_id: &str,
        user_id: &str,
        permission: &str,
    ) -> Result<(), StoreError> {
        sqlx::query(
            "INSERT INTO group_collaborators (id, group_id, user_id, permission, created_at) \
             VALUES ($1, $2, $3, $4, $5) \
             ON CONFLICT (group_id, user_id) DO UPDATE SET permission = $4",
        )
        .bind(new_id())
        .bind(group_id)
        .bind(user_id)
        .bind(permission)
        .bind(now_ms())
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Remove a group collaborator.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn remove_group_collaborator(
        &self,
        group_id: &str,
        user_id: &str,
    ) -> Result<(), StoreError> {
        sqlx::query("DELETE FROM group_collaborators WHERE group_id = $1 AND user_id = $2")
            .bind(group_id)
            .bind(user_id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// List a group's collaborators with their usernames resolved.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn list_group_collaborators(
        &self,
        group_id: &str,
    ) -> Result<Vec<Collaborator>, StoreError> {
        let rows = sqlx::query(
            "SELECT gc.user_id, u.username, gc.permission \
             FROM group_collaborators gc JOIN users u ON u.id = gc.user_id \
             WHERE gc.group_id = $1 ORDER BY u.username_lower ASC",
        )
        .bind(group_id)
        .fetch_all(&self.pool)
        .await?;
        rows.iter()
            .map(|r| {
                Ok(Collaborator {
                    user_id: r.try_get("user_id")?,
                    username: r.try_get("username")?,
                    permission: r.try_get("permission")?,
                })
            })
            .collect::<Result<_, sqlx::Error>>()
            .map_err(Into::into)
    }

    /// A user's collaborator permission on a group, if any.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn group_collaborator_permission(
        &self,
        group_id: &str,
        user_id: &str,
    ) -> Result<Option<String>, StoreError> {
        let row = sqlx::query(
            "SELECT permission FROM group_collaborators WHERE group_id = $1 AND user_id = $2",
        )
        .bind(group_id)
        .bind(user_id)
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(|r| r.try_get("permission")).transpose()?)
    }

    /// A user's effective collaborator permission on a repo: the strongest of its
    /// direct collaborator row and any group-collaborator row on the repo's group
    /// or an ancestor group.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn effective_permission(
        &self,
        repo_id: &str,
        user_id: &str,
    ) -> Result<Option<String>, StoreError> {
        let mut best: Option<String> = self.collaborator_permission(repo_id, user_id).await?;
        let mut best_rank = best.as_deref().map_or(0, perm_rank);

        let mut group_id = self.repo_by_id(repo_id).await?.and_then(|r| r.group_id);
        let mut guard = 0;
        while let Some(gid) = group_id {
            guard += 1;
            if guard > 64 {
                break;
            }
            if let Some(p) = self.group_collaborator_permission(&gid, user_id).await? {
                let r = perm_rank(&p);
                if r > best_rank {
                    best_rank = r;
                    best = Some(p);
                }
            }
            group_id = self.group_by_id(&gid).await?.and_then(|g| g.parent_id);
        }
        Ok(best)
    }

    /// A user's effective collaborator permission on a group: the strongest
    /// group-collaborator row on the group itself or any ancestor group.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn effective_group_permission(
        &self,
        group_id: &str,
        user_id: &str,
    ) -> Result<Option<String>, StoreError> {
        let mut best: Option<String> = None;
        let mut best_rank = 0;
        let mut next = Some(group_id.to_string());
        let mut guard = 0;
        while let Some(gid) = next {
            guard += 1;
            if guard > 64 {
                break;
            }
            if let Some(p) = self.group_collaborator_permission(&gid, user_id).await? {
                let r = perm_rank(&p);
                if r > best_rank {
                    best_rank = r;
                    best = Some(p);
                }
            }
            next = self.group_by_id(&gid).await?.and_then(|g| g.parent_id);
        }
        Ok(best)
    }

    /// The most-restrictive visibility across a group and its ancestors — the
    /// ceiling a repository in the group (or a descendant group) may not exceed.
    /// Auto groups default to public, so they leave the ceiling untouched.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn group_visibility_ceiling(&self, group_id: &str) -> Result<Visibility, StoreError> {
        let mut ceiling = Visibility::Public;
        let mut next = Some(group_id.to_string());
        let mut guard = 0;
        while let Some(gid) = next {
            guard += 1;
            if guard > 64 {
                break;
            }
            let Some(g) = self.group_by_id(&gid).await? else {
                break;
            };
            if g.visibility.rank() < ceiling.rank() {
                ceiling = g.visibility;
            }
            next = g.parent_id;
        }
        Ok(ceiling)
    }

    /// Clamp every descendant group and repository so none is more visible than
    /// its group ceiling. Call after lowering a group's visibility.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::Query`] on a database failure.
    pub async fn clamp_subtree_visibility(&self, group: &Group) -> Result<(), StoreError> {
        let child_prefix = format!("{}/", group.path);
        // Descendant groups first, top-down (path ASC) so each parent's ceiling is
        // settled before its children are measured against it.
        let groups = self.groups_by_owner(&group.owner_id).await?;
        for g in groups.iter().filter(|g| g.path.starts_with(&child_prefix)) {
            if let Some(parent_id) = &g.parent_id {
                let ceiling = self.group_visibility_ceiling(parent_id).await?;
                if g.visibility.rank() > ceiling.rank() {
                    self.set_group_visibility(&g.id, ceiling).await?;
                }
            }
        }
        // Then every repository anywhere in the subtree.
        let repos = self.repos_by_owner(&group.owner_id).await?;
        for r in repos.iter().filter(|r| r.path.starts_with(&child_prefix)) {
            if let Some(gid) = &r.group_id {
                let ceiling = self.group_visibility_ceiling(gid).await?;
                if r.visibility.rank() > ceiling.rank() {
                    self.set_visibility(&r.id, ceiling).await?;
                }
            }
        }
        Ok(())
    }

    /// Map a `groups` row (selected via [`GROUP_COLUMNS`]) to a [`Group`].
    fn map_group(&self, row: &AnyRow) -> Result<Group, sqlx::Error> {
        Ok(Group {
            id: row.try_get("id")?,
            owner_id: row.try_get("owner_id")?,
            parent_id: row.try_get("parent_id")?,
            name: row.try_get("name")?,
            path: row.try_get("path")?,
            description: row.try_get("description")?,
            manual: self.get_bool(row, "manual")?,
            visibility: Visibility::from_token(&row.try_get::<String, _>("visibility")?)
                .unwrap_or(Visibility::Private),
            created_at: row.try_get("created_at")?,
        })
    }
}