"""FC-3k: admin destructive Celery tasks.

Two long-running ops on the maintenance queue. task_run lifecycle is
captured automatically by FC-3i signals — these tasks just return
their summary dict so it lands in task_run.metadata (via Celery's
result backend) for the dashboard to surface.

Soft/hard time limits inherit the FC-3i recovery sweep: a runaway
task gets killed and flipped to status='timeout' by
recover_stalled_task_runs.
"""
from __future__ import annotations

import logging
from pathlib import Path

from sqlalchemy.exc import DBAPIError, OperationalError

from ..celery_app import celery
from ..services import cleanup_service
from ._sync_engine import sync_session_factory as _sync_session_factory

log = logging.getLogger(__name__)
IMAGES_ROOT = Path("/images")


@celery.task(
    name="backend.app.tasks.admin.delete_artist_cascade_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
)
def delete_artist_cascade_task(self, *, artist_id: int) -> dict:
    """Wraps cleanup_service.delete_artist_cascade. Returns the
    service's summary dict for FC-3i task_run.metadata capture."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        return cleanup_service.delete_artist_cascade(
            session, artist_id=artist_id, images_root=IMAGES_ROOT,
        )


@celery.task(
    name="backend.app.tasks.admin.bulk_delete_images_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=900, time_limit=1200,  # 15 min / 20 min
)
def bulk_delete_images_task(self, *, image_ids: list[int]) -> dict:
    """Wraps cleanup_service.delete_images."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        return cleanup_service.delete_images(
            session, image_ids=image_ids, images_root=IMAGES_ROOT,
        )


# Time-box one chunk well under the soft limit so a large archive back-catalog
# can't run the task into the Celery time limit (or hog the maintenance_long
# lane). The task re-enqueues itself with the resume cursor until the scan is
# exhausted — mirrors normalize_tags_task (operator-asked 2026-06-07: reasonable
# timeout, then re-queue so other work keeps flowing).
_REEXTRACT_CHUNK_SECONDS = 600


@celery.task(
    name="backend.app.tasks.admin.reextract_archive_attachments_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
)
def reextract_archive_attachments_task(self, after_id: int = 0) -> dict:
    """Wraps cleanup_service.reextract_archive_attachments (#713 part 2):
    re-extract PostAttachments that are actually archives but were filed
    opaquely before the magic-byte gate, and link their members to the post.

    Time-boxed + self-resuming: scans attachments after ``after_id`` and, on a
    chunk cut, re-enqueues from where it stopped so a big backlog finishes across
    chunks instead of dying at the soft limit."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        summary = cleanup_service.reextract_archive_attachments(
            session, images_root=IMAGES_ROOT,
            time_budget_seconds=_REEXTRACT_CHUNK_SECONDS, after_id=after_id,
        )
    # More attachments past this chunk's cursor — continue in the next.
    if summary.get("partial") and summary.get("resume_after_id", 0) > after_id:
        log.info(
            "reextract chunk done (%d scanned, %d archives, resume after id %s) "
            "— re-enqueuing to continue",
            summary.get("scanned", 0), summary.get("archives", 0),
            summary["resume_after_id"],
        )
        reextract_archive_attachments_task.delay(summary["resume_after_id"])
    return summary


# Time-box one chunk well under the soft limit so a large back-catalog (the
# first run recases the whole booru vocabulary) can't run the task into the
# Celery time limit — it timed out at 40 min, operator-flagged 2026-06-07. The
# task re-enqueues itself until nothing remains (idempotent — already-canonical
# groups are skipped). 600s keeps each chunk short enough that the recovery
# sweep and other maintenance tasks interleave on the concurrency-1 queue.
_NORMALIZE_CHUNK_SECONDS = 600


@celery.task(
    name="backend.app.tasks.admin.normalize_tags_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
)
def normalize_tags_task(self) -> dict:
    """Wraps tag_service.normalize_existing_tags (#714): Title-Case the
    back-catalog and merge case/whitespace-variant duplicate tags via the
    tested async merge path. Time-boxed + self-resuming so a huge first run
    finishes across chunks instead of timing out. Runs under its own asyncio
    loop + per-task async engine (NullPool), mirroring download_source."""
    import asyncio

    from ..services.tag_service import normalize_existing_tags
    from ._async_session import async_session_factory

    async def _run() -> dict:
        # lock_timeout=30s: a per-group merge repoints FKs across image_tag and
        # series_page; if a statement blocks on a lock (e.g. behind a schema
        # migration holding ACCESS EXCLUSIVE on series_page — the exact wedge that
        # made this task run to the 40-min hard limit with no progress,
        # operator-flagged 2026-06-07), it now fails fast. The per-group handler
        # catches it (rollback + error++) and the loop continues, so one blocked
        # group can't strand the whole chunk.
        async_factory, async_engine = async_session_factory(
            server_settings={"lock_timeout": "30s"}
        )
        try:
            async with async_factory() as session:
                # normalize_existing_tags commits per group internally.
                return await normalize_existing_tags(
                    session, dry_run=False,
                    time_budget_seconds=_NORMALIZE_CHUNK_SECONDS,
                )
        finally:
            await async_engine.dispose()

    summary = asyncio.run(_run())
    # More groups to canonicalize than fit this chunk — continue in the next.
    if summary.get("partial") and summary.get("remaining", 0) > 0:
        log.info(
            "normalize_tags_task chunk done (%d processed, %d remaining) — "
            "re-enqueuing to continue",
            summary.get("groups_processed", 0), summary["remaining"],
        )
        normalize_tags_task.delay()
    return summary


# Time-box one rescan chunk well under the soft limit and re-enqueue from the
# cursor — scoring every post against its artist's series is O(posts) and grows
# with the library (FC-6.3). Mirrors normalize_tags_task.
_SERIES_RESCAN_CHUNK_SECONDS = 600


@celery.task(
    name="backend.app.tasks.admin.rescan_series_suggestions_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
)
def rescan_series_suggestions_task(self, after_post_id: int = 0) -> dict:
    """Score posts against their artist's series and write pending suggestions
    (FC-6.3). Settings-gated; time-boxed + self-resuming from a post-id cursor.
    Per-task async engine (NullPool) under its own asyncio loop, like normalize."""
    import asyncio

    from ..models import ImportSettings
    from ..services.series_match_service import SeriesMatchService
    from ._async_session import async_session_factory

    async def _run() -> dict:
        async_factory, async_engine = async_session_factory()
        try:
            async with async_factory() as session:
                settings = await ImportSettings.load(session)
                if not settings.series_suggest_enabled:
                    return {"skipped": "series suggestions disabled"}
                threshold = settings.series_suggest_threshold
                return await SeriesMatchService(session).rescan(
                    threshold=threshold,
                    time_budget_seconds=_SERIES_RESCAN_CHUNK_SECONDS,
                    after_post_id=after_post_id,
                )
        finally:
            await async_engine.dispose()

    summary = asyncio.run(_run())
    if summary.get("partial") and summary.get("resume_after_id", 0) > after_post_id:
        log.info(
            "rescan_series_suggestions chunk done (%d scanned, %d suggested, "
            "resume after %s) — re-enqueuing",
            summary.get("scanned", 0), summary.get("suggested", 0),
            summary["resume_after_id"],
        )
        rescan_series_suggestions_task.delay(summary["resume_after_id"])
    return summary


@celery.task(
    name="backend.app.tasks.admin.prune_low_confidence_predictions_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=3600, time_limit=4200,  # 60 min / 70 min
)
def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict:
    """One-time #764 backfill: drop tagger_predictions entries below the DB
    store floor (ml_settings.tagger_store_floor) from existing image_record
    rows, and clamp any allowlist min_confidence below the floor up to it.

    The Camie tagger emits ~10k tags; the old 0.05 floor stored the entire
    near-zero tail, bloating image_record's TOAST to ~100 GB. This rewrites
    each row to the new floor. Keyset by id ASC (restart-safe via after_id);
    idempotent — already-pruned rows rewrite to themselves and are skipped.
    Rewriting rows generates bloat, so run VACUUM FULL / pg_repack on
    image_record afterward to return the disk to the OS.

    The keep predicate (confidence >= floor) mirrors Tagger.infer's store
    gate so backfilled rows match what new imports store. Self-resumes on the
    soft time limit (re-enqueues from the last committed id)."""
    from celery.exceptions import SoftTimeLimitExceeded
    from sqlalchemy import select, update

    from ..models import ImageRecord, MLSettings, TagAllowlist

    SessionLocal = _sync_session_factory()
    scanned = 0
    pruned = 0
    clamped = 0
    last_id = after_id
    try:
        with SessionLocal() as session:
            floor = session.execute(
                select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
            ).scalar_one()
            # Clamp allowlist thresholds below the new floor once, on the
            # first pass (#764 consumer #4) — a sub-floor min_confidence can't
            # apply more permissively now that nothing below it is stored.
            if after_id == 0:
                clamped = session.execute(
                    update(TagAllowlist)
                    .where(TagAllowlist.min_confidence < floor)
                    .values(min_confidence=floor)
                ).rowcount or 0
                session.commit()

            while True:
                rows = session.execute(
                    select(ImageRecord.id, ImageRecord.tagger_predictions)
                    .where(ImageRecord.id > last_id)
                    .where(ImageRecord.tagger_predictions.is_not(None))
                    .order_by(ImageRecord.id.asc())
                    .limit(500)
                ).all()
                if not rows:
                    break
                for image_id, preds in rows:
                    scanned += 1
                    if not preds:
                        continue
                    kept = {
                        name: p for name, p in preds.items()
                        if float(p.get("confidence", 0.0)) >= floor
                    }
                    if len(kept) != len(preds):
                        session.execute(
                            update(ImageRecord)
                            .where(ImageRecord.id == image_id)
                            .values(tagger_predictions=kept)
                        )
                        pruned += 1
                session.commit()
                last_id = rows[-1].id  # advance only after commit, for resume
    except SoftTimeLimitExceeded:
        log.warning(
            "prune_low_confidence_predictions soft-limited at id=%s "
            "(scanned=%d pruned=%d) — re-enqueuing", last_id, scanned, pruned,
        )
        prune_low_confidence_predictions_task.delay(last_id)
        return {
            "partial": True, "last_id": last_id,
            "scanned": scanned, "pruned": pruned,
        }

    log.info(
        "prune_low_confidence_predictions complete: floor=%s scanned=%d "
        "pruned=%d allowlist_clamped=%d", floor, scanned, pruned, clamped,
    )
    return {
        "floor": floor, "scanned": scanned, "pruned": pruned,
        "allowlist_clamped": clamped, "last_id": last_id,
    }