"""Scheduled cleanup for unreferenced browser-direct uploads."""

from __future__ import annotations

import logging
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.core.config import settings
from app.core.storage import delete_object_if_unchanged, resolve_storage_config
from app.integrations.models import StorageUploadGrant
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

_CLAIM_STATUS = "cleanup_claimed"
_RETRY_DELAY = timedelta(hours=6)


@dataclass(frozen=True)
class _CleanupClaim:
    grant_id: uuid.UUID
    attempt: int
    object_key: str
    content_type: str
    size_bytes: int
    checksum_sha256: str
    reclaimed: bool


def _eligible_claim(now: datetime, cutoff: datetime):
    retry_cutoff = now - _RETRY_DELAY
    return and_(
        StorageUploadGrant.created_at <= cutoff,
        # **Never delete an object whose grant is still claimable.**
        #
        # This used to read ``created_at`` alone, which was safe only by
        # coincidence: every purpose's TTL was 600s and the retention floor
        # is one hour, so anything old enough to sweep was necessarily
        # already unclaimable. AI Labs now issues 4h grants (it uploads at
        # file-pick and claims at send, so the gap is "how long the
        # employee spends typing"), and at a retention setting below that —
        # 1-3 hours is an ordinary storage-cost tuning value this job
        # accepts — the sweep would delete the object out from under a
        # valid grant. The employee's send then fails with
        # ``UPLOAD_NOT_FOUND`` on a file they can still see attached.
        #
        # Reading the grant's own expiry makes the ordering an invariant
        # instead of a coincidence, for any TTL anyone sets later.
        StorageUploadGrant.expires_at <= now,
        or_(
            and_(
                StorageUploadGrant.status == "issued",
                or_(
                    StorageUploadGrant.cleanup_attempted_at.is_(None),
                    StorageUploadGrant.cleanup_attempted_at <= retry_cutoff,
                ),
            ),
            and_(
                StorageUploadGrant.status == _CLAIM_STATUS,
                StorageUploadGrant.cleanup_attempted_at <= retry_cutoff,
            ),
        ),
    )


async def _claim_next(
    db: AsyncSession, *, cutoff: datetime,
) -> _CleanupClaim | None:
    """Claim one row and commit before any object-storage call."""
    now = datetime.now(UTC)
    row = (await db.execute(
        select(StorageUploadGrant)
        .where(_eligible_claim(now, cutoff))
        .order_by(StorageUploadGrant.created_at, StorageUploadGrant.id)
        .limit(1)
        .with_for_update(skip_locked=True)
    )).scalar_one_or_none()
    if row is None:
        await db.commit()
        return None

    reclaimed = row.status == _CLAIM_STATUS
    row.status = _CLAIM_STATUS
    row.cleanup_attempted_at = now
    row.cleanup_attempts = (row.cleanup_attempts or 0) + 1
    claim = _CleanupClaim(
        grant_id=row.id,
        attempt=row.cleanup_attempts,
        object_key=row.object_key,
        content_type=row.content_type,
        size_bytes=row.declared_size_bytes,
        checksum_sha256=row.checksum_sha256,
        reclaimed=reclaimed,
    )
    await db.commit()
    return claim


async def _finish_claim(
    db: AsyncSession,
    claim: _CleanupClaim,
    *,
    deleted: bool,
) -> bool:
    """Finalize or release one claim using ``cleanup_attempts`` as a fence."""
    row = (await db.execute(
        select(StorageUploadGrant)
        .where(
            StorageUploadGrant.id == claim.grant_id,
            StorageUploadGrant.status == _CLAIM_STATUS,
            StorageUploadGrant.cleanup_attempts == claim.attempt,
        )
        .with_for_update(skip_locked=True)
    )).scalar_one_or_none()
    if row is None:
        await db.commit()
        return False
    if deleted:
        row.status = "deleted"
        row.deleted_at = datetime.now(UTC)
    else:
        # Keep cleanup_attempted_at at the claim time. The retry delay keeps a
        # degraded S3 endpoint from being hammered again on the next tick.
        row.status = "issued"
    await db.commit()
    return True


@register_scheduled_job(
    key="storage.orphan_upload_cleanup",
    display_name="对象存储 · 孤儿上传清理",
    description="每小时删除超过保留期仍未被业务对象引用的直传文件。",
)
async def cleanup_orphan_uploads(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    """Claim, externally delete, then finalize each orphan in short phases."""
    retention_hours = max(
        1,
        min(
            int(config.get("retention_hours", settings.UPLOAD_ORPHAN_RETENTION_HOURS)),
            168,
        ),
    )
    batch_size = max(1, min(int(config.get("batch_size", 100)), 500))
    cutoff = datetime.now(UTC) - timedelta(hours=retention_hours)

    # Snapshot the destination in a short read transaction. The blocking S3
    # phases below receive plain values and cannot accidentally keep even a
    # read-only DB transaction open.
    storage_config = await resolve_storage_config(db)
    await db.commit()

    claimed = 0
    reclaimed = 0
    deleted = 0
    already_missing = 0
    failed = 0
    changed = 0
    lost_claims = 0
    for _ in range(batch_size):
        claim = await _claim_next(db, cutoff=cutoff)
        if claim is None:
            break
        claimed += 1
        reclaimed += int(claim.reclaimed)

        try:
            outcome = await delete_object_if_unchanged(
                bucket=storage_config["bucket"],
                region=storage_config["region"],
                object_key=claim.object_key,
                expected_size_bytes=claim.size_bytes,
                expected_content_type=claim.content_type,
                expected_checksum_sha256=claim.checksum_sha256,
            )
        except Exception:
            failed += 1
            logger.exception(
                "storage orphan delete failed grant=%s attempt=%d",
                claim.grant_id,
                claim.attempt,
            )
            if not await _finish_claim(db, claim, deleted=False):
                lost_claims += 1
            continue

        if outcome == "changed":
            changed += 1
            failed += 1
            logger.warning(
                "storage orphan changed during cleanup grant=%s attempt=%d",
                claim.grant_id,
                claim.attempt,
            )
            if not await _finish_claim(db, claim, deleted=False):
                lost_claims += 1
            continue

        if not await _finish_claim(db, claim, deleted=True):
            lost_claims += 1
            continue
        deleted += 1
        already_missing += int(outcome == "missing")

    detail = {
        "candidate_count": claimed,
        "claimed": claimed,
        "reclaimed": reclaimed,
        "deleted": deleted,
        "already_missing": already_missing,
        "failed": failed,
        "changed": changed,
        "lost_claims": lost_claims,
        # Kept for dashboard/backward compatibility. A referenced row is no
        # longer selected or blocked; the claim CAS simply skips it.
        "skipped_referenced": 0,
        "retention_hours": retention_hours,
    }
    await AuditService(db).log(
        trace_id=f"scheduled:storage.orphan_upload_cleanup:{datetime.now(UTC).isoformat()}",
        actor_type="system",
        actor_user_id=None,
        action="storage.upload.orphan_cleanup",
        resource_type="storage_upload_grant",
        resource_id=None,
        status="partial" if failed or lost_claims else "success",
        detail=detail,
    )
    return detail