"""Administrator actions on user accounts (SPEC §6.1 admin panel).

Search, per-user quota, blocking and deletion. Blocking keeps every byte of the
user's data (and keeps their mail syncing) so unblocking is instant; it only
refuses sign-in and revokes the sessions they hold. Deletion is total: mail
accounts (through the same path the user's own "remove account" takes, so the
sync runners stop), plugin rows, blobs on disk, the search index and the user
row, whose foreign keys cascade to everything else core owns.
"""

from __future__ import annotations

import datetime as dt
import shutil
import uuid
from pathlib import Path

from sqlalchemy import func, or_, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.accounts.service import delete_account
from app.config import get_settings
from app.events import get_event_bus
from app.identity import trusted_devices
from app.identity.sessions import destroy_all_sessions
from app.models import Account, Message, User
from app.plugins.db import plugin_session
from app.search.factory import get_search_backend

SEARCH_LIMIT = 25


async def search_users(db: AsyncSession, q: str) -> list[tuple[User, int, list[str]]]:
    """Users whose login, or one of whose mail accounts' address or display name,
    contains ``q`` (case-insensitive). An empty ``q`` lists the first users by
    login. Each row comes with its storage used and its mail account addresses."""
    stmt = select(User).order_by(User.handle).limit(SEARCH_LIMIT)
    q = q.strip()
    if q:
        # Typed % and _ are literal characters, not wildcards.
        escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        like = f"%{escaped}%"
        by_account = select(Account.user_id).where(
            or_(Account.email.ilike(like), Account.display_name.ilike(like))
        )
        stmt = stmt.where(or_(User.handle.ilike(like), User.id.in_(by_account)))
    users = list(await db.scalars(stmt))
    if not users:
        return []
    ids = [u.id for u in users]
    used: dict[uuid.UUID, int] = {
        user_id: int(total)
        for user_id, total in await db.execute(
            select(Message.user_id, func.coalesce(func.sum(Message.size_bytes), 0))
            .where(Message.user_id.in_(ids))
            .group_by(Message.user_id)
        )
    }
    emails: dict[uuid.UUID, list[str]] = {}
    for user_id, email in await db.execute(
        select(Account.user_id, Account.email)
        .where(Account.user_id.in_(ids))
        .order_by(Account.email)
    ):
        emails.setdefault(user_id, []).append(email)
    return [(u, int(used.get(u.id, 0)), emails.get(u.id, [])) for u in users]


async def set_blocked(db: AsyncSession, user: User, blocked: bool) -> None:
    """Block or unblock. Blocking also signs the user out everywhere, and shuts
    their hosted mailbox to every sign-in (mail apps included); it still
    receives mail. The mail server goes first: if it cannot be told, nothing
    changes."""
    from app import hosted

    if hosted.enabled():
        from app.hosted import admin as mail_admin

        await mail_admin.set_blocked(user.handle, blocked)
    if blocked and user.blocked_at is None:
        user.blocked_at = dt.datetime.now(dt.UTC)
        await db.flush()
        await _sign_out_everywhere(user.id)
    elif not blocked:
        user.blocked_at = None


async def _sign_out_everywhere(user_id: uuid.UUID) -> None:
    """Revoke every session, and tell the user's open tabs so they leave at once
    instead of showing already-loaded mail until their next request."""
    await destroy_all_sessions(user_id)
    await trusted_devices.revoke_all(user_id)
    await get_event_bus().publish("user.signed_out", {"user_id": str(user_id)})


async def _plugin_tables(db: AsyncSession) -> list[str]:
    """Every plugin table holding per-user rows (``plg_*`` with a ``user_id``)."""
    rows = await db.execute(
        text(
            "SELECT table_name FROM information_schema.columns "
            "WHERE table_schema = current_schema() AND column_name = 'user_id' "
            "AND table_name LIKE 'plg\\_%' ORDER BY table_name"
        )
    )
    return [r[0] for r in rows]


async def delete_user(db: AsyncSession, user: User) -> None:
    """Delete ``user`` and everything they own. Commits.

    Their hosted mailbox goes first, with all its mail: if the mail server
    cannot be reached nothing is deleted — a mailbox left behind would be
    handed, mail and all, to whoever registers the address next."""
    from app import hosted

    user_id = user.id
    if hosted.enabled():
        from app.hosted import admin as mail_admin

        await mail_admin.delete_account(user.handle)

    # Mail accounts first, one by one: this is what stops their sync runners and
    # tidies threads, exactly as when the user removes an account themselves.
    for account_id in list(await db.scalars(select(Account.id).where(Account.user_id == user_id))):
        await delete_account(db, user_id=user_id, account_id=account_id)
    await db.commit()

    # Plugin tables carry no foreign key to users and sit behind row-level
    # security, so they are cleared from inside the user's own RLS context.
    tables = await _plugin_tables(db)
    if tables:
        async with plugin_session(user_id) as ps:
            for table in tables:
                await ps.execute(
                    text(f'DELETE FROM "{table}" WHERE user_id = :uid'), {"uid": user_id}
                )

    await _sign_out_everywhere(user_id)
    await get_search_backend().delete_user(user_id)

    # The user row: its foreign keys cascade to the rest of core's tables.
    await db.delete(user)
    await db.commit()

    # Whatever blobs are left (drafts' attachments and the like) go with the folder.
    shutil.rmtree(Path(get_settings().blob_root) / str(user_id), ignore_errors=True)