"""Mailbox sync Taskiq tasks — Outlook pull + triage + ingest on own stream."""

from __future__ import annotations

import logging
import os
from datetime import datetime,timezone

import redis.asyncio as redis
from dotenv import load_dotenv
from taskiq import TaskiqEvents

from db_setup import session_scope
from inbox.models import MailboxSyncRun
from inbox.views import Email
from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY
from taskiq_management.mailbox_sync_broker_setup import MAILBOX_SYNC_QUEUE_NAME,mailbox_sync_broker
from taskiq_management.middleware import PermanentTaskError

load_dotenv()

logger=logging.getLogger("inbox.mailbox_sync")
REDIS_URL=os.getenv("REDIS_URL","redis://localhost:6379/0")
_LOCK_KEY="inbox:mailbox_sync:lock"
_LOCK_TTL=900
_CONSUMER_GROUP=os.getenv("TASKIQ_CONSUMER_GROUP","taskiq")
# taskiq-redis listen() skips XAUTOCLAIM while this key exists. redis-py Lock has
# no TTL by default, so SIGKILL (compose rebuild) leaves it forever and the Sync
# button stays on a running row nobody will finish.
_AUTOCLAIM_KEY=f"autoclaim:{_CONSUMER_GROUP}:{MAILBOX_SYNC_QUEUE_NAME}"


@mailbox_sync_broker.on_event(TaskiqEvents.WORKER_STARTUP)
async def _drop_stale_autoclaim(_state) -> None:
    client=redis.from_url(REDIS_URL,decode_responses=True)
    try:
        if await client.delete(_AUTOCLAIM_KEY):
            logger.warning("dropped stale autoclaim lock %s",_AUTOCLAIM_KEY)
    finally:
        await client.aclose()


async def _fail(run_id:str,error:str) -> dict:
    async with session_scope() as session:
        await MailboxSyncRun.update_run(session,run_id,{
            "status":"failed",
            "error":error,
            "finished_at":datetime.now(timezone.utc),
        })
    return {"status":"failed","error":error}


@mailbox_sync_broker.task(
    task_name="inbox.sync_mailbox",
    retry_on_error=True,
    max_retries=MAX_RETRIES,
    delay=RETRY_DELAY,
)
async def sync_mailbox(run_id:str) -> dict:
    if not run_id or not str(run_id).strip():
        raise PermanentTaskError("run_id is required")
    run_id=str(run_id).strip()

    client=redis.from_url(REDIS_URL,decode_responses=True)
    try:
        acquired=await client.set(_LOCK_KEY,run_id,nx=True,ex=_LOCK_TTL)
        if not acquired:
            holder=await client.get(_LOCK_KEY)
            # Crash/restart redelivers the same run_id while the TTL lock is
            # still set. Failing that as "another sync" strands the lock until
            # expiry and every later click also bounces.
            if holder==run_id:
                await client.expire(_LOCK_KEY,_LOCK_TTL)
                logger.warning("mailbox sync %s reclaimed its own stale lock",run_id)
            else:
                logger.warning("mailbox sync %s skipped: lock held by %s",run_id,holder)
                return await _fail(run_id,"another mailbox sync is already running")

        try:
            async with session_scope() as session:
                row=await MailboxSyncRun.get_by_id(session,run_id)
                if not row:
                    raise PermanentTaskError(f"sync run {run_id} not found")
                if row.status in ("completed","failed"):
                    logger.info("mailbox sync %s already %s, skipping",run_id,row.status)
                    return {"status":row.status,"error":row.error}
                await MailboxSyncRun.update_run(session,run_id,{
                    "status":"running",
                    "started_at":datetime.now(timezone.utc),
                    "error":None,
                })
                top=row.top or 100
                skip=row.skip or 0
                test_on=True if row.test_on is None else bool(row.test_on)

            async with session_scope() as session:
                service=Email(session=session)
                if not service.token:
                    return await _fail(run_id,"EMAIL_API_TOKEN is not configured")

                async def on_progress(processed,expected,entries):
                    ingested=sum(1 for e in entries if e.get("status") in ("ingested","known"))
                    skipped=sum(1 for e in entries if e.get("status")=="skipped")
                    errors=sum(1 for e in entries if e.get("status")=="error")
                    await MailboxSyncRun.update_run(session,run_id,{
                        "entries":list(entries),
                        "triage":{
                            "expected":expected,
                            "processed":processed,
                            "ingested":ingested,
                            "skipped":skipped,
                            "errors":errors,
                            "total":expected,
                        },
                    })

                try:
                    summary=await service.run_mailbox_sync_page(
                        top=top,skip=skip,test_on=test_on,on_progress=on_progress,
                    )
                except Exception as e:
                    logger.exception("mailbox sync failed for run %s",run_id)
                    return await _fail(run_id,str(e))

                await MailboxSyncRun.update_run(session,run_id,{
                    "status":"completed",
                    "entries":summary["entries"],
                    "triage":summary["triage"],
                    "error":None,
                    "finished_at":datetime.now(timezone.utc),
                })
                return {
                    "status":"completed",
                    "triage":summary["triage"],
                    "entries":len(summary["entries"]),
                }
        finally:
            current=await client.get(_LOCK_KEY)
            if current==run_id:
                await client.delete(_LOCK_KEY)
    finally:
        await client.aclose()