"""Background recovery scanner — runs every minute via arq cron (see ``WorkerSettings.cron_jobs``).

Detects and self-heals three classes of infrastructure drift that can leave
projects permanently broken without operator intervention:

1. **Setup failures** — projects stuck in ``status=error`` because the GitHub
   push raced with repo provisioning (or any other transient setup error).
   Action: reset to ``creating`` + re-enqueue ``setup_project_repo`` (idempotent).

2. **Missing live hostname** — projects that completed setup but whose initial
   production deploy never ran (hosting_bootstrap disabled, enqueue hiccup, or
   deploy task exhausted retries).
   Action: create a fresh recovery deploy row + enqueue ``run_project_deploy``.

3. **Missing preview hostname** — projects where ``active_preview_id`` is set
   but ``preview_hostname`` was never written (fixed going forward by the patch
   in provision_preview_for_revision, but this cleans up pre-existing rows).
   Action: compute hostname from ``project_ref`` + apex and write directly.

Guards:
- Per-scan caps (``_MAX_PER_SCAN``) prevent thundering-herd on a large backlog.
- Age windows avoid racing with in-flight tasks.
- Recovery attempts are limited by marking deploys with ``recovery_attempt``
  metadata and refusing to create more than ``_MAX_DEPLOY_RECOVERY_ATTEMPTS``.
"""
from __future__ import annotations

import structlog

logger = structlog.get_logger()

_MAX_PER_SCAN = 20
_MAX_DEPLOY_RECOVERY_ATTEMPTS = 3

# Age thresholds (seconds)
_SETUP_MIN_AGE_S = 5 * 60         # 5 min  — let in-flight task finish first
_SETUP_MAX_AGE_S = 2 * 60 * 60   # 2 hrs  — beyond this, needs manual attention
_STALE_CREATING_AGE_S = 15 * 60  # 15 min — Phase 1+2 together finish in <10 min; longer gap avoids racing in-flight tasks
# Legacy: single-phase setup applies boilerplate in-project_setup; this gap is rare.
_STALLED_BOILERPLATE_AGE_S = 3 * 60
_DEPLOY_MIN_AGE_S = 10 * 60     # 10 min — deploy task needs time to poll CF Pages
_DEPLOY_MAX_AGE_S = 6 * 60 * 60 # 6 hrs

# Stale in-flight thresholds — records older than these are never going to complete
_STALE_DEPLOY_AGE_S = 45 * 60        # 45 min — deploy task max_tries * 30s ≈ 20 min
_STALE_PREVIEW_PROVISIONING_AGE_S = 30 * 60  # 30 min — GA build + provision should finish in < 15 min
_STALE_PREVIEW_RUNNING_AGE_S = 7 * 24 * 60 * 60  # 7 days — orphaned running previews


async def run_recovery_scan(ctx: dict) -> dict:
    """Entry point called by arq cron every minute."""
    db_factory = ctx.get("db_factory")
    if db_factory is None:
        from src.core.database import async_session_factory
        db_factory = async_session_factory

    ctx.setdefault("db_factory", db_factory)

    results: dict[str, int] = {
        "stale_deploys_expired": 0,
        "stale_previews_expired": 0,
        "manifest_runs_expired": 0,
        "codegen_stale_marked_failed": 0,
        "setup_requeued": 0,
        "boilerplate_requeued": 0,
        "deploy_created": 0,
        "preview_hostname_fixed": 0,
        "cf_synced": 0,
        "storage_ingest_requeued": 0,
    }

    # Pass 0: expire stale in-flight records FIRST so later passes aren't blocked by them
    try:
        expired = await _expire_stale_in_flight_records(db_factory)
        results["stale_deploys_expired"] = expired["deploys"]
        results["stale_previews_expired"] = expired["previews"]
    except Exception as exc:
        logger.error("recovery_scan_expire_pass_failed", error=str(exc), exc_info=True)

    # Pass 0c: expire manifest runs past their TTL
    try:
        results["manifest_runs_expired"] = await _expire_stale_manifest_runs(db_factory)
    except Exception as exc:
        logger.error("recovery_scan_manifest_runs_failed", error=str(exc), exc_info=True)

    # Pass 0b: codegen runs stuck ``running`` (AI timeout / worker loss) → failed + WS so UI stops "AI is coding…"
    try:
        from src.workers.tasks.codegen_stale_sweep import sweep_stale_codegen_runs

        cg = await sweep_stale_codegen_runs(ctx)
        results["codegen_stale_marked_failed"] = int(cg.get("marked_failed") or 0)
    except Exception as exc:
        logger.error("recovery_scan_codegen_stale_failed", error=str(exc), exc_info=True)

    try:
        results["setup_requeued"] = await _recover_failed_setups(db_factory)
    except Exception as exc:
        logger.error("recovery_scan_setup_pass_failed", error=str(exc), exc_info=True)

    try:
        results["boilerplate_requeued"] = await _recover_stalled_boilerplate_phase2(db_factory)
    except Exception as exc:
        logger.error("recovery_scan_boilerplate_pass_failed", error=str(exc), exc_info=True)

    try:
        results["deploy_created"] = await _recover_missing_live_hostnames(db_factory)
    except Exception as exc:
        logger.error("recovery_scan_deploy_pass_failed", error=str(exc), exc_info=True)

    try:
        results["preview_hostname_fixed"] = await _fix_missing_preview_hostnames(db_factory)
    except Exception as exc:
        logger.error("recovery_scan_preview_hostname_pass_failed", error=str(exc), exc_info=True)

    try:
        results["cf_synced"] = await _sync_from_cf_pages(db_factory)
    except Exception as exc:
        logger.error("recovery_scan_cf_sync_pass_failed", error=str(exc), exc_info=True)

    try:
        results["storage_ingest_requeued"] = await _recover_stuck_storage_ingest_jobs(db_factory)
    except Exception as exc:
        logger.error("recovery_scan_storage_ingest_pass_failed", error=str(exc), exc_info=True)

    logger.info("recovery_scan_complete", **results)
    return results


# ---------------------------------------------------------------------------
# Pass 0: expire stale in-flight deploy and preview records
# ---------------------------------------------------------------------------

async def _expire_stale_in_flight_records(db_factory) -> dict[str, int]:
    """Mark old stuck records as failed/stopped so later passes aren't blocked by them.

    Records that have been in an in-flight state longer than the expected maximum
    task duration are permanently stuck (worker crashed, task dropped, retries exhausted
    without updating status). Expire them so the rest of the recovery scan can act.
    """
    from datetime import datetime, timezone, timedelta
    from sqlalchemy import select, and_, update

    from src.domain.deployments.models import ProjectDeploy
    from src.domain.previews.models import PreviewEnvironment

    now = datetime.now(timezone.utc)
    deploy_cutoff = now - timedelta(seconds=_STALE_DEPLOY_AGE_S)
    preview_prov_cutoff = now - timedelta(seconds=_STALE_PREVIEW_PROVISIONING_AGE_S)
    preview_run_cutoff = now - timedelta(seconds=_STALE_PREVIEW_RUNNING_AGE_S)

    expired_deploys = 0
    expired_previews = 0

    # ── Stale project deploys ────────────────────────────────────────────────
    async with db_factory() as db:
        result = await db.execute(
            select(ProjectDeploy)
            .where(
                and_(
                    ProjectDeploy.status.in_(["pending", "deploying"]),
                    ProjectDeploy.created_at < deploy_cutoff,
                )
            )
            .limit(50)
        )
        stale_deploys = list(result.scalars().all())

        for deploy in stale_deploys:
            deploy.status = "failed"
            deploy.deployment_metadata = {
                **(deploy.deployment_metadata or {}),
                "expired_by_recovery_scan": True,
                "expired_at": now.isoformat(),
            }
            expired_deploys += 1
            logger.info(
                "recovery_expire_stale_deploy",
                deploy_id=str(deploy.id),
                project_id=str(deploy.project_id),
                environment=deploy.environment,
                age_minutes=int((now - deploy.created_at.replace(tzinfo=timezone.utc)).total_seconds() / 60),
            )

        if stale_deploys:
            await db.commit()

    # ── Stale preview environments ────────────────────────────────────────────
    async with db_factory() as db:
        result = await db.execute(
            select(PreviewEnvironment)
            .where(
                and_(
                    PreviewEnvironment.status == "provisioning",
                    PreviewEnvironment.created_at < preview_prov_cutoff,
                )
            )
            .limit(50)
        )
        stale_provisioning = list(result.scalars().all())

        result2 = await db.execute(
            select(PreviewEnvironment)
            .where(
                and_(
                    PreviewEnvironment.status == "running",
                    PreviewEnvironment.created_at < preview_run_cutoff,
                )
            )
            .limit(50)
        )
        stale_running = list(result2.scalars().all())

        for pe in stale_provisioning:
            pe.status = "failed"
            expired_previews += 1
            logger.info(
                "recovery_expire_stale_preview",
                preview_id=str(pe.id),
                project_id=str(pe.project_id),
                old_status="provisioning",
                age_minutes=int((now - pe.created_at.replace(tzinfo=timezone.utc)).total_seconds() / 60),
            )

        for pe in stale_running:
            pe.status = "stopped"
            pe.stopped_at = now
            expired_previews += 1
            logger.info(
                "recovery_expire_stale_preview",
                preview_id=str(pe.id),
                project_id=str(pe.project_id),
                old_status="running",
                age_minutes=int((now - pe.created_at.replace(tzinfo=timezone.utc)).total_seconds() / 60),
            )

        if stale_provisioning or stale_running:
            await db.commit()

    return {"deploys": expired_deploys, "previews": expired_previews}


# ---------------------------------------------------------------------------
# Pass 1: re-queue failed project setups
# ---------------------------------------------------------------------------

async def _recover_failed_setups(db_factory) -> int:
    """Re-enqueue setup for projects stuck in 'error' or stale 'creating' status.

    Two cases handled:
    - status=error: worker ran and called _mark_error(); recover after 5-min cooldown.
    - status=creating (stale): worker never ran, or _mark_error() itself threw and
      never updated the status; recover after 15-min cooldown to avoid racing in-flight tasks.
    """
    from datetime import datetime, timezone, timedelta
    from sqlalchemy import select, and_, or_

    from src.core.database import async_session_factory
    from src.domain.projects.models import Project, ProjectRepo
    from src.domain.projects.repository import ProjectRepository, ProjectRepoRepository

    now = datetime.now(timezone.utc)
    error_min_age = now - timedelta(seconds=_SETUP_MIN_AGE_S)
    stale_creating_cutoff = now - timedelta(seconds=_STALE_CREATING_AGE_S)
    max_age = now - timedelta(seconds=_SETUP_MAX_AGE_S)

    async with db_factory() as db:
        stmt = (
            select(Project)
            .outerjoin(ProjectRepo, Project.id == ProjectRepo.project_id)
            .where(
                and_(
                    Project.status.in_(["error", "creating"]),
                    Project.deleted_at.is_(None),
                    Project.created_at > max_age,
                    or_(
                        ProjectRepo.id.is_(None),
                        ProjectRepo.setup_stage != "completed",
                    ),
                    or_(
                        # error projects: short cooldown is enough
                        and_(Project.status == "error", Project.updated_at < error_min_age),
                        # creating projects: longer gap to avoid racing an in-flight worker
                        and_(Project.status == "creating", Project.created_at < stale_creating_cutoff),
                    ),
                )
            )
            .limit(_MAX_PER_SCAN)
        )
        rows = list((await db.execute(stmt)).scalars().all())

    if not rows:
        return 0

    from src.workers.enqueue import enqueue_apply_boilerplate, enqueue_project_setup

    recovered = 0
    for project in rows:
        try:
            # Legacy: ``skeleton_pushed`` — re-enqueue boilerplate repair task only.
            async with db_factory() as db:
                pr = await ProjectRepoRepository(db).get_by_project_id(project.id)
                phase1_done = pr and pr.setup_stage == "skeleton_pushed"

            if phase1_done:
                ok = await enqueue_apply_boilerplate(str(project.id))
                if ok:
                    # Reset status so project shows as recovering
                    async with db_factory() as db:
                        await ProjectRepository(db).update(project.id, status="creating")
                        await db.commit()
            else:
                async with db_factory() as db:
                    await ProjectRepository(db).update(project.id, status="creating")
                    await db.commit()
                ok = await enqueue_project_setup(str(project.id))
            if ok:
                recovered += 1
                logger.info(
                    "recovery_setup_requeued",
                    project_id=str(project.id),
                    project_type=project.type,
                    age_minutes=int((now - project.created_at.replace(tzinfo=timezone.utc)).total_seconds() / 60),
                )
            else:
                logger.warning("recovery_setup_enqueue_skipped", project_id=str(project.id))
        except Exception as exc:
            logger.error("recovery_setup_project_failed", project_id=str(project.id), error=str(exc))

    return recovered


async def _recover_stalled_boilerplate_phase2(db_factory) -> int:
    """Re-enqueue apply_boilerplate for legacy ``ready`` + ``skeleton_pushed`` rows."""
    from datetime import datetime, timezone, timedelta
    from sqlalchemy import select, and_

    from src.domain.projects.models import Project, ProjectRepo

    now = datetime.now(timezone.utc)
    min_age = now - timedelta(seconds=_STALLED_BOILERPLATE_AGE_S)
    max_age = now - timedelta(seconds=_SETUP_MAX_AGE_S)

    async with db_factory() as db:
        stmt = (
            select(Project.id)
            .join(ProjectRepo, Project.id == ProjectRepo.project_id)
            .where(
                and_(
                    Project.status == "ready",
                    Project.deleted_at.is_(None),
                    ProjectRepo.setup_stage == "skeleton_pushed",
                    Project.updated_at < min_age,
                    Project.created_at > max_age,
                )
            )
            .limit(_MAX_PER_SCAN)
        )
        project_ids = list((await db.execute(stmt)).scalars().all())

    if not project_ids:
        return 0

    from src.workers.enqueue import enqueue_apply_boilerplate

    requeued = 0
    for pid in project_ids:
        ok = await enqueue_apply_boilerplate(str(pid))
        if ok:
            requeued += 1
            logger.info("recovery_boilerplate_requeued", project_id=str(pid))
        else:
            logger.warning("recovery_boilerplate_enqueue_skipped", project_id=str(pid))
    return requeued


# ---------------------------------------------------------------------------
# Pass 2: create recovery deploy for projects missing live_hostname
# ---------------------------------------------------------------------------

async def _recover_missing_live_hostnames(db_factory) -> int:
    """Create and enqueue a production deploy for projects that never got one."""
    from datetime import datetime, timezone, timedelta
    from sqlalchemy import select, and_

    from src.domain.deployments.models import ProjectDeploy
    from src.domain.deployments.repository import DeployRepository
    from src.domain.projects.models import Project

    now = datetime.now(timezone.utc)
    min_age = now - timedelta(seconds=_DEPLOY_MIN_AGE_S)
    max_age = now - timedelta(seconds=_DEPLOY_MAX_AGE_S)

    async with db_factory() as db:
        stmt = (
            select(Project)
            .where(
                and_(
                    Project.status == "ready",
                    Project.deleted_at.is_(None),
                    Project.live_hostname.is_(None),
                    Project.active_revision_id.isnot(None),
                    Project.updated_at < min_age,
                    Project.created_at > max_age,
                )
            )
            .limit(_MAX_PER_SCAN)
        )
        projects = list((await db.execute(stmt)).scalars().all())

    if not projects:
        return 0

    from src.workers.enqueue import enqueue_run_project_deploy

    created = 0
    for project in projects:
        try:
            new_deploy_id = await _create_recovery_deploy_if_needed(
                db_factory, project, now
            )
            if new_deploy_id is None:
                continue
            ok = await enqueue_run_project_deploy(str(new_deploy_id))
            if ok:
                created += 1
                logger.info(
                    "recovery_deploy_enqueued",
                    project_id=str(project.id),
                    deploy_id=str(new_deploy_id),
                    project_type=project.type,
                )
            else:
                logger.warning("recovery_deploy_enqueue_skipped", project_id=str(project.id))
        except Exception as exc:
            logger.error("recovery_deploy_project_failed", project_id=str(project.id), error=str(exc))

    return created


async def _create_recovery_deploy_if_needed(db_factory, project, now):
    """Return a new deploy UUID if safe to create, or None to skip."""
    from datetime import timedelta, timezone
    from sqlalchemy import select, and_

    from src.domain.deployments.models import ProjectDeploy
    from src.domain.deployments.repository import DeployRepository

    async with db_factory() as db:
        deploys = await DeployRepository(db).list_for_project(project.id)

    # Already an in-flight deploy? Skip — let it complete.
    in_flight_statuses = {"pending", "deploying"}
    for d in deploys:
        if d.environment == "production" and d.status in in_flight_statuses:
            logger.debug(
                "recovery_deploy_skipped_in_flight",
                project_id=str(project.id),
                deploy_id=str(d.id),
                status=d.status,
            )
            return None

    # Count prior recovery attempts — don't loop forever.
    recovery_deploys = [
        d for d in deploys
        if (d.deployment_metadata or {}).get("recovery_attempt")
    ]
    if len(recovery_deploys) >= _MAX_DEPLOY_RECOVERY_ATTEMPTS:
        recovery_deploys.sort(key=lambda d: d.created_at or now)
        logger.warning(
            "recovery_deploy_max_attempts_reached",
            project_id=str(project.id),
            attempts=len(recovery_deploys),
            active_revision_id=str(project.active_revision_id) if project.active_revision_id else None,
            recovery_deployments=[
                {
                    "id": str(d.id),
                    "status": d.status,
                    "environment": d.environment,
                    "created_at": d.created_at.isoformat() if d.created_at else None,
                    "metadata_keys": list((d.deployment_metadata or {}).keys()),
                }
                for d in recovery_deploys
            ],
            operator_hint=(
                "Not a codegen/AI issue: production URL (live_hostname) never appeared after "
                f"{_MAX_DEPLOY_RECOVERY_ATTEMPTS} recovery deploys. Check GitHub Actions, "
                "Cloudflare Pages project, and `project_deploys` error metadata for this project."
            ),
        )
        return None

    async with db_factory() as db:
        deploy_repo = DeployRepository(db)
        d = await deploy_repo.create(
            project.id,
            project.active_revision_id,
            "production",
            created_by=None,
            deployment_metadata={
                "recovery_attempt": True,
                "recovery_attempt_number": len(recovery_deploys) + 1,
                "bootstrap_initial_hosting": True,
            },
        )
        new_id = d.id
        await db.commit()

    return new_id


# ---------------------------------------------------------------------------
# Pass 3: backfill missing preview_hostname (direct DB fix, no re-deploy)
# ---------------------------------------------------------------------------

async def _fix_missing_preview_hostnames(db_factory) -> int:
    """Directly write preview_hostname for projects that have a preview but no hostname.

    This is a data-only fix — no tasks enqueued. The hostname is deterministic
    from project_ref + the public apex domain.
    """
    from urllib.parse import urlparse
    from sqlalchemy import select, and_

    from src.config import get_settings
    from src.domain.hosting.hostnames import canonical_preview_url
    from src.domain.projects.models import Project
    from src.domain.projects.repository import ProjectRepository

    settings = get_settings()
    apex = settings.hosting_public_apex or "sites.p2hs.com"

    async with db_factory() as db:
        stmt = (
            select(Project)
            .where(
                and_(
                    Project.deleted_at.is_(None),
                    Project.active_preview_id.isnot(None),
                    Project.preview_hostname.is_(None),
                    Project.project_ref.isnot(None),
                )
            )
            .limit(_MAX_PER_SCAN)
        )
        projects = list((await db.execute(stmt)).scalars().all())

    if not projects:
        return 0

    fixed = 0
    for project in projects:
        try:
            project_ref = (project.project_ref or "").strip()
            if not project_ref:
                continue
            hostname = urlparse(canonical_preview_url(project_ref, apex)).netloc or ""
            if not hostname:
                continue
            async with db_factory() as db:
                await ProjectRepository(db).update_hostnames(project.id, preview_hostname=hostname)
                await db.commit()
            fixed += 1
            logger.info(
                "recovery_preview_hostname_fixed",
                project_id=str(project.id),
                preview_hostname=hostname,
            )
        except Exception as exc:
            logger.error("recovery_preview_hostname_failed", project_id=str(project.id), error=str(exc))

    return fixed


# ---------------------------------------------------------------------------
# Pass 4: sync from CF Pages — heal projects where GA deployed but backend missed it
# ---------------------------------------------------------------------------

_CF_SYNC_MIN_AGE_S = 10 * 60   # 10 min — give the normal webhook path time to fire first
_CF_SYNC_MAX_AGE_S = 24 * 60 * 60  # 24 hrs


async def _sync_from_cf_pages(db_factory) -> int:
    """Query CF Pages directly for projects missing live_hostname or preview_hostname.

    Handles cases where:
    - GitHub Actions ran (manually or automatically) but the webhook was missed
    - The deploy task polled the wrong commit SHA and exhausted retries
    - The project was manually re-run from the GitHub Actions dashboard

    For each eligible project, fetches the latest live CF Pages deployment for the
    relevant branch, finds or creates a revision matching its commit SHA, then
    enqueues run_project_deploy so the routing plane (W4P) is set up and the
    hostname is written to the DB.
    """
    from datetime import datetime, timezone, timedelta
    from sqlalchemy import select, and_, or_

    from src.config import get_settings
    from src.domain.deployments.repository import DeployRepository
    from src.domain.hosting.adapters.cloudflare_pages import CloudflarePagesProvider
    from src.domain.projects.models import Project, ProjectRepo
    from src.domain.revisions.repository import RevisionRepository
    from src.workers.enqueue import enqueue_run_project_deploy

    settings = get_settings()
    if settings.hosting_provider != "cloudflare_pages":
        return 0
    if not settings.cloudflare_api_token or not settings.cloudflare_account_id:
        return 0

    now = datetime.now(timezone.utc)
    min_age = now - timedelta(seconds=_CF_SYNC_MIN_AGE_S)
    max_age = now - timedelta(seconds=_CF_SYNC_MAX_AGE_S)

    async with db_factory() as db:
        stmt = (
            select(Project, ProjectRepo)
            .join(ProjectRepo, Project.id == ProjectRepo.project_id)
            .where(
                and_(
                    Project.status == "ready",
                    Project.deleted_at.is_(None),
                    Project.created_at < min_age,
                    Project.created_at > max_age,
                    Project.active_revision_id.isnot(None),
                    or_(
                        Project.live_hostname.is_(None),
                        Project.preview_hostname.is_(None),
                    ),
                    ProjectRepo.repo_full_name.isnot(None),
                )
            )
            .limit(_MAX_PER_SCAN)
        )
        rows = list((await db.execute(stmt)).all())

    if not rows:
        return 0

    cf = CloudflarePagesProvider(settings)
    preview_branch = settings.hosting_preview_branch or "preview"
    synced = 0

    for project, repo in rows:
        try:
            project_ref = (project.project_ref or "").strip()
            if not project_ref:
                continue

            # Determine which environments need healing
            envs_to_check = []
            if project.live_hostname is None:
                envs_to_check.append(("production", repo.default_branch or "main"))
            if project.preview_hostname is None:
                envs_to_check.append(("preview", preview_branch))

            for environment, branch in envs_to_check:
                try:
                    cf_deploy = await cf.find_latest_live_deployment_for_branch(project_ref, branch)
                    if not cf_deploy:
                        logger.debug(
                            "recovery_cf_no_live_deploy",
                            project_id=str(project.id),
                            environment=environment,
                            branch=branch,
                        )
                        continue

                    # Extract commit SHA from CF Pages deployment trigger metadata
                    trigger_meta = (cf_deploy.get("deployment_trigger") or {}).get("metadata") or {}
                    commit_sha = trigger_meta.get("commit_hash", "")
                    if not commit_sha:
                        continue

                    # Guard: skip if there's already an in-flight deploy for this environment,
                    # or if we've already exhausted recovery attempts via this pass.
                    async with db_factory() as db:
                        deploys = await DeployRepository(db).list_for_project(project.id)
                    in_flight = any(
                        d.environment == environment and d.status in {"pending", "deploying"}
                        for d in deploys
                    )
                    if in_flight:
                        continue
                    cf_sync_attempts = sum(
                        1 for d in deploys
                        if (d.deployment_metadata or {}).get("cf_sync")
                        and d.environment == environment
                    )
                    if cf_sync_attempts >= _MAX_DEPLOY_RECOVERY_ATTEMPTS:
                        logger.warning(
                            "recovery_cf_sync_max_attempts",
                            project_id=str(project.id),
                            environment=environment,
                            attempts=cf_sync_attempts,
                        )
                        continue

                    # Find or create a revision for the commit SHA that CF Pages built
                    async with db_factory() as db:
                        rev_repo = RevisionRepository(db)
                        rev = await rev_repo.get_by_git_sha(project.id, commit_sha)
                        if not rev:
                            # Inherit source from the latest codegen revision so the
                            # deploy gate (source == "codegen") passes for preview
                            fallback = await rev_repo.get_by_source(project.id, "codegen")
                            rev = await rev_repo.create(
                                project_id=project.id,
                                git_sha=commit_sha,
                                source="codegen",
                                source_id=fallback.source_id if fallback else None,
                                commit_message=f"[p2hs] CF Pages sync at {commit_sha[:12]}",
                            )
                            await db.commit()

                        deploy_repo = DeployRepository(db)
                        d = await deploy_repo.create(
                            project.id,
                            rev.id,
                            environment,
                            created_by=None,
                            deployment_metadata={
                                "recovery_attempt": True,
                                "cf_sync": True,
                                "cf_deployment_id": cf_deploy.get("id"),
                                "commit_sha": commit_sha,
                            },
                        )
                        new_deploy_id = d.id
                        await db.commit()

                    ok = await enqueue_run_project_deploy(str(new_deploy_id))
                    if ok:
                        synced += 1
                        logger.info(
                            "recovery_cf_sync_enqueued",
                            project_id=str(project.id),
                            environment=environment,
                            deploy_id=str(new_deploy_id),
                            commit_sha=commit_sha[:12],
                        )
                except Exception as env_exc:
                    logger.error(
                        "recovery_cf_sync_env_failed",
                        project_id=str(project.id),
                        environment=environment,
                        error=str(env_exc),
                    )

        except Exception as exc:
            logger.error("recovery_cf_sync_project_failed", project_id=str(project.id), error=str(exc))

    return synced


async def _expire_stale_manifest_runs(db_factory) -> int:
    """Mark active manifest runs that are past their expires_at as expired."""
    async with db_factory() as db:
        from src.domain.manifest.repository import ManifestRunRepository
        repo = ManifestRunRepository(db)
        count = await repo.expire_stale()
        await db.commit()
        return count


_STUCK_STORAGE_INGEST_PENDING_AGE_S = 2 * 60


async def _recover_stuck_storage_ingest_jobs(db_factory) -> int:
    """Re-enqueue ingest jobs stuck in pending after upload completed."""
    from datetime import datetime, timedelta, timezone

    from sqlalchemy import select

    from src.domain.storage_ingest.models import StorageIngestJob, StorageIngestJobStatus
    from src.workers.enqueue import enqueue_storage_ingest_job

    cutoff = datetime.now(timezone.utc) - timedelta(seconds=_STUCK_STORAGE_INGEST_PENDING_AGE_S)
    requeued = 0

    async with db_factory() as db:
        result = await db.execute(
            select(StorageIngestJob)
            .where(
                StorageIngestJob.status == StorageIngestJobStatus.PENDING.value,
                StorageIngestJob.updated_at < cutoff,
            )
            .order_by(StorageIngestJob.updated_at.asc())
            .limit(_MAX_PER_SCAN)
        )
        rows = list(result.scalars().all())

    for row in rows:
        try:
            if row.expected_size_bytes and row.received_bytes != row.expected_size_bytes:
                continue
            ok = await enqueue_storage_ingest_job(str(row.id), retry=True)
            if ok:
                requeued += 1
                logger.info("recovery_storage_ingest_requeued", job_id=str(row.id))
        except Exception as exc:
            logger.error(
                "recovery_storage_ingest_failed",
                job_id=str(row.id),
                error=str(exc),
                exc_info=True,
            )

    return requeued