"""On-demand state audit and cleanup — /sync command.

Audits all state maps against live tmux windows and reports issues.
A "Fix" button runs cleanup operations and re-audits in place.
Enforcement: closes ghost topics, recreates dead topics, and adopts orphaned windows.

Key functions:
  - sync_command(): /sync command handler
  - handle_sync_fix(): fix button callback — run cleanup, re-audit, edit in place
  - handle_sync_dismiss(): dismiss button callback — remove keyboard
"""

from __future__ import annotations

from typing import TYPE_CHECKING
import asyncio
import contextlib
import re

import structlog
from telegram import (
    CallbackQuery,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    Update,
)
from telegram.error import BadRequest, TelegramError
from .. import window_query
from ..config import config
from ..session import AuditIssue, AuditResult, session_manager
from ..session_map import session_map_sync
from ..telegram_client import PTBTelegramClient, TelegramClient
from ..thread_router import thread_router
from ..tmux_manager import tmux_manager
from ..user_preferences import user_preferences
from .callback_data import CB_SYNC_DISMISS, CB_SYNC_FIX
from .callback_registry import register
from .cleanup import clear_topic_state
from .messaging_pipeline.message_sender import is_thread_gone, safe_edit, safe_reply
from .status.topic_emoji import sync_topic_name

if TYPE_CHECKING:
    from telegram.ext import ContextTypes

logger = structlog.get_logger()

_GHOST_RE = re.compile(r"user:(\d+) thread:(\d+) window:(@\d+)")
_WINDOW_RE = re.compile(r"(@\d+)")

_CATEGORY_LABELS: dict[str, str] = {
    "ghost_binding": "ghost binding (dead window)",
    "dead_topic": "dead topic (window alive, topic deleted)",
    "orphaned_display_name": "orphaned display name",
    "orphaned_group_chat_id": "orphaned group chat ID",
    "stale_window_state": "stale window state",
    "stale_offset": "stale offset entry",
    "display_name_drift": "display name drift",
    "orphaned_window": "unbound tmux window (no topic)",
}


async def _run_audit() -> AuditResult:
    """Fetch live tmux state and run audit."""
    all_windows = await tmux_manager.list_windows()
    live_ids = {w.window_id for w in all_windows}
    live_pairs = [(w.window_id, w.window_name) for w in all_windows]
    return session_manager.audit_state(live_ids, live_pairs)


def _issue_summary_lines(audit: AuditResult) -> list[str]:
    """Build category summary lines from audit issues."""
    category_counts: dict[str, int] = {}
    for issue in audit.issues:
        if issue.category in ("ghost_binding", "dead_topic"):
            continue  # shown in dedicated report lines
        category_counts[issue.category] = category_counts.get(issue.category, 0) + 1

    if category_counts:
        return [
            f"⚠ {count} {_CATEGORY_LABELS.get(cat, cat)}"
            for cat, count in category_counts.items()
        ]
    if audit.total_bindings > 0:
        return ["✓ No orphaned entries", "✓ Tmux display cache in sync"]
    return []


async def _sync_live_topic_names(
    client: TelegramClient, live_ids: set[str] | None = None
) -> None:
    """Best-effort reconciliation of bound live topic titles."""
    if live_ids is None:
        all_windows = await tmux_manager.list_windows()
        live_ids = {w.window_id for w in all_windows}

    for user_id, thread_id, window_id in thread_router.iter_thread_bindings():
        if window_id not in live_ids:
            continue
        chat_id = thread_router.resolve_chat_id(user_id, thread_id)
        if chat_id == user_id:
            continue
        await sync_topic_name(
            client,
            chat_id,
            thread_id,
            thread_router.get_display_name(window_id),
        )


def _format_report(
    audit: AuditResult,
    *,
    fixed_count: int = 0,
    closed_topic_count: int = 0,
    recreated_topic_count: int = 0,
) -> tuple[str, InlineKeyboardMarkup | None]:
    """Build report text and optional keyboard."""
    lines: list[str] = []

    if fixed_count > 0:
        issue_word = "issue" if fixed_count == 1 else "issues"
        lines.append(f"✅ Fixed {fixed_count} {issue_word}\n")
    else:
        lines.append("\U0001f50d State audit\n")

    if closed_topic_count > 0:
        topic_word = "topic" if closed_topic_count == 1 else "topics"
        lines.append(f"ℹ Removed {closed_topic_count} stale {topic_word}")

    if recreated_topic_count > 0:
        topic_word = "topic" if recreated_topic_count == 1 else "topics"
        lines.append(f"ℹ Recreated {recreated_topic_count} {topic_word}")

    # Binding summary
    if audit.total_bindings == 0:
        lines.append("ℹ No topic bindings")
    elif audit.live_binding_count == audit.total_bindings:
        lines.append(f"✓ {audit.total_bindings} topics bound, all windows alive")
    else:
        dead = audit.total_bindings - audit.live_binding_count
        lines.append(
            f"⚠ {dead} ghost binding(s) "
            f"({audit.live_binding_count}/{audit.total_bindings} alive)"
        )

    # Dead topic summary (window alive, but Telegram topic deleted)
    dead_topic_count = sum(1 for i in audit.issues if i.category == "dead_topic")
    if dead_topic_count > 0:
        topic_word = "topic" if dead_topic_count == 1 else "topics"
        lines.append(f"⚠ {dead_topic_count} dead {topic_word} (deleted in Telegram)")

    lines.extend(_issue_summary_lines(audit))

    text = "\n".join(lines)

    # Build keyboard
    fixable = audit.fixable_count
    if fixable > 0:
        issue_word = "issue" if fixable == 1 else "issues"
        keyboard = InlineKeyboardMarkup(
            [
                [
                    InlineKeyboardButton(
                        f"\U0001f527 Fix {fixable} {issue_word}",
                        callback_data=CB_SYNC_FIX,
                    ),
                    InlineKeyboardButton("✕ Dismiss", callback_data=CB_SYNC_DISMISS),
                ]
            ]
        )
    else:
        keyboard = None

    return text, keyboard


async def _remove_topic(client: TelegramClient, chat_id: int, thread_id: int) -> bool:
    """Try to delete a topic, fall back to close. Returns True on success.

    Only "topic not found" BadRequest is treated as success; other BadRequest
    errors (e.g. insufficient rights) fall through to the close fallback.
    """
    try:
        await client.delete_forum_topic(chat_id, thread_id)
        return True
    except BadRequest as e:
        if is_thread_gone(e):
            return True
    except TelegramError:
        pass
    try:
        await client.close_forum_topic(chat_id, thread_id)
        return True
    except TelegramError:
        return False


async def _close_ghost_topics(client: TelegramClient, issues: list[AuditIssue]) -> int:
    """Delete (or close) Telegram topics for ghost bindings.

    Tries ``delete_forum_topic`` first to fully remove the dead topic from the
    sidebar.  Falls back to ``close_forum_topic`` if deletion fails (e.g.
    missing ``can_manage_topics`` or General topic).  Returns count of
    topics successfully deleted/closed.
    """
    closed_count = 0
    for issue in issues:
        if issue.category != "ghost_binding":
            continue
        match = _GHOST_RE.search(issue.detail)
        if not match:
            continue
        user_id = int(match.group(1))
        thread_id = int(match.group(2))
        window_id = match.group(3)
        current_window_id = thread_router.get_window_for_thread(user_id, thread_id)
        if current_window_id != window_id:
            continue
        chat_id = thread_router.resolve_chat_id(user_id, thread_id)
        topic_removed = False
        if chat_id == user_id:
            logger.warning(
                "No group chat_id for ghost topic thread=%d, skipping close",
                thread_id,
            )
        else:
            topic_removed = await _remove_topic(client, chat_id, thread_id)
            if not topic_removed:
                logger.warning(
                    "Failed to delete/close ghost topic thread=%d window=%s",
                    thread_id,
                    window_id,
                )
        if topic_removed or chat_id == user_id:
            try:
                await clear_topic_state(
                    user_id, thread_id, client=client, window_id=window_id
                )
                thread_router.unbind_thread(user_id, thread_id)
                if topic_removed:
                    closed_count += 1
            except OSError, TelegramError:
                logger.exception(
                    "Failed to clean up ghost binding thread=%d window=%s",
                    thread_id,
                    window_id,
                )
    return closed_count


async def _adopt_orphaned_windows(
    client: TelegramClient, issues: list[AuditIssue]
) -> None:
    """Create Telegram topics for unbound tmux windows."""
    # Lazy: bidirectional cycle — topic_orchestration.adopt_unbound_windows
    # also lazy-imports _adopt_orphaned_windows from this module.  Either
    # side must remain lazy until one is split into a third module.
    # Lazy: session_monitor / topic_orchestration cycle through window-creation flow
    from ..session_monitor import NewWindowEvent

    # Lazy: session_monitor / topic_orchestration cycle through window-creation flow
    from .topics.topic_orchestration import handle_new_window as _handle_new_window

    for issue in issues:
        if issue.category != "orphaned_window":
            continue
        match = _WINDOW_RE.search(issue.detail)
        if not match:
            continue
        window_id = match.group(1)
        view = window_query.view_window(window_id)
        name = (view.window_name if view else "") or thread_router.get_display_name(
            window_id
        )
        event = NewWindowEvent(
            window_id=window_id,
            session_id=view.session_id if view else "",
            window_name=name,
            cwd=view.cwd if view else "",
        )
        try:
            await _handle_new_window(event, client)
        except TelegramError:
            logger.exception("Failed to adopt orphaned window %s", window_id)


async def _probe_dead_topics(client: TelegramClient) -> list[AuditIssue]:
    """Probe Telegram topics for all live bindings, return dead_topic issues.

    Sends a silent zero-width-space message to each thread and deletes it
    immediately. ``send_chat_action`` does NOT validate thread existence —
    only ``send_message`` reliably throws "thread not found" for deleted topics.
    """
    bindings = [
        (uid, tid, wid, thread_router.resolve_chat_id(uid, tid))
        for uid, tid, wid in thread_router.iter_thread_bindings()
    ]
    # Only probe bindings with a group chat (chat_id != user_id)
    bindings = [(uid, tid, wid, cid) for uid, tid, wid, cid in bindings if cid != uid]
    if not bindings:
        return []

    sem = asyncio.Semaphore(5)  # limit concurrent Telegram API calls

    async def _probe_one(
        user_id: int, thread_id: int, window_id: str, chat_id: int
    ) -> AuditIssue | None:
        async with sem:
            try:
                msg = await client.send_message(
                    chat_id,
                    "​",  # zero-width space — invisible
                    message_thread_id=thread_id,
                    disable_notification=True,
                )
                # Topic exists — clean up probe message
                with contextlib.suppress(TelegramError):
                    await client.delete_message(chat_id, msg.message_id)
            except BadRequest as exc:
                if is_thread_gone(exc):
                    display = thread_router.get_display_name(window_id)
                    return AuditIssue(
                        category="dead_topic",
                        detail=f"user:{user_id} thread:{thread_id} window:{window_id} ({display})",
                        fixable=True,
                    )
            except TelegramError:
                pass  # network error, skip — not a dead topic
        return None

    results = await asyncio.gather(
        *(_probe_one(*b) for b in bindings), return_exceptions=True
    )
    issues: list[AuditIssue] = []
    for r in results:
        if isinstance(r, AuditIssue):
            issues.append(r)
        elif isinstance(r, BaseException):
            logger.error("Unexpected error probing dead topics", exc_info=r)
    return issues


async def _recreate_dead_topics(
    client: TelegramClient, issues: list[AuditIssue]
) -> int:
    """Unbind dead topics and recreate them via _handle_new_window.

    Returns count of successfully recreated topics.
    """
    # Lazy: same sync_command ↔ topic_orchestration cycle as
    # _adopt_orphaned_windows.
    # Lazy: session_monitor / topic_orchestration cycle through window-creation flow
    from ..session_monitor import NewWindowEvent

    # Lazy: session_monitor / topic_orchestration cycle through window-creation flow
    from .topics.topic_orchestration import handle_new_window as _handle_new_window

    recreated = 0
    for issue in issues:
        if issue.category != "dead_topic":
            continue
        match = _GHOST_RE.search(issue.detail)
        if not match:
            continue
        user_id = int(match.group(1))
        thread_id = int(match.group(2))
        window_id = match.group(3)
        current_window_id = thread_router.get_window_for_thread(user_id, thread_id)
        if current_window_id != window_id:
            continue

        view = window_query.view_window(window_id)
        name = (view.window_name if view else "") or thread_router.get_display_name(
            window_id
        )
        event = NewWindowEvent(
            window_id=window_id,
            session_id=view.session_id if view else "",
            window_name=name,
            cwd=view.cwd if view else "",
        )

        # Preserve group_chat_id before unbinding — unbind_thread deletes it,
        # but _handle_new_window needs it to know which chat to create the topic in.
        chat_id = thread_router.resolve_chat_id(user_id, thread_id)

        # Unbind THEN recreate — must unbind first so _handle_new_window
        # doesn't skip the window as "already bound".  On failure, restore.
        thread_router.unbind_thread(user_id, thread_id)

        # Inject a temporary in-memory-only group_chat_id so _handle_new_window
        # can discover the chat.  Direct dict mutation avoids _save_state() —
        # if the process crashes, the placeholder won't persist to state.json.
        _placeholder_key = f"{user_id}:0"
        if chat_id != user_id:
            thread_router.group_chat_ids[_placeholder_key] = chat_id

        try:
            await _handle_new_window(event, client)
            recreated += 1
        except TelegramError, OSError:
            logger.exception("Failed to recreate topic for window %s", window_id)
            # Restore binding so the window isn't orphaned
            thread_router.bind_thread(user_id, thread_id, window_id, window_name=name)
            if chat_id != user_id:
                thread_router.set_group_chat_id(user_id, thread_id, chat_id)
        finally:
            thread_router.group_chat_ids.pop(_placeholder_key, None)
    return recreated


async def sync_command(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle /sync — audit state and show report."""
    user = update.effective_user
    if not user or not update.message:
        return

    if not config.is_user_allowed(user.id):
        await safe_reply(update.message, "You are not authorized to use this bot.")
        return

    client = PTBTelegramClient(update.get_bot())
    await _sync_live_topic_names(client)
    audit = await _run_audit()
    # Probe Telegram topics for live bindings (async, needs client)
    dead_issues = await _probe_dead_topics(client)
    audit.issues.extend(dead_issues)
    text, keyboard = _format_report(audit)
    await safe_reply(update.message, text, reply_markup=keyboard)


async def handle_sync_fix(query: CallbackQuery) -> None:
    """Run all fix operations, re-audit, and edit message in place."""
    # Single list_windows call — reused for both audit and fix
    all_windows = await tmux_manager.list_windows()
    live_ids = {w.window_id for w in all_windows}
    live_pairs = [(w.window_id, w.window_name) for w in all_windows]

    # Audit before fixing to count fixable issues
    client = PTBTelegramClient(query.get_bot())
    pre_audit = session_manager.audit_state(live_ids, live_pairs)
    dead_issues = await _probe_dead_topics(client)
    pre_audit.issues.extend(dead_issues)

    # Run state cleanup operations
    try:
        session_manager.sync_display_names(live_pairs)
        session_manager.prune_stale_state(live_ids)
        session_map_sync.prune_session_map(live_ids)
        session_manager.prune_stale_window_states(live_ids)
        bound_ids: set[str] = {
            wid for _, _, wid in thread_router.iter_thread_bindings()
        }
        state_ids = set(window_query.iter_window_ids())
        user_preferences.prune_stale_offsets(live_ids | bound_ids | state_ids)
    except OSError:
        logger.exception("Error during sync fix operations")

    await _sync_live_topic_names(client, live_ids)

    # Enforcement: adopt orphans first so stale same-name topics can be rebound.
    await _adopt_orphaned_windows(client, pre_audit.issues)
    closed_count = await _close_ghost_topics(client, pre_audit.issues)
    recreated_count = await _recreate_dead_topics(client, pre_audit.issues)

    # Re-audit and compute actual fixed count (handles partial failures).
    # No skip_threads here: successful recreations use a new thread_id (old
    # one is unbound and won't be probed), while failed ones restore the old
    # binding and must be re-probed to avoid inflating actual_fixed.
    post_audit = await _run_audit()
    post_dead = await _probe_dead_topics(client)
    post_audit.issues.extend(post_dead)
    actual_fixed = pre_audit.fixable_count - post_audit.fixable_count
    text, keyboard = _format_report(
        post_audit,
        fixed_count=actual_fixed,
        closed_topic_count=closed_count,
        recreated_topic_count=recreated_count,
    )
    await safe_edit(query, text, reply_markup=keyboard)


async def handle_sync_dismiss(query: CallbackQuery) -> None:
    """Remove keyboard from sync message."""
    original_text = getattr(query.message, "text", None) if query.message else None
    await safe_edit(query, original_text or "Dismissed", reply_markup=None)


@register(CB_SYNC_FIX, CB_SYNC_DISMISS)
async def _dispatch(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None:
    query = update.callback_query
    if not query or not query.data:
        return

    if query.data == CB_SYNC_FIX:
        await handle_sync_fix(query)
        await query.answer("Fixed")
    elif query.data == CB_SYNC_DISMISS:
        await handle_sync_dismiss(query)
        await query.answer("Dismissed")
