//! SyncKit groups: the shared-changelog membership and sealed-GCK-grant layer.
//!
//! A group is owned by one admin who mints a Group Content Key (GCK) and seals it
//! to each member's identity public key. The server stores membership and the
//! opaque sealed grants; it never sees the GCK or any plaintext. The group-scoped
//! push/pull that reads `sync_log.group_id` lives in `log.rs` (p2-changelog); this
//! module owns the group, membership, and grant records. Design: wiki
//! synckit-groups-design.

use sqlx::PgPool;

use crate::db::models::{DbSyncGroup, DbSyncGroupMember};
use crate::db::{SyncAppId, SyncGroupId, UserId};
use crate::error::Result;

/// Create a group and enroll its admin as the first member.
///
/// Runs in one transaction: the `sync_groups` row (GCK generation 1) and the
/// admin's own `sync_group_members` row (`role = 'admin'`, carrying the admin's
/// self-sealed GCK grant) are written together, so a group never exists without
/// its admin able to read it.
#[tracing::instrument(skip_all)]
pub async fn create_group(
    pool: &PgPool,
    id: SyncGroupId,
    app_id: SyncAppId,
    admin_user_id: UserId,
    name: &str,
    admin_sealed_gck: &str,
    admin_pubkey: &str,
) -> Result<DbSyncGroup> {
    let mut tx = pool.begin().await?;

    // The id is client-generated (the admin's GCK grant is sealed bound to it
    // before the group exists). A PK collision surfaces as a DB error.
    let group = sqlx::query_as::<_, DbSyncGroup>(
        r"
        INSERT INTO sync_groups (id, app_id, admin_user_id, name)
        VALUES ($1, $2, $3, $4)
        RETURNING id, app_id, admin_user_id, name, gck_version, created_at
        ",
    )
    .bind(id)
    .bind(app_id)
    .bind(admin_user_id)
    .bind(name)
    .fetch_one(&mut *tx)
    .await?;

    sqlx::query(
        r"
        INSERT INTO sync_group_members (group_id, user_id, role, sealed_gck, gck_version, member_pubkey)
        VALUES ($1, $2, 'admin', $3, $4, $5)
        ",
    )
    .bind(group.id)
    .bind(admin_user_id)
    .bind(admin_sealed_gck)
    .bind(group.gck_version)
    .bind(admin_pubkey)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(group)
}

/// Fetch a group scoped to its app (the app scope guards against cross-app id
/// guessing).
#[tracing::instrument(skip_all)]
pub async fn get_group(
    pool: &PgPool,
    app_id: SyncAppId,
    group_id: SyncGroupId,
) -> Result<Option<DbSyncGroup>> {
    let group = sqlx::query_as::<_, DbSyncGroup>(
        r"
        SELECT id, app_id, admin_user_id, name, gck_version, created_at
        FROM sync_groups
        WHERE app_id = $1 AND id = $2
        ",
    )
    .bind(app_id)
    .bind(group_id)
    .fetch_optional(pool)
    .await?;
    Ok(group)
}

/// List the groups a user belongs to within an app, most recent first.
#[tracing::instrument(skip_all)]
pub async fn list_groups_for_user(
    pool: &PgPool,
    app_id: SyncAppId,
    user_id: UserId,
) -> Result<Vec<DbSyncGroup>> {
    let groups = sqlx::query_as::<_, DbSyncGroup>(
        r"
        SELECT g.id, g.app_id, g.admin_user_id, g.name, g.gck_version, g.created_at
        FROM sync_groups g
        JOIN sync_group_members m ON m.group_id = g.id
        WHERE g.app_id = $1 AND m.user_id = $2
        ORDER BY g.created_at DESC
        ",
    )
    .bind(app_id)
    .bind(user_id)
    .fetch_all(pool)
    .await?;
    Ok(groups)
}

/// All members of a group.
#[tracing::instrument(skip_all)]
pub async fn list_members(pool: &PgPool, group_id: SyncGroupId) -> Result<Vec<DbSyncGroupMember>> {
    let members = sqlx::query_as::<_, DbSyncGroupMember>(
        r"
        SELECT m.group_id, m.user_id, u.email, m.role, m.sealed_gck, m.gck_version, m.added_at
        FROM sync_group_members m
        JOIN users u ON u.id = m.user_id
        WHERE m.group_id = $1
        ORDER BY m.added_at ASC
        ",
    )
    .bind(group_id)
    .fetch_all(pool)
    .await?;
    Ok(members)
}

/// The `(user_id, member_pubkey)` of every member with a stored public key. The
/// admin fetches this to re-seal a rotated GCK to the remaining members without
/// re-collecting their keys out of band. Members without a stored key (legacy
/// rows) are omitted.
#[tracing::instrument(skip_all)]
pub async fn list_member_pubkeys(
    pool: &PgPool,
    group_id: SyncGroupId,
) -> Result<Vec<(UserId, String)>> {
    let rows: Vec<(UserId, String)> = sqlx::query_as(
        "SELECT user_id, member_pubkey FROM sync_group_members \
         WHERE group_id = $1 AND member_pubkey IS NOT NULL \
         ORDER BY added_at ASC",
    )
    .bind(group_id)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Whether `user_id` is a member of `group_id`. The membership gate for
/// group-scoped push/pull.
#[tracing::instrument(skip_all)]
pub async fn is_group_member(
    pool: &PgPool,
    group_id: SyncGroupId,
    user_id: UserId,
) -> Result<bool> {
    let exists: bool = sqlx::query_scalar(
        "SELECT EXISTS (SELECT 1 FROM sync_group_members WHERE group_id = $1 AND user_id = $2)",
    )
    .bind(group_id)
    .bind(user_id)
    .fetch_one(pool)
    .await?;
    Ok(exists)
}

/// Whether `user_id` is the admin of `group_id`. Admin-only actions (add/remove
/// member, rotate) gate on this.
#[tracing::instrument(skip_all)]
pub async fn is_group_admin(pool: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<bool> {
    let exists: bool = sqlx::query_scalar(
        "SELECT EXISTS (SELECT 1 FROM sync_groups WHERE id = $1 AND admin_user_id = $2)",
    )
    .bind(group_id)
    .bind(user_id)
    .fetch_one(pool)
    .await?;
    Ok(exists)
}

/// Fetch a member's current sealed GCK grant `(sealed_gck, gck_version)`, so the
/// member's device can open the GCK and read the group changelog. `None` if the
/// user is not a member.
#[tracing::instrument(skip_all)]
pub async fn get_member_grant(
    pool: &PgPool,
    group_id: SyncGroupId,
    user_id: UserId,
) -> Result<Option<(String, i32)>> {
    let grant: Option<(String, i32)> = sqlx::query_as(
        "SELECT sealed_gck, gck_version FROM sync_group_members WHERE group_id = $1 AND user_id = $2",
    )
    .bind(group_id)
    .bind(user_id)
    .fetch_optional(pool)
    .await?;
    Ok(grant)
}

/// Add a member, or replace an existing member's grant (idempotent upsert).
///
/// The admin calls this with a grant it sealed to the member's public key at the
/// group's current `gck_version`. Re-adding an existing member updates their
/// grant and role in place.
#[tracing::instrument(skip_all)]
pub async fn add_or_update_member(
    pool: &PgPool,
    group_id: SyncGroupId,
    user_id: UserId,
    role: &str,
    sealed_gck: &str,
    gck_version: i32,
    member_pubkey: &str,
) -> Result<()> {
    sqlx::query(
        r"
        INSERT INTO sync_group_members (group_id, user_id, role, sealed_gck, gck_version, member_pubkey)
        VALUES ($1, $2, $3, $4, $5, $6)
        ON CONFLICT (group_id, user_id)
        DO UPDATE SET role = EXCLUDED.role,
                      sealed_gck = EXCLUDED.sealed_gck,
                      gck_version = EXCLUDED.gck_version,
                      member_pubkey = EXCLUDED.member_pubkey
        ",
    )
    .bind(group_id)
    .bind(user_id)
    .bind(role)
    .bind(sealed_gck)
    .bind(gck_version)
    .bind(member_pubkey)
    .execute(pool)
    .await?;
    Ok(())
}

/// Remove a member. Returns `true` if a membership row was deleted.
///
/// This drops the member's access to future group writes; forward secrecy for
/// writes after removal comes from the caller then rotating the GCK
/// ([`rotate_group_gck`]). Data the removed member already pulled is,
/// unavoidably, already in their hands.
#[tracing::instrument(skip_all)]
pub async fn remove_member(pool: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<bool> {
    let result = sqlx::query("DELETE FROM sync_group_members WHERE group_id = $1 AND user_id = $2")
        .bind(group_id)
        .bind(user_id)
        .execute(pool)
        .await?;
    Ok(result.rows_affected() > 0)
}

/// Rotate the group's GCK to `new_version`, atomically replacing every member's
/// grant with one sealed under the new key.
///
/// The admin mints a fresh GCK, seals it to the remaining members' stored public
/// keys, and passes `(user_id, sealed_gck)` for each. In one transaction this
/// bumps `sync_groups.gck_version`, deletes any member not in `grants` (the
/// removed set), and upserts each provided grant at `new_version`. Callers must
/// include the admin's own re-sealed grant.
#[tracing::instrument(skip_all)]
pub async fn rotate_group_gck(
    pool: &PgPool,
    group_id: SyncGroupId,
    new_version: i32,
    grants: &[(UserId, String)],
) -> Result<()> {
    let mut tx = pool.begin().await?;

    sqlx::query("UPDATE sync_groups SET gck_version = $1 WHERE id = $2")
        .bind(new_version)
        .bind(group_id)
        .execute(&mut *tx)
        .await?;

    // Anyone not in the new grant set is removed by the rotation.
    let keep: Vec<UserId> = grants.iter().map(|(u, _)| *u).collect();
    sqlx::query("DELETE FROM sync_group_members WHERE group_id = $1 AND user_id <> ALL($2)")
        .bind(group_id)
        .bind(&keep)
        .execute(&mut *tx)
        .await?;

    for (user_id, sealed_gck) in grants {
        sqlx::query(
            r"
            UPDATE sync_group_members
            SET sealed_gck = $3, gck_version = $4
            WHERE group_id = $1 AND user_id = $2
            ",
        )
        .bind(group_id)
        .bind(user_id)
        .bind(sealed_gck)
        .bind(new_version)
        .execute(&mut *tx)
        .await?;
    }

    tx.commit().await?;
    Ok(())
}