use std::time::Duration;

use export::{AppState, KenAuthError};
use models::dir::{dir_overwrite_group, dir_overwrite_users, dirs};
use models::repo::repos;
use models::room::rooms;
use models::space::{
    SpaceStatus, space_extras, space_group_permissions, space_groups, space_invites,
    space_join_requests, space_join_rules, space_member_extras, space_member_groups, space_members,
    space_warn, spaces,
};
use models::system::{account_deletion, space_deletion};
use models::users::{
    user, user_agreement_acceptances, user_blocks, user_external_identity, user_favorite_memes,
    user_followers, user_notifications, user_preferences, user_profiles, user_push_subscriptions,
    user_recent_mentions, user_saved_messages,
};
use proto::KenShellClient;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect, Set,
};
use serde::Serialize;
use time::OffsetDateTime;
use uuid::Uuid;

const RETRY_BASE_SECS: u64 = 30;
const RETRY_MAX_SECS: u64 = 3600;

#[derive(Debug, thiserror::Error)]
pub enum DeletionError {
    #[error("database: {0}")]
    Database(#[from] sea_orm::DbErr),
    #[error("{0}")]
    Conflict(String),
    #[error("{0}")]
    NotFound(String),
    #[error("kenoauth: {0}")]
    KenAuth(String),
    #[error("kenshell: {0}")]
    KenShell(String),
    #[error("config: {0}")]
    Config(String),
}

impl DeletionError {
    pub fn is_retryable(&self) -> bool {
        !matches!(
            self,
            Self::Conflict(_) | Self::NotFound(_) | Self::Config(_)
        )
    }
}

fn ken_auth(error: KenAuthError) -> DeletionError {
    match error {
        KenAuthError::NotFound => {
            DeletionError::NotFound("account or organization not found".into())
        }
        KenAuthError::Rejected(message) => DeletionError::Conflict(message),
        other => DeletionError::KenAuth(other.to_string()),
    }
}

fn ken_shell(error: proto::KenShellError) -> DeletionError {
    DeletionError::KenShell(error.to_string())
}

fn client(state: &AppState) -> Result<&dyn export::AuthApi, DeletionError> {
    state
        .kenauth
        .as_deref()
        .ok_or_else(|| DeletionError::Config("kenoauth is not configured".into()))
}

fn shell_client(state: &AppState) -> Result<KenShellClient, DeletionError> {
    let cfg = &state.config.git;
    KenShellClient::connect_with_key(
        &cfg.kenshell_endpoint,
        &cfg.kenshell_rpc_key,
        Duration::from_secs(cfg.kenshell_timeout_secs.max(1)),
    )
    .map_err(ken_shell)
}

fn now() -> OffsetDateTime {
    OffsetDateTime::now_utc()
}

fn grace_until(days: u32) -> OffsetDateTime {
    now() + time::Duration::days(i64::from(days.max(1)))
}

fn retry_at(attempts: i32) -> OffsetDateTime {
    let exponent = attempts.clamp(1, 8) as u32;
    let delay = (RETRY_BASE_SECS.saturating_mul(1u64 << exponent)).min(RETRY_MAX_SECS);
    now() + time::Duration::seconds(delay as i64)
}

#[derive(Debug, Clone, Serialize)]
pub struct AccountDeletionStatus {
    pub status: String,
    pub requested_at: OffsetDateTime,
    pub purge_after: OffsetDateTime,
    pub last_error: Option<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct SpaceDeletionStatus {
    pub status: String,
    pub requested_at: OffsetDateTime,
    pub purge_after: OffsetDateTime,
    pub last_error: Option<String>,
}

pub async fn request_account_deletion(
    state: &AppState,
    user_id: i64,
    reason: &str,
) -> Result<AccountDeletionStatus, DeletionError> {
    let db = state.database.write();
    if let Some(existing) = account_deletion::Entity::find()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .filter(account_deletion::Column::Status.eq("pending"))
        .one(db)
        .await?
    {
        return Ok(status_of_account(&existing));
    }

    let owned: Vec<Uuid> = spaces::Entity::find()
        .filter(spaces::Column::OwnerUserId.eq(user_id))
        .filter(spaces::Column::DeletedAt.is_null())
        .select_only()
        .column(spaces::Column::Id)
        .into_tuple()
        .all(db)
        .await?;
    for space_id in owned {
        let pending = space_deletion::Entity::find()
            .filter(space_deletion::Column::SpaceId.eq(space_id))
            .filter(space_deletion::Column::Status.eq("pending"))
            .one(db)
            .await?;
        if pending.is_none() {
            return Err(DeletionError::Conflict(
                "delete or transfer owned spaces before deleting the account".into(),
            ));
        }
    }
    let client = client(state)?;
    let status = client.get_user_status(user_id).await.map_err(ken_auth)?;
    if status.deletion_requested_at.is_none() && status.deleted_at.is_none() {
        client
            .delete_account(user_id, reason)
            .await
            .map_err(ken_auth)?;
    }

    let now = now();
    let purge_after = grace_until(state.config.user.deletion_grace_days);
    let row = match account_deletion::Entity::find()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .one(db)
        .await?
    {
        Some(existing) => {
            let mut active: account_deletion::ActiveModel = existing.into();
            active.status = Set("pending".into());
            active.requested_at = Set(now);
            active.purge_after = Set(purge_after);
            active.available_at = Set(purge_after);
            active.last_error = Set(None);
            active.attempts = Set(0);
            active.updated_at = Set(now);
            active.update(db).await?
        }
        None => {
            account_deletion::ActiveModel {
                id: Set(Uuid::now_v7()),
                user_id: Set(user_id),
                requested_at: Set(now),
                purge_after: Set(purge_after),
                status: Set("pending".into()),
                last_error: Set(None),
                attempts: Set(0),
                available_at: Set(purge_after),
                created_at: Set(now),
                updated_at: Set(now),
            }
            .insert(db)
            .await?
        }
    };
    Ok(status_of_account(&row))
}

pub async fn cancel_account_deletion(state: &AppState, user_id: i64) -> Result<(), DeletionError> {
    client(state)?
        .cancel_account_deletion(user_id)
        .await
        .map_err(ken_auth)?;
    account_deletion::Entity::delete_many()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .exec(state.database.write())
        .await?;
    Ok(())
}

pub async fn account_deletion_status(
    state: &AppState,
    user_id: i64,
) -> Result<Option<AccountDeletionStatus>, DeletionError> {
    Ok(account_deletion::Entity::find()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .one(state.database.read())
        .await?
        .as_ref()
        .map(status_of_account))
}

pub async fn request_space_deletion(
    state: &AppState,
    actor: i64,
    space_id: Uuid,
    reason: &str,
) -> Result<SpaceDeletionStatus, DeletionError> {
    let db = state.database.write();
    let space = spaces::Entity::find_by_id(space_id)
        .one(db)
        .await?
        .ok_or_else(|| DeletionError::NotFound("space not found".into()))?;
    if space.status != SpaceStatus::Archived {
        let mut active: spaces::ActiveModel = space.into();
        active.status = Set(SpaceStatus::Archived);
        active.archived_at = Set(Some(sea_orm::prelude::DateTimeUtc::from(
            std::time::SystemTime::now(),
        )));
        active.suspended_reason = Set(Some(reason.to_string()));
        active.update(db).await?;
    }

    let now = now();
    let purge_after = grace_until(state.config.space.deletion_grace_days);
    let row = match space_deletion::Entity::find()
        .filter(space_deletion::Column::SpaceId.eq(space_id))
        .one(db)
        .await?
    {
        Some(existing) => {
            let mut active: space_deletion::ActiveModel = existing.into();
            active.requested_by = Set(actor);
            active.requested_at = Set(now);
            active.purge_after = Set(purge_after);
            active.status = Set("pending".into());
            active.available_at = Set(purge_after);
            active.last_error = Set(None);
            active.attempts = Set(0);
            active.updated_at = Set(now);
            active.update(db).await?
        }
        None => {
            space_deletion::ActiveModel {
                id: Set(Uuid::now_v7()),
                space_id: Set(space_id),
                requested_by: Set(actor),
                requested_at: Set(now),
                purge_after: Set(purge_after),
                status: Set("pending".into()),
                last_error: Set(None),
                attempts: Set(0),
                available_at: Set(purge_after),
                created_at: Set(now),
                updated_at: Set(now),
            }
            .insert(db)
            .await?
        }
    };
    Ok(status_of_space(&row))
}

pub async fn cancel_space_deletion(
    state: &AppState,
    _actor: i64,
    space_id: Uuid,
) -> Result<(), DeletionError> {
    let db = state.database.write();
    if let Some(space) = spaces::Entity::find_by_id(space_id).one(db).await? {
        let mut active: spaces::ActiveModel = space.into();
        active.status = Set(SpaceStatus::Active);
        active.archived_at = Set(None);
        active.update(db).await?;
    }
    space_deletion::Entity::delete_many()
        .filter(space_deletion::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    Ok(())
}

pub async fn space_deletion_status(
    state: &AppState,
    space_id: Uuid,
) -> Result<Option<SpaceDeletionStatus>, DeletionError> {
    Ok(space_deletion::Entity::find()
        .filter(space_deletion::Column::SpaceId.eq(space_id))
        .one(state.database.read())
        .await?
        .as_ref()
        .map(status_of_space))
}

fn status_of_account(row: &account_deletion::Model) -> AccountDeletionStatus {
    AccountDeletionStatus {
        status: row.status.clone(),
        requested_at: row.requested_at,
        purge_after: row.purge_after,
        last_error: row.last_error.clone(),
    }
}

fn status_of_space(row: &space_deletion::Model) -> SpaceDeletionStatus {
    SpaceDeletionStatus {
        status: row.status.clone(),
        requested_at: row.requested_at,
        purge_after: row.purge_after,
        last_error: row.last_error.clone(),
    }
}

pub async fn run_worker(state: AppState) {
    tracing::info!("deletion worker started");
    let mut interval = tokio::time::interval(Duration::from_secs(30));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        interval.tick().await;
        if let Err(error) = purge_due(&state).await {
            tracing::warn!(%error, "deletion worker pass failed");
        }
    }
}

async fn purge_due(state: &AppState) -> Result<(), DeletionError> {
    let db = state.database.read();
    let cutoff = now();
    let spaces = space_deletion::Entity::find()
        .filter(space_deletion::Column::Status.eq("pending"))
        .filter(space_deletion::Column::AvailableAt.lte(cutoff))
        .all(db)
        .await?;
    for row in spaces {
        if let Err(error) = purge_space(state, &row).await {
            record_space_failure(state, &row, &error).await?;
        }
    }

    let accounts = account_deletion::Entity::find()
        .filter(account_deletion::Column::Status.eq("pending"))
        .filter(account_deletion::Column::AvailableAt.lte(cutoff))
        .all(db)
        .await?;
    for row in accounts {
        if let Err(error) = purge_account(state, &row).await {
            record_account_failure(state, &row, &error).await?;
        }
    }
    Ok(())
}

async fn purge_space(state: &AppState, row: &space_deletion::Model) -> Result<(), DeletionError> {
    // Never delete a space whose ownership changed after the request.
    let space = {
        let db = state.database.write();
        spaces::Entity::find_by_id(row.space_id).one(db).await?
    };
    let Some(space) = space else {
        space_deletion::Entity::delete_by_id(row.id)
            .exec(state.database.write())
            .await?;
        return Ok(());
    };
    if space.owner_user_id != row.requested_by {
        let db = state.database.write();
        let mut active: spaces::ActiveModel = space.into();
        active.status = Set(SpaceStatus::Active);
        active.archived_at = Set(None);
        active.update(db).await?;
        space_deletion::Entity::delete_by_id(row.id)
            .exec(db)
            .await?;
        return Ok(());
    }

    // 1. KenShell is the owner of repository/placement/LFS bytes.
    let shell = shell_client(state)?;
    let repos = shell
        .list_repositories_for_org(&row.space_id.to_string())
        .await
        .map_err(ken_shell)?;
    for repo in repos {
        match shell.delete_repository(&repo.id).await {
            Ok(()) => {}
            Err(error) if error.is_not_found() => {}
            Err(error) => return Err(ken_shell(error)),
        }
    }

    // 2. Local KenApi resources become unreachable and space-scoped metadata is removed.
    archive_local_space(state, row.space_id).await?;

    // 3. The space row and its members/roles/invites cascade away.
    let db = state.database.write();
    spaces::Entity::delete_by_id(row.space_id).exec(db).await?;

    space_deletion::Entity::delete_by_id(row.id)
        .exec(db)
        .await?;
    tracing::info!(space_id = %row.space_id, "space purged");
    Ok(())
}

async fn archive_local_space(state: &AppState, space_id: Uuid) -> Result<(), DeletionError> {
    let db = state.database.write();
    let now = now();

    rooms::Entity::update_many()
        .col_expr(
            rooms::Column::IsActive,
            sea_orm::sea_query::Expr::value(false),
        )
        .filter(rooms::Column::SpaceId.eq(Some(space_id)))
        .exec(db)
        .await?;

    // Only the root nodes carry `space_id`; walk `parent_dir_id` to collect the
    // whole tree, otherwise nested directories/repos survive the purge.
    let mut dir_ids: Vec<Uuid> = dirs::Entity::find()
        .filter(dirs::Column::SpaceId.eq(Some(space_id)))
        .select_only()
        .column(dirs::Column::Id)
        .into_tuple()
        .all(db)
        .await?;
    let mut seen: std::collections::HashSet<Uuid> = dir_ids.iter().copied().collect();
    let mut frontier = dir_ids.clone();
    while !frontier.is_empty() {
        let children: Vec<Uuid> = dirs::Entity::find()
            .filter(dirs::Column::ParentDirId.is_in(frontier))
            .select_only()
            .column(dirs::Column::Id)
            .into_tuple()
            .all(db)
            .await?;
        let mut next = Vec::new();
        for id in children {
            if seen.insert(id) {
                dir_ids.push(id);
                next.push(id);
            }
        }
        frontier = next;
    }
    if !dir_ids.is_empty() {
        dirs::Entity::update_many()
            .col_expr(
                dirs::Column::IsArchived,
                sea_orm::sea_query::Expr::value(true),
            )
            .col_expr(
                dirs::Column::ArchivedAt,
                sea_orm::sea_query::Expr::value(Some(now)),
            )
            .filter(dirs::Column::Id.is_in(dir_ids.clone()))
            .exec(db)
            .await?;
        repos::Entity::update_many()
            .col_expr(
                repos::Column::IsArchived,
                sea_orm::sea_query::Expr::value(true),
            )
            .filter(repos::Column::DirId.is_in(dir_ids.clone()))
            .exec(db)
            .await?;
        dir_overwrite_group::Entity::delete_many()
            .filter(dir_overwrite_group::Column::DirId.is_in(dir_ids.clone()))
            .exec(db)
            .await?;
        dir_overwrite_users::Entity::delete_many()
            .filter(dir_overwrite_users::Column::DirId.is_in(dir_ids))
            .exec(db)
            .await?;
    }

    let group_ids: Vec<Uuid> = space_groups::Entity::find()
        .filter(space_groups::Column::SpaceId.eq(space_id))
        .select_only()
        .column(space_groups::Column::Id)
        .into_tuple()
        .all(db)
        .await?;
    if !group_ids.is_empty() {
        space_group_permissions::Entity::delete_many()
            .filter(space_group_permissions::Column::GroupId.is_in(group_ids))
            .exec(db)
            .await?;
    }
    space_member_groups::Entity::delete_many()
        .filter(space_member_groups::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    space_groups::Entity::delete_many()
        .filter(space_groups::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    space_invites::Entity::delete_many()
        .filter(space_invites::Column::OrganizationId.eq(space_id))
        .exec(db)
        .await?;
    space_join_requests::Entity::delete_many()
        .filter(space_join_requests::Column::OrganizationId.eq(space_id))
        .exec(db)
        .await?;
    space_join_rules::Entity::delete_many()
        .filter(space_join_rules::Column::OrganizationId.eq(space_id))
        .exec(db)
        .await?;
    space_warn::Entity::delete_many()
        .filter(space_warn::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    space_extras::Entity::delete_by_id(space_id)
        .exec(db)
        .await?;
    space_member_extras::Entity::delete_many()
        .filter(space_member_extras::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    Ok(())
}

async fn purge_account(
    state: &AppState,
    row: &account_deletion::Model,
) -> Result<(), DeletionError> {
    let db = state.database.write();
    let owned = spaces::Entity::find()
        .filter(spaces::Column::OwnerUserId.eq(row.user_id))
        .filter(spaces::Column::DeletedAt.is_null())
        .count(db)
        .await?;
    if owned > 0 {
        return Err(DeletionError::Conflict(
            "owned spaces are still being purged".into(),
        ));
    }

    let now = now();

    space_members::Entity::delete_many()
        .filter(space_members::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_preferences::Entity::delete_many()
        .filter(user_preferences::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_profiles::Entity::delete_many()
        .filter(user_profiles::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_notifications::Entity::delete_many()
        .filter(user_notifications::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_push_subscriptions::Entity::delete_many()
        .filter(user_push_subscriptions::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_favorite_memes::Entity::delete_many()
        .filter(user_favorite_memes::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_recent_mentions::Entity::delete_many()
        .filter(user_recent_mentions::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_saved_messages::Entity::delete_many()
        .filter(user_saved_messages::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_external_identity::Entity::delete_many()
        .filter(user_external_identity::Column::UserId.eq(Some(row.user_id)))
        .exec(db)
        .await?;
    user_agreement_acceptances::Entity::delete_many()
        .filter(user_agreement_acceptances::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_blocks::Entity::delete_many()
        .filter(
            sea_orm::Condition::any()
                .add(user_blocks::Column::BlockerId.eq(row.user_id))
                .add(user_blocks::Column::BlockedId.eq(row.user_id)),
        )
        .exec(db)
        .await?;
    user_followers::Entity::delete_many()
        .filter(
            sea_orm::Condition::any()
                .add(user_followers::Column::FollowerId.eq(row.user_id))
                .add(user_followers::Column::FollowedId.eq(row.user_id)),
        )
        .exec(db)
        .await?;

    if let Some(user) = user::Entity::find_by_id(row.user_id).one(db).await? {
        let mut active: user::ActiveModel = user.into();
        active.is_restricted = Set(true);
        active.local_avatar_url = Set(String::new());
        active.visibility = Set(models::users::user::ProfileVisibility::Private);
        active.last_login_at = Set(None);
        active.last_login_ip_hash = Set(None);
        active.updated_at = Set(now);
        active.update(db).await?;
    }

    let client = client(state)?;
    match client.purge_account(row.user_id).await {
        Ok(()) | Err(KenAuthError::NotFound) => {}
        Err(error) => return Err(ken_auth(error)),
    }
    account_deletion::Entity::delete_by_id(row.id)
        .exec(db)
        .await?;
    tracing::info!(user_id = %row.user_id, "account purged");
    Ok(())
}

async fn record_space_failure(
    state: &AppState,
    row: &space_deletion::Model,
    error: &DeletionError,
) -> Result<(), DeletionError> {
    let attempts = row.attempts.saturating_add(1);
    let mut active: space_deletion::ActiveModel = row.clone().into();
    active.attempts = Set(attempts);
    active.available_at = Set(retry_at(attempts));
    active.last_error = Set(Some(error.to_string()));
    active.status = Set("pending".into());
    active.updated_at = Set(now());
    active.update(state.database.write()).await?;
    tracing::warn!(space_id = %row.space_id, %error, attempts, "space purge failed; will retry");
    Ok(())
}

async fn record_account_failure(
    state: &AppState,
    row: &account_deletion::Model,
    error: &DeletionError,
) -> Result<(), DeletionError> {
    let attempts = row.attempts.saturating_add(1);
    let mut active: account_deletion::ActiveModel = row.clone().into();
    active.attempts = Set(attempts);
    active.available_at = Set(retry_at(attempts));
    active.last_error = Set(Some(error.to_string()));
    active.status = Set("pending".into());
    active.updated_at = Set(now());
    active.update(state.database.write()).await?;
    tracing::warn!(user_id = %row.user_id, %error, attempts, "account purge failed; will retry");
    Ok(())
}