"""
Celery tasks for routine maintenance operations.
"""

from __future__ import annotations

import datetime as dt
import logging
import os
import tempfile
import time
from pathlib import Path

from ..celery_app import celery_app

logger = logging.getLogger(__name__)

# Files older than this many days (by mtime) are deleted from the attachment cache.
ATTACHMENT_CACHE_TTL_DAYS = 30

# ── Attachment prefetch (background cache warming) ──
# Warm the disk cache for CUSTOMER (inbound) attachments received recently, so
# the first time an agent opens them they hit the fast cache path (~15ms) instead
# of a ~15s cold Playwright fetch. Newest-first and sequential: the most recent
# customer attachment (most likely to be opened next) warms first. See
# enqueue_attachment_prefetch + prefetch_recent_attachments below.
PREFETCH_WINDOW_DAYS = 7        # only attachments from messages this recent
PREFETCH_MAX_PER_RUN = 25       # cap per tenant per run (bounds worker time)
PREFETCH_LOCK_TTL = 1800        # seconds; guards against overlapping runs


def _prefetch_redis():
    """Lazy Redis client for the per-tenant prefetch lock; None if unavailable."""
    try:
        import redis  # redis-py — already a dependency (Celery broker)
        return redis.Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"))
    except Exception as exc:  # pragma: no cover
        logger.warning("[prefetch] Redis unavailable (%s); running without lock", exc)
        return None


def _fetch_attachment_bytes(url: str) -> bytes | None:
    """Fetch attachment bytes via headless Playwright (bypasses Cloudflare).

    Replicates main.py's _playwright_fetch_attachment (a create_app closure that
    can't be imported here). Keep in sync with it — notably wait_until='networkidle'
    is required: the Addons URL is an HTML page whose file arrives via a
    JS-triggered download after load.
    """
    from playwright.sync_api import sync_playwright

    url = url.replace(
        'addons.prestashop.com/customer-messages-detail.php',
        'addons.prestashop.com/en/customer-messages-detail.php',
    )
    pw = sync_playwright().start()
    try:
        br = pw.chromium.launch(
            headless=True,
            args=['--disable-blink-features=AutomationControlled'],
        )
        ctx = br.new_context(
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            accept_downloads=True,
        )
        page = ctx.new_page()
        page.add_init_script('Object.defineProperty(navigator, "webdriver", {get: () => undefined})')

        _downloads = []
        page.on('download', lambda d: _downloads.append(d))

        try:
            resp = page.goto(url, wait_until='networkidle', timeout=30000)
            if resp and resp.status == 200:
                body = resp.body()
                ct = resp.headers.get('content-type', '')
                if ct.startswith('image/') or ct == 'application/pdf' or ct.startswith('text/plain'):
                    return body
                if len(body) > 100 and not body[:15].startswith(b'<!DOCTYPE'):
                    return body
        except Exception:
            page.wait_for_timeout(5000)  # "Download is starting" — let it complete

        if _downloads:
            dl = _downloads[0]
            tmp = tempfile.mktemp()
            dl.save_as(tmp)
            body = Path(tmp).read_bytes()
            Path(tmp).unlink(missing_ok=True)
            if body and len(body) > 50:
                return body

        return None
    finally:
        try:
            ctx.close()
            br.close()
        except Exception:
            pass
        pw.stop()


@celery_app.task
def prune_attachment_cache() -> dict:
    """
    Delete attachment cache files older than ATTACHMENT_CACHE_TTL_DAYS days.

    The cache directory mirrors main.py: Path(app.instance_path) / 'attachment_cache'.
    The serve route (attachment_view) already handles a missing cached_path by
    re-fetching via Playwright, so deleting stale files is always safe — no DB
    updates are needed.

    Runs daily at 04:30 UTC via Celery beat (see celery_app.py).
    """
    from ..celery_app import get_flask_app

    app = get_flask_app()

    cache_dir = Path(app.instance_path) / "attachment_cache"

    if not cache_dir.is_dir():
        logger.info(
            "[prune_attachment_cache] Cache directory does not exist: %s — nothing to prune",
            cache_dir,
        )
        return {"status": "ok", "deleted": 0, "bytes_freed": 0, "errors": 0}

    cutoff = time.time() - ATTACHMENT_CACHE_TTL_DAYS * 86400
    deleted = 0
    bytes_freed = 0
    errors = 0

    for entry in cache_dir.iterdir():
        if not entry.is_file():
            # Skip sub-directories (none expected, but be defensive).
            continue
        try:
            mtime = entry.stat().st_mtime
            if mtime < cutoff:
                size = entry.stat().st_size
                entry.unlink()
                deleted += 1
                bytes_freed += size
                logger.debug(
                    "[prune_attachment_cache] Deleted %s (%.1f KB, mtime %s)",
                    entry.name,
                    size / 1024,
                    time.strftime("%Y-%m-%d", time.gmtime(mtime)),
                )
        except OSError as exc:
            logger.warning(
                "[prune_attachment_cache] Could not process %s: %s", entry.name, exc
            )
            errors += 1

    logger.info(
        "[prune_attachment_cache] Done. deleted=%d bytes_freed=%d errors=%d",
        deleted,
        bytes_freed,
        errors,
    )
    return {"status": "ok", "deleted": deleted, "bytes_freed": bytes_freed, "errors": errors}


@celery_app.task(bind=True, max_retries=1, default_retry_delay=300)
def prefetch_recent_attachments(self, org_id: int, schema: str) -> dict:
    """Warm the attachment cache for one tenant: fetch recent CUSTOMER (inbound)
    attachments, newest-first, SEQUENTIALLY, skipping any already on disk.

    Sequential by design: the ordering guarantee ("newest customer attachment
    warms first") only holds one-at-a-time, and each fetch spins a full headless
    Chromium (~15s, heavy) — parallel bursts would also trip Cloudflare. Capped
    at PREFETCH_MAX_PER_RUN; a Redis lock prevents overlapping runs per tenant.
    """
    from flask import g
    from ..celery_app import get_flask_app
    from ..db import session_scope, set_worker_tenant_schema
    from ..services.attachment_cache import attachment_cache_path
    from ..models import AddonsAttachments, AddonsMessages

    r = _prefetch_redis()
    lock_key = f"prefetch_lock:{schema}"
    if r is not None:
        try:
            if not r.set(lock_key, "1", nx=True, ex=PREFETCH_LOCK_TTL):
                logger.info("[prefetch] org=%s already running, skipping", org_id)
                return {"status": "skipped", "detail": "locked"}
        except Exception:
            r = None  # lock unavailable — proceed without it

    set_worker_tenant_schema(schema)
    app = get_flask_app()
    cache_dir = Path(app.instance_path) / "attachment_cache"
    cache_dir.mkdir(parents=True, exist_ok=True)
    cutoff = dt.datetime.utcnow() - dt.timedelta(days=PREFETCH_WINDOW_DAYS)

    fetched = skipped = failed = 0
    try:
        with app.app_context():
            g.tenant_schema = schema

            # Candidate inbound attachments, newest message first.
            with session_scope() as session:
                rows = (
                    session.query(
                        AddonsAttachments.id,
                        AddonsAttachments.filename,
                        AddonsAttachments.provider_url,
                        AddonsAttachments.cached_path,
                    )
                    .join(AddonsMessages, AddonsAttachments.message_id == AddonsMessages.id)
                    .filter(AddonsMessages.direction == "inbound")
                    .filter(AddonsMessages.created_at >= cutoff)
                    .order_by(AddonsMessages.created_at.desc())
                    .all()
                )
                candidates = [
                    {"id": a_id, "filename": fn, "provider_url": url, "cached_path": cp}
                    for (a_id, fn, url, cp) in rows
                ]

            for c in candidates:
                if fetched >= PREFETCH_MAX_PER_RUN:
                    break
                # Must match the route's key exactly — see services/attachment_cache.
                # A flat {id}_{filename} key here leaked attachments between tenants,
                # and this beat task warms the cache unattended for every org, so it
                # was the thing actively widening the collision surface.
                cache_file = attachment_cache_path(cache_dir, schema, c["id"], c["filename"])
                if (c["cached_path"] and Path(c["cached_path"]).exists()) or cache_file.exists():
                    skipped += 1
                    continue
                if not c["provider_url"]:
                    continue
                try:
                    body = _fetch_attachment_bytes(c["provider_url"])
                except Exception as exc:
                    logger.warning("[prefetch] fetch error att=%s: %s", c["id"], exc)
                    body = None
                if not body:
                    failed += 1
                    continue
                try:
                    cache_file.write_bytes(body)
                    with session_scope() as s2:
                        att = s2.query(AddonsAttachments).filter_by(id=c["id"]).one_or_none()
                        if att:
                            att.cached_path = str(cache_file)
                    fetched += 1
                except OSError as exc:
                    logger.warning("[prefetch] write error att=%s: %s", c["id"], exc)
                    failed += 1
    finally:
        if r is not None:
            try:
                r.delete(lock_key)
            except Exception:
                pass

    logger.info(
        "[prefetch] org=%s fetched=%d skipped=%d failed=%d", org_id, fetched, skipped, failed
    )
    return {"status": "ok", "fetched": fetched, "skipped": skipped, "failed": failed}


@celery_app.task
def enqueue_attachment_prefetch() -> dict:
    """Beat dispatcher: fire a per-tenant attachment prefetch for every eligible
    org (has an Addons key or is the internal owner; not suspended). Mirrors the
    eligibility of enqueue_periodic_syncs so only tenants with synced data run.
    """
    from ..db import session_scope
    from ..public_models import Organization

    enqueued = 0
    with session_scope() as session:
        orgs = (
            session.query(Organization)
            .filter(Organization.is_suspended.is_(False))
            .filter(
                Organization.addons_api_key_encrypted.isnot(None)
                | (Organization.plan == "internal")
            )
            .all()
        )
        for org in orgs:
            prefetch_recent_attachments.delay(org.id, org.schema_name)
            enqueued += 1

    logger.info("[enqueue_attachment_prefetch] enqueued=%d", enqueued)
    return {"status": "ok", "enqueued": enqueued}