//! CRUD over the `groups` tree (Phase 2).
//!
//! Groups form a single-parent org tree above apps. Structural mutations
//! (reparent/rename/delete) must keep the tree acyclic and non-orphaning:
//!
//! - **delete = RESTRICT** — refused if the group has child groups or apps
//!   (the DB FKs enforce this; we surface a clean conflict).
//! - **slug-freeze** — `rename` edits name/description only; the slug is
//!   set once at creation and never rewritten.
//! - **cycle guard** — `reparent` walks the destination's ancestors under a
//!   coarse instance-wide advisory lock and refuses a move that would make
//!   a node its own ancestor. A SQL `CHECK` can't express this.
//! - **structure_version** — bumped on every structural mutation so a
//!   future CLI/orchestrator can detect structural drift (§6).

use async_trait::async_trait;
use picloud_shared::{AppId, Group, GroupId};
use sqlx::PgPool;
use uuid::Uuid;

/// Instance-wide advisory-lock key for structural group mutations. Coarse
/// on purpose: reparent/rename/delete all take it so the ancestor-walk
/// cycle guard and the `parent_id` write run serialized — two concurrent
/// reparents can't race into a cycle. Distinct from the per-app
/// `apply_lock_key` space (a fixed sentinel, hashed-namespace-free).
const GROUP_STRUCTURAL_LOCK_KEY: i64 = 0x6701_0047_0000_0001;

/// Well-known slug of the instance root group seeded by migration 0047.
pub const ROOT_GROUP_SLUG: &str = "root";

#[derive(Debug, thiserror::Error)]
pub enum GroupRepositoryError {
    #[error("database error: {0}")]
    Db(#[from] sqlx::Error),
    #[error("not found: {0}")]
    NotFound(GroupId),
    #[error("conflict: {0}")]
    Conflict(String),
}

/// Counts of a group's direct children — used to enforce delete=RESTRICT
/// with an actionable message and to render the tree.
#[derive(Debug, Clone, Copy)]
pub struct GroupChildCounts {
    pub subgroups: i64,
    pub apps: i64,
}

impl GroupChildCounts {
    #[must_use]
    pub fn is_empty(self) -> bool {
        self.subgroups == 0 && self.apps == 0
    }
}

#[async_trait]
pub trait GroupRepository: Send + Sync {
    /// Every group on the instance, ordered by name. The tree is small
    /// (org structure), so callers assemble the hierarchy in memory.
    async fn list(&self) -> Result<Vec<Group>, GroupRepositoryError>;
    async fn get_by_id(&self, id: GroupId) -> Result<Option<Group>, GroupRepositoryError>;
    async fn get_by_slug(&self, slug: &str) -> Result<Option<Group>, GroupRepositoryError>;
    /// Direct children groups of `parent`.
    async fn list_children(&self, parent: GroupId) -> Result<Vec<Group>, GroupRepositoryError>;
    /// The node plus its ancestors up to the root, nearest-first. Used for
    /// path display and as the reparent cycle-guard input.
    async fn ancestors(&self, id: GroupId) -> Result<Vec<Group>, GroupRepositoryError>;
    async fn child_counts(&self, id: GroupId) -> Result<GroupChildCounts, GroupRepositoryError>;
    async fn create(
        &self,
        slug: &str,
        name: &str,
        description: Option<&str>,
        parent_id: Option<GroupId>,
    ) -> Result<Group, GroupRepositoryError>;
    /// Edit display fields only — the slug is frozen at creation. Bumps
    /// `structure_version`.
    async fn rename(
        &self,
        id: GroupId,
        name: Option<&str>,
        description: Option<Option<&str>>,
    ) -> Result<Group, GroupRepositoryError>;
    /// Move `id` under `new_parent` (or to root if `None`). Runs the
    /// ancestor-walk cycle guard under a coarse structural lock and bumps
    /// `structure_version`. Refuses a move that would create a cycle.
    async fn reparent(
        &self,
        id: GroupId,
        new_parent: Option<GroupId>,
    ) -> Result<Group, GroupRepositoryError>;
    /// Delete an empty group (delete = RESTRICT). Refused with a clean
    /// conflict if it still has child groups or apps.
    async fn delete(&self, id: GroupId) -> Result<(), GroupRepositoryError>;
}

pub struct PostgresGroupRepository {
    pool: PgPool,
}

impl PostgresGroupRepository {
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

const GROUP_COLS: &str =
    "id, parent_id, slug, name, description, structure_version, created_at, updated_at";

#[async_trait]
impl GroupRepository for PostgresGroupRepository {
    async fn list(&self) -> Result<Vec<Group>, GroupRepositoryError> {
        let rows = sqlx::query_as::<_, GroupRow>(&format!(
            "SELECT {GROUP_COLS} FROM groups ORDER BY name"
        ))
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(Into::into).collect())
    }

    async fn get_by_id(&self, id: GroupId) -> Result<Option<Group>, GroupRepositoryError> {
        let row = sqlx::query_as::<_, GroupRow>(&format!(
            "SELECT {GROUP_COLS} FROM groups WHERE id = $1"
        ))
        .bind(id.into_inner())
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(Into::into))
    }

    async fn get_by_slug(&self, slug: &str) -> Result<Option<Group>, GroupRepositoryError> {
        let row = sqlx::query_as::<_, GroupRow>(&format!(
            "SELECT {GROUP_COLS} FROM groups WHERE slug = $1"
        ))
        .bind(slug)
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(Into::into))
    }

    async fn list_children(&self, parent: GroupId) -> Result<Vec<Group>, GroupRepositoryError> {
        let rows = sqlx::query_as::<_, GroupRow>(&format!(
            "SELECT {GROUP_COLS} FROM groups WHERE parent_id = $1 ORDER BY name"
        ))
        .bind(parent.into_inner())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(Into::into).collect())
    }

    async fn ancestors(&self, id: GroupId) -> Result<Vec<Group>, GroupRepositoryError> {
        // Recursive walk node → root, nearest-first. Depth-bounded as a
        // runaway guard (the cycle guard already prevents cycles).
        let rows = sqlx::query_as::<_, GroupRow>(&format!(
            "WITH RECURSIVE chain AS (
                 SELECT {GROUP_COLS}, 0 AS depth FROM groups WHERE id = $1
               UNION ALL
                 SELECT g.id, g.parent_id, g.slug, g.name, g.description, \
                        g.structure_version, g.created_at, g.updated_at, c.depth + 1 \
                 FROM groups g JOIN chain c ON g.id = c.parent_id \
                 WHERE c.depth < 64
             )
             SELECT {GROUP_COLS} FROM chain ORDER BY depth"
        ))
        .bind(id.into_inner())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(Into::into).collect())
    }

    async fn child_counts(&self, id: GroupId) -> Result<GroupChildCounts, GroupRepositoryError> {
        let row: (i64, i64) = sqlx::query_as(
            "SELECT \
                 (SELECT COUNT(*) FROM groups WHERE parent_id = $1), \
                 (SELECT COUNT(*) FROM apps WHERE group_id = $1)",
        )
        .bind(id.into_inner())
        .fetch_one(&self.pool)
        .await?;
        Ok(GroupChildCounts {
            subgroups: row.0,
            apps: row.1,
        })
    }

    async fn create(
        &self,
        slug: &str,
        name: &str,
        description: Option<&str>,
        parent_id: Option<GroupId>,
    ) -> Result<Group, GroupRepositoryError> {
        let mut tx = self.pool.begin().await?;
        // Interactive create: UI/API-owned (no project claim — §7.5).
        let g = create_group_tx(&mut tx, slug, name, description, parent_id, None).await?;
        tx.commit().await?;
        Ok(g)
    }

    async fn rename(
        &self,
        id: GroupId,
        name: Option<&str>,
        description: Option<Option<&str>>,
    ) -> Result<Group, GroupRepositoryError> {
        // Slug is intentionally absent from the SET list — it is frozen.
        let row = sqlx::query_as::<_, GroupRow>(&format!(
            "UPDATE groups SET \
                 name = COALESCE($2, name), \
                 description = CASE WHEN $3::bool THEN $4 ELSE description END, \
                 structure_version = structure_version + 1, \
                 updated_at = NOW() \
             WHERE id = $1 \
             RETURNING {GROUP_COLS}"
        ))
        .bind(id.into_inner())
        .bind(name)
        .bind(description.is_some())
        .bind(description.and_then(|d| d))
        .fetch_optional(&self.pool)
        .await?;
        row.map(Into::into)
            .ok_or(GroupRepositoryError::NotFound(id))
    }

    async fn reparent(
        &self,
        id: GroupId,
        new_parent: Option<GroupId>,
    ) -> Result<Group, GroupRepositoryError> {
        let mut tx = self.pool.begin().await?;
        // Coarse structural lock: serialize all structural mutations so the
        // cycle guard + parent write can't race a concurrent reparent.
        acquire_structural_lock_tx(&mut tx).await?;
        let g = reparent_group_tx(&mut tx, id, new_parent).await?;
        tx.commit().await?;
        Ok(g)
    }

    async fn delete(&self, id: GroupId) -> Result<(), GroupRepositoryError> {
        let mut tx = self.pool.begin().await?;
        delete_group_tx(&mut tx, id).await?;
        tx.commit().await?;
        Ok(())
    }
}

// ----------------------------------------------------------------------------
// Transaction-aware structural mutations (Phase 5+ declarative tree apply).
//
// The declarative `apply_tree` reconciles a whole subtree in ONE transaction,
// so group create/reparent/delete must run inside the caller's tx (not on the
// pool) to stay all-or-nothing. These free functions hold the SQL; the trait
// methods above delegate to them (begin → fn → commit) so there is one SQL
// definition each. The caller takes [`acquire_structural_lock_tx`] ONCE before
// the structure phase, so the cycle guard + parent writes serialize against
// concurrent reparents exactly as the single-shot `reparent` does.
// ----------------------------------------------------------------------------

/// Take the coarse structural lock inside the caller's transaction. Held until
/// the tx commits/rolls back. Call once before any `*_group_tx` mutation.
pub(crate) async fn acquire_structural_lock_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<(), GroupRepositoryError> {
    sqlx::query("SELECT pg_advisory_xact_lock($1)")
        .bind(GROUP_STRUCTURAL_LOCK_KEY)
        .execute(&mut **tx)
        .await?;
    Ok(())
}

/// Insert a group inside the caller's tx. Maps unique/FK violations to a clean
/// conflict. (The structural lock is not required for a pure insert, but the
/// apply path holds it for the whole phase anyway.)
pub(crate) async fn create_group_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    slug: &str,
    name: &str,
    description: Option<&str>,
    parent_id: Option<GroupId>,
    owner_project: Option<Uuid>,
) -> Result<Group, GroupRepositoryError> {
    let res = sqlx::query_as::<_, GroupRow>(&format!(
        "INSERT INTO groups (slug, name, description, parent_id, owner_project) \
         VALUES ($1, $2, $3, $4, $5) RETURNING {GROUP_COLS}"
    ))
    .bind(slug)
    .bind(name)
    .bind(description)
    .bind(parent_id.map(GroupId::into_inner))
    .bind(owner_project)
    .fetch_one(&mut **tx)
    .await;
    match res {
        Ok(row) => Ok(row.into()),
        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => Err(
            GroupRepositoryError::Conflict(format!("slug {slug:?} is already in use")),
        ),
        Err(sqlx::Error::Database(e)) if e.is_foreign_key_violation() => Err(
            GroupRepositoryError::Conflict("parent group does not exist".into()),
        ),
        Err(e) => Err(e.into()),
    }
}

/// Reparent a group inside the caller's tx. Runs the ancestor-walk cycle guard
/// against the IN-TX tree (so it sees parent writes made earlier in the same
/// apply). The caller must already hold [`acquire_structural_lock_tx`].
pub(crate) async fn reparent_group_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    id: GroupId,
    new_parent: Option<GroupId>,
) -> Result<Group, GroupRepositoryError> {
    if let Some(parent) = new_parent {
        if parent == id {
            return Err(GroupRepositoryError::Conflict(
                "a group cannot be its own parent".into(),
            ));
        }
        let mut cursor = Some(parent);
        let mut hops = 0u32;
        while let Some(node) = cursor {
            if node == id {
                return Err(GroupRepositoryError::Conflict(
                    "cannot reparent a group beneath one of its own descendants".into(),
                ));
            }
            hops += 1;
            if hops > 64 {
                return Err(GroupRepositoryError::Conflict(
                    "group ancestry exceeds the maximum depth".into(),
                ));
            }
            let parent_of: Option<(Option<Uuid>,)> =
                sqlx::query_as("SELECT parent_id FROM groups WHERE id = $1")
                    .bind(node.into_inner())
                    .fetch_optional(&mut **tx)
                    .await?;
            match parent_of {
                Some((p,)) => cursor = p.map(GroupId::from),
                None => {
                    return Err(GroupRepositoryError::Conflict(
                        "destination parent group does not exist".into(),
                    ));
                }
            }
        }
    }
    let row = sqlx::query_as::<_, GroupRow>(&format!(
        "UPDATE groups SET parent_id = $2, structure_version = structure_version + 1, \
             updated_at = NOW() WHERE id = $1 RETURNING {GROUP_COLS}"
    ))
    .bind(id.into_inner())
    .bind(new_parent.map(GroupId::into_inner))
    .fetch_optional(&mut **tx)
    .await?;
    row.map(Into::into)
        .ok_or(GroupRepositoryError::NotFound(id))
}

/// Delete an empty group inside the caller's tx (delete = RESTRICT). Refused
/// with a clean conflict if it still has child groups or apps.
pub(crate) async fn delete_group_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    id: GroupId,
) -> Result<(), GroupRepositoryError> {
    let counts: (i64, i64) = sqlx::query_as(
        "SELECT (SELECT COUNT(*) FROM groups WHERE parent_id = $1), \
                (SELECT COUNT(*) FROM apps WHERE group_id = $1)",
    )
    .bind(id.into_inner())
    .fetch_one(&mut **tx)
    .await?;
    if counts.0 != 0 || counts.1 != 0 {
        return Err(GroupRepositoryError::Conflict(format!(
            "group still has {} subgroup(s) and {} app(s); move or delete them first",
            counts.0, counts.1
        )));
    }
    let res = sqlx::query("DELETE FROM groups WHERE id = $1")
        .bind(id.into_inner())
        .execute(&mut **tx)
        .await;
    match res {
        Ok(r) if r.rows_affected() == 0 => Err(GroupRepositoryError::NotFound(id)),
        Ok(_) => Ok(()),
        Err(sqlx::Error::Database(e)) if e.is_foreign_key_violation() => {
            Err(GroupRepositoryError::Conflict(
                "group still has descendants; move or delete them first".into(),
            ))
        }
        Err(e) => Err(e.into()),
    }
}

// ----------------------------------------------------------------------------
// Project ownership (§7, M3). A `projects` row is keyed by the stable, opaque
// key the CLI mints in `.picloud/`; `groups.owner_project` references it. These
// helpers are free functions (pool + tx variants) so the apply path can claim
// ownership inside the single apply transaction (first-commit-wins), while the
// read-only plan path can surface conflicts without writing.
// ----------------------------------------------------------------------------

/// Resolve a project key to its id, if the project has ever been persisted.
/// `None` means no apply has claimed under this key yet (so `plan` treats every
/// node as claimable). Read-only — used by the plan path.
pub(crate) async fn get_project_id_by_key(
    pool: &PgPool,
    key: &str,
) -> Result<Option<Uuid>, GroupRepositoryError> {
    let row: Option<(Uuid,)> = sqlx::query_as("SELECT id FROM projects WHERE key = $1")
        .bind(key)
        .fetch_optional(pool)
        .await?;
    Ok(row.map(|r| r.0))
}

/// Upsert the project keyed by `key` inside the apply tx and return its id.
/// Idempotent: re-applying the same repo reuses the same project identity.
pub(crate) async fn upsert_project_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    key: &str,
) -> Result<Uuid, GroupRepositoryError> {
    let row: (Uuid,) = sqlx::query_as(
        "INSERT INTO projects (key) VALUES ($1) \
         ON CONFLICT (key) DO UPDATE SET key = EXCLUDED.key RETURNING id",
    )
    .bind(key)
    .fetch_one(&mut **tx)
    .await?;
    Ok(row.0)
}

/// The project that owns `group_id`, as `(project_id, project_key)`. `None` when
/// the node is UI/API-owned (no claim). Read-only — used by plan + the in-tx
/// authoritative re-check. Joins through `projects` so the caller gets the
/// human-facing key for a conflict message.
pub(crate) async fn get_group_owner(
    pool: &PgPool,
    group_id: GroupId,
) -> Result<Option<(Uuid, String)>, GroupRepositoryError> {
    let row: Option<(Uuid, String)> = sqlx::query_as(
        "SELECT p.id, p.key FROM groups g \
         JOIN projects p ON p.id = g.owner_project WHERE g.id = $1",
    )
    .bind(group_id.into_inner())
    .fetch_optional(pool)
    .await?;
    Ok(row)
}

/// The same owner lookup, but inside the apply tx — the authoritative read that
/// the claim/conflict decision keys on (so a concurrent claim that committed
/// after `plan` is seen under this tx's per-node advisory lock).
pub(crate) async fn get_group_owner_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    group_id: GroupId,
) -> Result<Option<(Uuid, String)>, GroupRepositoryError> {
    let row: Option<(Uuid, String)> = sqlx::query_as(
        "SELECT p.id, p.key FROM groups g \
         JOIN projects p ON p.id = g.owner_project WHERE g.id = $1",
    )
    .bind(group_id.into_inner())
    .fetch_optional(&mut **tx)
    .await?;
    Ok(row)
}

/// Stamp (claim or take over) `group_id`'s owning project inside the apply tx.
/// Bumps `structure_version` so an ownership flip trips a bound plan token.
pub(crate) async fn set_group_owner_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    group_id: GroupId,
    project_id: Uuid,
) -> Result<(), GroupRepositoryError> {
    let res = sqlx::query(
        "UPDATE groups SET owner_project = $2, \
             structure_version = structure_version + 1, updated_at = NOW() WHERE id = $1",
    )
    .bind(group_id.into_inner())
    .bind(project_id)
    .execute(&mut **tx)
    .await?;
    if res.rows_affected() == 0 {
        return Err(GroupRepositoryError::NotFound(group_id));
    }
    Ok(())
}

/// Every group `(id, slug)` owned by `project_id` — the candidate set for
/// structural prune (those absent from the manifest are the ones to delete).
/// Read-only; the apply path filters and deletes leaf-first under `delete_tx`.
pub(crate) async fn list_owned_groups(
    pool: &PgPool,
    project_id: Uuid,
) -> Result<Vec<(GroupId, String)>, GroupRepositoryError> {
    let rows: Vec<(Uuid, String)> =
        sqlx::query_as("SELECT id, slug FROM groups WHERE owner_project = $1")
            .bind(project_id)
            .fetch_all(pool)
            .await?;
    Ok(rows
        .into_iter()
        .map(|(id, slug)| (id.into(), slug))
        .collect())
}

/// The same owned-group enumeration inside the apply tx, with each node's
/// `parent_id` so the prune can delete leaf-first (children before parents).
pub(crate) async fn list_owned_groups_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    project_id: Uuid,
) -> Result<Vec<(GroupId, String, Option<GroupId>)>, GroupRepositoryError> {
    let rows: Vec<(Uuid, String, Option<Uuid>)> =
        sqlx::query_as("SELECT id, slug, parent_id FROM groups WHERE owner_project = $1")
            .bind(project_id)
            .fetch_all(&mut **tx)
            .await?;
    Ok(rows
        .into_iter()
        .map(|(id, slug, parent)| (id.into(), slug, parent.map(GroupId::from)))
        .collect())
}

/// Recursive descendant-app query, the inverse of the ancestor `CHAIN_LEVELS_CTE`
/// in `config_resolver`: walk `groups.parent_id` DOWN from `$1` to collect the
/// whole subtree of group ids, then every app whose `group_id` falls in it.
/// Ordered by `app_id` so callers lock/iterate deterministically. Depth-bounded
/// `< 64` as a runaway guard (the cycle guard already forbids real cycles).
const DESCENDANT_APPS_SQL: &str = "\
    WITH RECURSIVE subtree AS ( \
        SELECT id, 0 AS depth FROM groups WHERE id = $1 \
      UNION ALL \
        SELECT g.id, s.depth + 1 FROM groups g JOIN subtree s ON g.parent_id = s.id \
        WHERE s.depth < 64 \
    ) \
    SELECT a.id FROM apps a WHERE a.group_id IN (SELECT id FROM subtree) ORDER BY a.id";

/// Every app in the subtree rooted at `group_id` (the group itself and all
/// descendant groups), `app_id`-ordered. The read-only pool variant — used by
/// `plan` to size a template's true blast radius across the whole DB subtree,
/// not just the apps present in the current apply.
///
/// # Errors
/// Propagates sqlx errors.
pub async fn descendant_app_ids(
    pool: &PgPool,
    group_id: GroupId,
) -> Result<Vec<AppId>, GroupRepositoryError> {
    let rows: Vec<(Uuid,)> = sqlx::query_as(DESCENDANT_APPS_SQL)
        .bind(group_id.into_inner())
        .fetch_all(pool)
        .await?;
    Ok(rows.into_iter().map(|(id,)| id.into()).collect())
}

/// The same subtree enumeration inside the apply tx, so groups/apps created
/// earlier in this transaction (M2 structural reconcile) are included when
/// fanning a template out to its descendants (§4.5, M7).
pub(crate) async fn descendant_app_ids_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    group_id: GroupId,
) -> Result<Vec<AppId>, GroupRepositoryError> {
    let rows: Vec<(Uuid,)> = sqlx::query_as(DESCENDANT_APPS_SQL)
        .bind(group_id.into_inner())
        .fetch_all(&mut **tx)
        .await?;
    Ok(rows.into_iter().map(|(id,)| id.into()).collect())
}

#[derive(sqlx::FromRow)]
struct GroupRow {
    id: Uuid,
    parent_id: Option<Uuid>,
    slug: String,
    name: String,
    description: Option<String>,
    structure_version: i64,
    created_at: chrono::DateTime<chrono::Utc>,
    updated_at: chrono::DateTime<chrono::Utc>,
}

impl From<GroupRow> for Group {
    fn from(r: GroupRow) -> Self {
        Self {
            id: r.id.into(),
            parent_id: r.parent_id.map(Into::into),
            slug: r.slug,
            name: r.name,
            description: r.description,
            structure_version: r.structure_version,
            created_at: r.created_at,
            updated_at: r.updated_at,
        }
    }
}