"""Enqueue background jobs for P2HS workers."""
from __future__ import annotations

from datetime import datetime, timedelta
from typing import Any, Optional

import structlog

from src.config import get_settings

logger = structlog.get_logger()


def _redis_target_for_logs() -> dict:
    """Safe connection target for logs (no password)."""
    from arq.connections import RedisSettings

    rs = RedisSettings.from_dsn(get_settings().redis_url)
    return {
        "redis_host": rs.host,
        "redis_port": rs.port,
        "redis_database": rs.database,
        "redis_ssl": rs.ssl,
        "arq_queue": "arq:queue",
    }


async def _get_redis_pool():
    from arq import create_pool
    from arq.connections import RedisSettings
    settings = get_settings()
    return await create_pool(RedisSettings.from_dsn(settings.redis_url))


async def enqueue_codegen_run(run_id: str, project_id: Optional[str] = None) -> bool:
    """Enqueue a codegen run job. Returns True if enqueued.

    Publishes a ``codegen.queued`` WebSocket event when project_id is provided.
    """
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("run_codegen", run_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="run_codegen",
                run_id=run_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="run_codegen",
            job_id=job.job_id,
            run_id=run_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_external_admin_chat_job(job_id: str, *, defer_seconds: float = 2.0) -> bool:
    """Enqueue Core-owned external-admin assistant chat job."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "process_external_admin_chat_job",
            job_id,
            _defer_by=timedelta(seconds=defer_seconds),
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="process_external_admin_chat_job",
                job_id=job_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="process_external_admin_chat_job",
            arq_job_id=job.job_id,
            chat_job_id=job_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_wordpress_site_understanding_run(run_id: str, *, defer_seconds: float = 2.0) -> bool:
    """Enqueue processing for a WordPress Site Understanding export."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "process_wordpress_site_understanding_run",
            run_id,
            _defer_by=timedelta(seconds=defer_seconds),
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="process_wordpress_site_understanding_run",
                run_id=run_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="process_wordpress_site_understanding_run",
            arq_job_id=job.job_id,
            run_id=run_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_shopify_site_understanding_run(run_id: str, *, defer_seconds: float = 2.0) -> bool:
    """Enqueue processing for a Shopify Store Understanding export."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "process_shopify_site_understanding_run",
            run_id,
            _defer_by=timedelta(seconds=defer_seconds),
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="process_shopify_site_understanding_run",
                run_id=run_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="process_shopify_site_understanding_run",
            arq_job_id=job.job_id,
            run_id=run_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_execute_wp_mutation(session_id: str, *, defer_seconds: float = 1.0) -> bool:
    """Enqueue execution of an approved WP mutation plan."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "execute_wp_mutation_plan",
            session_id,
            _defer_by=timedelta(seconds=defer_seconds),
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="execute_wp_mutation_plan",
                session_id=session_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="execute_wp_mutation_plan",
            arq_job_id=job.job_id,
            session_id=session_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_execute_shopify_mutation(session_id: str, *, defer_seconds: float = 1.0) -> bool:
    """Enqueue direct execution of an approved Shopify mutation plan via Admin API."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "execute_shopify_mutation_plan",
            session_id,
            _defer_by=timedelta(seconds=defer_seconds),
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="execute_shopify_mutation_plan",
                session_id=session_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="execute_shopify_mutation_plan",
            arq_job_id=job.job_id,
            session_id=session_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_project_setup(project_id: str) -> bool:
    """Enqueue project setup worker (GitHub repo, boilerplate, revision #1). Returns True if enqueued."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "setup_project_repo",
            project_id,
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="setup_project_repo",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="setup_project_repo",
            job_id=job.job_id,
            project_id=project_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_provision_preview_for_revision(
    preview_id: str, project_id: str, revision_id: str, webhook_triggered: bool = False
) -> bool:
    """Enqueue preview provisioning for a revision.

    Pass webhook_triggered=True when called from a workflow_run webhook — CF Pages
    already has the deployment at that point so the task skips its polling retry loop.
    """
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "provision_preview_for_revision",
            preview_id,
            project_id,
            revision_id,
            webhook_triggered=webhook_triggered,
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="provision_preview_for_revision",
                preview_id=preview_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="provision_preview_for_revision",
            job_id=job.job_id,
            preview_id=preview_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_run_project_deploy(deploy_id: str) -> bool:
    """Enqueue hosting pipeline for a project deploy row. Returns True if enqueued."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("run_project_deploy", deploy_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="run_project_deploy",
                deploy_id=deploy_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="run_project_deploy",
            job_id=job.job_id,
            deploy_id=deploy_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_merge_and_deploy(project_id: str, triggered_by: str | None = None) -> bool:
    """Enqueue the merge-preview-into-main + production deploy job. Returns True if enqueued."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("merge_and_deploy_to_production", project_id, triggered_by)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="merge_and_deploy_to_production",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info("arq_enqueue_ok", function="merge_and_deploy_to_production", job_id=job.job_id, project_id=project_id, **_redis_target_for_logs())
        return True
    finally:
        await redis.close()


async def enqueue_discovery_job(kind: str, project_id: str, version: int) -> bool:
    redis = await _get_redis_pool()
    try:
        name = "run_project_discovery" if kind == "discovery_turn" else "apply_project_discovery"
        await redis.enqueue_job(name, project_id, version,
                                _job_id=f"{name}:{project_id}:{version}")
        return True  # An already queued job is also a successful durable handoff.
    finally:
        await redis.close()


async def enqueue_apply_boilerplate(project_id: str) -> bool:
    """Enqueue legacy boilerplate task (stuck ``skeleton_pushed`` or repair). Returns True if enqueued."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("apply_boilerplate_snapshot", project_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="apply_boilerplate_snapshot",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="apply_boilerplate_snapshot",
            job_id=job.job_id,
            project_id=project_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_fix_deploy_workflows(
    project_id: str | None = None,
    *,
    sync_runtime_adapters: bool = False,
) -> bool:
    """Enqueue fix_deploy_workflows task.

    If project_id is given, fixes only that project. Otherwise scans all active projects.
    When enabled, also refreshes the generated app's managed runtime BFF adapters.
    """
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "fix_deploy_workflows",
            project_id=project_id,
            sync_runtime_adapters=sync_runtime_adapters,
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="fix_deploy_workflows",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="fix_deploy_workflows",
            job_id=job.job_id,
            project_id=project_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_migrate_runtime_primitives(project_id: str | None = None) -> bool:
    """Enqueue leftover-auth-path migration for one project or all active projects."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("migrate_runtime_primitives", project_id=project_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="migrate_runtime_primitives",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="migrate_runtime_primitives",
            job_id=job.job_id,
            project_id=project_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_migrate_workspace_to_w4p(
    workspace_id: str,
    project_ids: list[str] | None = None,
    dry_run: bool = False,
) -> bool:
    """Enqueue the idempotent workspace/project W4P migration orchestrator."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "migrate_workspace_to_w4p",
            workspace_id,
            project_ids=project_ids,
            dry_run=dry_run,
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="migrate_workspace_to_w4p",
                workspace_id=workspace_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="migrate_workspace_to_w4p",
            job_id=job.job_id,
            workspace_id=workspace_id,
            project_count=len(project_ids or []),
            dry_run=dry_run,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_project_outbox_dispatch_nudge(*, defer_seconds: float = 3.0) -> bool:
    """Schedule the ``dispatch_project_outbox`` worker after a short delay.

    Correctness does not depend on this (cron runs every minute); it reduces latency
    until pending outbox rows are published to Redis after DB commits.
    """
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "dispatch_project_outbox",
            _defer_by=timedelta(seconds=defer_seconds),
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="dispatch_project_outbox",
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="dispatch_project_outbox",
            job_id=job.job_id,
            defer_seconds=defer_seconds,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_reconcile_workspace_projection(workspace_id: str) -> bool:
    """Ask for a convergence pass over a workspace's projects. Reconciliation reads canonical
    state, so a lost or late enqueue is repaired by the hourly sweep."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("reconcile_workspace_projection", workspace_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="reconcile_workspace_projection",
                workspace_id=workspace_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        return True
    finally:
        await redis.close()


async def enqueue_sync_membership_projection(project_id: str) -> bool:
    """Push a project's canonical membership projections to Cloud. Repeat-safe."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("sync_membership_projection", project_id)
        return job is not None
    finally:
        await redis.close()


async def enqueue_reconcile_project_membership(project_id: str) -> bool:
    """Repair one project's membership projection from Workspace authority."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("reconcile_project_membership", project_id)
        return job is not None
    finally:
        await redis.close()


async def enqueue_cloud_sync_project(project_id: str) -> bool:
    """Enqueue cloud control-plane bootstrap for a project. Returns True if enqueued."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("sync_project_to_cloud", project_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="sync_project_to_cloud",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="sync_project_to_cloud",
            job_id=job.job_id,
            project_id=project_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_template_deploy(template_id: str, environment: str = "production") -> bool:
    """Enqueue a template demo deployment. Returns True if enqueued."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("run_template_deploy", template_id, environment)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="run_template_deploy",
                template_id=template_id,
                environment=environment,
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="run_template_deploy",
            job_id=job.job_id,
            template_id=template_id,
            environment=environment,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_env_sync(project_id: str) -> bool:
    """Enqueue CF Pages env var sync for a project. Returns True if enqueued."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("sync_project_env_vars", project_id=project_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="sync_project_env_vars",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="sync_project_env_vars",
            job_id=job.job_id,
            project_id=project_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_sync_platform_files(
    *,
    project_id: Optional[str] = None,
    template_id: Optional[str] = None,
) -> bool:
    """Enqueue platform-managed file sync.

    project_id or template_id → sync that single entity.
    Neither → bulk: syncs all stale projects and templates (up to 50 each).
    """
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "sync_platform_files",
            project_id=project_id,
            template_id=template_id,
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="sync_platform_files",
                project_id=project_id,
                template_id=template_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="sync_platform_files",
            job_id=job.job_id,
            project_id=project_id,
            template_id=template_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_shopify_app_rebuild(project_id: str) -> bool:
    """Enqueue Shopify App Worker rebuild (regenerates registry.ts + wrangler.toml + function files)."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("rebuild_shopify_app_worker", project_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="rebuild_shopify_app_worker",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="rebuild_shopify_app_worker",
            job_id=job.job_id,
            project_id=project_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_shopify_app_provision(project_id: str) -> bool:
    """Enqueue CF resource provisioning for a shopify_app project (D1, Queue, R2, KV, Cloud token)."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job("provision_shopify_app_resources", project_id)
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="provision_shopify_app_resources",
                project_id=project_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="provision_shopify_app_resources",
            job_id=job.job_id,
            project_id=project_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_pre_registration_complete(
    *,
    pending_registration_id: str,
    stripe_customer_id: str,
    stripe_subscription_id: Optional[str] = None,
) -> bool:
    """Enqueue pre-registration completion after Stripe checkout (mark paid + send magic link)."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(
            "complete_pre_registration",
            pending_registration_id=pending_registration_id,
            stripe_customer_id=stripe_customer_id,
            stripe_subscription_id=stripe_subscription_id,
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="complete_pre_registration",
                pending_registration_id=pending_registration_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="complete_pre_registration",
            job_id=job.job_id,
            pending_registration_id=pending_registration_id,
            **_redis_target_for_logs(),
        )
        return True
    except Exception as exc:
        logger.error(
            "pre_registration_enqueue_error",
            pending_registration_id=pending_registration_id,
            error=str(exc),
            exc_info=True,
        )
        return False
    finally:
        await redis.close()


async def enqueue_storage_ingest_job(job_id: str, *, retry: bool = False) -> bool:
    """Enqueue server-side storage ingest (staging → extract → R2)."""
    import time

    redis = await _get_redis_pool()
    try:
        arq_job_id = f"storage-ingest:{job_id}"
        if retry:
            arq_job_id = f"storage-ingest:{job_id}:retry:{int(time.time())}"
        job = await redis.enqueue_job(
            "process_storage_ingest_job",
            job_id,
            _job_id=arq_job_id,
        )
        if job is None:
            logger.warning(
                "arq_enqueue_skipped",
                function="process_storage_ingest_job",
                job_id=job_id,
                reason="job_id_collision_or_watch_conflict",
                **_redis_target_for_logs(),
            )
            return False
        logger.info(
            "arq_enqueue_ok",
            function="process_storage_ingest_job",
            arq_job_id=job.job_id,
            job_id=job_id,
            **_redis_target_for_logs(),
        )
        return True
    finally:
        await redis.close()


async def enqueue_billing_email(job_name: str, **kwargs: Any) -> bool:
    """Enqueue a billing email job. Returns True if enqueued. Logs on failure."""
    redis = await _get_redis_pool()
    try:
        job = await redis.enqueue_job(job_name, **kwargs)
        if job is None:
            logger.warning(
                "billing_email_enqueue_failed",
                task_name=job_name,
                workspace_id=kwargs.get("workspace_id"),
                source_object_id=kwargs.get("stripe_invoice_id") or kwargs.get("stripe_checkout_session_id") or kwargs.get("stripe_subscription_id"),
            )
            return False
        return True
    except Exception as exc:
        logger.error(
            "billing_email_enqueue_error",
            task_name=job_name,
            workspace_id=kwargs.get("workspace_id"),
            source_object_id=kwargs.get("stripe_invoice_id") or kwargs.get("stripe_checkout_session_id") or kwargs.get("stripe_subscription_id"),
            error=str(exc),
            exc_info=True,
        )
        return False
    finally:
        await redis.close()


async def enqueue_admin_email(
    *,
    recipient_emails: list[str],
    from_email: str,
    subject: str,
    html_body: str,
    requested_by: str,
    scheduled_at: datetime | None = None,
) -> bool:
    """Queue an operator-authored email, optionally for a future UTC time."""
    from uuid import uuid4

    redis = await _get_redis_pool()
    try:
        options: dict[str, Any] = {"_job_id": f"admin-email:{uuid4()}"}
        if scheduled_at is not None:
            options["_defer_until"] = scheduled_at
        job = await redis.enqueue_job(
            "send_admin_email",
            recipient_emails=recipient_emails,
            from_email=from_email,
            subject=subject,
            html_body=html_body,
            requested_by=requested_by,
            **options,
        )
        if job is None:
            logger.warning(
                "admin_email_enqueue_failed",
                requested_by=requested_by,
                recipient_count=len(recipient_emails),
            )
            return False
        return True
    except Exception as exc:
        logger.error(
            "admin_email_enqueue_error",
            requested_by=requested_by,
            error=str(exc),
            exc_info=True,
        )
        return False
    finally:
        await redis.close()