"""Worker entry point - runs arq worker with all registered tasks.

Run (from repository root, same as the API):

  arq src.workers.main.WorkerSettings

Or:

  python -m arq src.workers.main.WorkerSettings

Or:

  python -m src.workers.main
"""
import logging
from src.workers.tasks.project_discovery import run_project_discovery, apply_project_discovery

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
log = logging.getLogger("src.workers.main")

log.info("importing worker entrypoint")

# Register all tables on Base.metadata before task modules pull in a subset of models (avoids
# NoReferencedTableError e.g. projects.created_by → users.id on worker/Linux + some import paths).
from src.core.orm_import import import_all_orm_models

import_all_orm_models()

from arq import run_worker
from arq.connections import RedisSettings
from arq.cron import cron
from arq.worker import func

# The arq CLI installs its own StreamHandler before importing WorkerSettings.
# Use the application's single root handler instead, otherwise every job and
# the startup summary are emitted once by arq and once by the application.
_arq_log = logging.getLogger("arq")
_arq_log.handlers.clear()
_arq_log.propagate = True
for _noisy_logger_name in ("httpx", "httpcore", "redis", "redis.asyncio"):
    # Successful transport/statistics lines are useful only while diagnosing;
    # warnings and errors still pass through at their normal levels.
    logging.getLogger(_noisy_logger_name).setLevel(logging.WARNING)

from src.config import get_settings
from src.workers.tasks.admin_emails import send_admin_email
from src.workers.tasks.ai_reembed import ai_reembed
from src.workers.tasks.apply_boilerplate import apply_boilerplate_snapshot
from src.workers.tasks.billing_emails import (
    send_invoice_paid_email,
    send_payment_failed_email,
    send_subscription_canceled_email,
    send_usage_topup_confirmation_email,
    send_wallet_threshold_email,
)
from src.workers.tasks.build import run_build
from src.workers.tasks.cleanup import cleanup_expired_guest_drafts, cleanup_expired_previews
from src.workers.tasks.cleanup_work_branches import cleanup_codegen_work_branches
from src.workers.tasks.cloud_sync import sync_project_to_cloud
from src.workers.tasks.codegen import resolve_codegen_capabilities, run_codegen
from src.workers.tasks.device_health_monitor import monitor_device_health
from src.workers.tasks.domain_renewals import reconcile_domain_renewals
from src.workers.tasks.env_sync import sync_project_env_vars
from src.workers.tasks.execute_shopify_mutation import execute_shopify_mutation_plan
from src.workers.tasks.execute_wp_mutation import execute_wp_mutation_plan
from src.workers.tasks.external_admin_chat import process_external_admin_chat_job
from src.workers.tasks.fix_deploy_workflows import fix_deploy_workflows
from src.workers.tasks.migrate_runtime_primitives import migrate_runtime_primitives
from src.workers.tasks.migrate_workspace_to_w4p import migrate_workspace_to_w4p
from src.workers.tasks.migrate_legacy_form_uploads import migrate_legacy_form_uploads
from src.workers.tasks.project_membership_projection import (
    reconcile_all_project_memberships,
    reconcile_project_membership,
    reconcile_workspace_projection,
    sync_all_membership_projections,
    sync_membership_projection,
)
from src.workers.tasks.reconcile_project_capabilities import reconcile_project_capabilities
from src.workers.tasks.fleet_monitor import monitor_worker_fleet
from src.workers.tasks.merge_and_deploy import merge_and_deploy_to_production
from src.workers.tasks.pre_registration import (
    complete_pre_registration,
    scan_abandoned_pre_registrations,
    send_pre_registration_recovery_email,
)
from src.workers.tasks.preview import provision_preview
from src.workers.tasks.project_deploy import run_project_deploy
from src.workers.tasks.project_outbox_dispatch import dispatch_project_outbox
from src.workers.tasks.project_recovery import run_recovery_scan
from src.workers.tasks.project_setup import setup_project_repo
from src.workers.tasks.provision_preview_for_revision import provision_preview_for_revision
from src.workers.tasks.shopify_app_provision import provision_shopify_app_resources
from src.workers.tasks.shopify_app_rebuild import rebuild_shopify_app_worker
from src.workers.tasks.shopify_site_understanding import process_shopify_site_understanding_run
from src.workers.tasks.storage_ingest import process_storage_ingest_job
from src.workers.tasks.sync_platform_files import sync_platform_files
from src.workers.tasks.bill_email_overage import bill_email_overage
from src.workers.tasks.template_deploy import run_template_deploy
from src.workers.tasks.usage_snapshot import daily_usage_snapshot
from src.workers.tasks.wordpress_site_understanding import process_wordpress_site_understanding_run

from src.workers.tasks.agent_operations import dispatch_agent_operations
from src.workers.tasks.agent_purchases import reconcile_agent_purchases

BILLING_EMAIL_MAX_TRIES = 5
# run_project_deploy polls CF Pages every 30s; allow up to 20 min for a slow GH Actions build
_DEPLOY_MAX_TRIES = 40
_WORKER_SETTINGS = get_settings()
_CODEGEN_JOB_TIMEOUT = _WORKER_SETTINGS.codegen_job_timeout_seconds

async def startup(ctx: dict) -> None:
    # arq's CLI installs its formatter after this module is imported. Repeat
    # the handler normalization at the lifecycle boundary so all subsequent
    # startup/task logs use one path.
    _arq_log.handlers.clear()
    _arq_log.propagate = True
    log.info("worker startup hook begin")
    from src.core.database import async_session_factory

    rs = RedisSettings.from_dsn(_WORKER_SETTINGS.redis_url)
    ctx["db_factory"] = async_session_factory

    from src.core.diagnostics import run_and_log_warnings
    try:
        await run_and_log_warnings()
    except Exception as exc:
        log.warning("diagnostics_startup_error: %s", exc)

    # Deploy / sync the dispatcher Worker on every startup (idempotent — one CF API call).
    # Skipped automatically when hosting_provider=stub or CF credentials are absent.
    if (
        _WORKER_SETTINGS.worker_dispatcher_deploy_on_startup
        and _WORKER_SETTINGS.hosting_provider == "cloudflare_pages"
        and _WORKER_SETTINGS.cloudflare_api_token
    ):
        try:
            from src.domain.hosting.routing_plane import CloudflareRoutingProvider
            result = await CloudflareRoutingProvider(_WORKER_SETTINGS).deploy_dispatcher()
            log.info("dispatcher_deploy_ok script=%s routes=%s", result.get("script_name"), result.get("routes", []))
        except Exception as exc:
            # Non-fatal: log and continue. Worker still processes jobs normally.
            log.warning("dispatcher_deploy_failed error=%s", exc)
    elif _WORKER_SETTINGS.hosting_provider == "cloudflare_pages":
        log.info("dispatcher_deploy_skipped reason=explicit_reconciliation_required")

    log.info(
        "worker startup hook complete — polling arq:queue on Redis host=%s port=%s db=%s ssl=%s max_jobs=%s "
        "(API must use the same REDIS_URL or jobs never arrive)",
        rs.host,
        rs.port,
        rs.database,
        rs.ssl,
        WorkerSettings.max_jobs,
    )

async def shutdown(ctx: dict) -> None:
    log.info("worker shutdown hook")

class WorkerSettings:
    functions = [
        func(run_project_discovery, max_tries=3),
        func(apply_project_discovery, max_tries=3),
        resolve_codegen_capabilities,
        setup_project_repo,
        func(run_codegen, timeout=_CODEGEN_JOB_TIMEOUT, max_tries=3),
        run_build,
        provision_preview,
        provision_preview_for_revision,
        func(run_project_deploy, max_tries=_DEPLOY_MAX_TRIES),
        func(merge_and_deploy_to_production, max_tries=_DEPLOY_MAX_TRIES),
        func(apply_boilerplate_snapshot, max_tries=3),
        fix_deploy_workflows,
        migrate_runtime_primitives,
        func(migrate_workspace_to_w4p, max_tries=1),
        func(migrate_legacy_form_uploads, max_tries=3),
        func(reconcile_project_capabilities, max_tries=3),
        func(reconcile_workspace_projection, max_tries=5),
        func(reconcile_project_membership, max_tries=5),
        func(reconcile_all_project_memberships, max_tries=3),
        func(sync_membership_projection, max_tries=5),
        func(sync_all_membership_projections, max_tries=3),
        func(sync_project_to_cloud, max_tries=5),
        func(run_template_deploy, max_tries=_DEPLOY_MAX_TRIES),
        cleanup_expired_previews,
        func(send_invoice_paid_email, max_tries=BILLING_EMAIL_MAX_TRIES),
        func(send_usage_topup_confirmation_email, max_tries=BILLING_EMAIL_MAX_TRIES),
        func(send_payment_failed_email, max_tries=BILLING_EMAIL_MAX_TRIES),
        func(send_subscription_canceled_email, max_tries=BILLING_EMAIL_MAX_TRIES),
        func(send_wallet_threshold_email, max_tries=BILLING_EMAIL_MAX_TRIES),
        # Admin-authored sends are not retried to avoid duplicate delivery after partial success.
        func(send_admin_email, max_tries=1),
        func(ai_reembed, max_tries=3),
        monitor_worker_fleet,
        monitor_device_health,
        func(reconcile_domain_renewals, max_tries=3),
        daily_usage_snapshot,
        func(sync_project_env_vars, max_tries=3),
        func(sync_platform_files, max_tries=3),
        func(rebuild_shopify_app_worker, max_tries=3),
        func(provision_shopify_app_resources, max_tries=3),
        dispatch_project_outbox,
        func(process_external_admin_chat_job, max_tries=3),
        func(process_wordpress_site_understanding_run, max_tries=3),
        func(process_shopify_site_understanding_run, max_tries=3),
        func(execute_wp_mutation_plan, max_tries=2),
        func(execute_shopify_mutation_plan, max_tries=2),
        func(process_storage_ingest_job, max_tries=3, timeout=3600),
        func(complete_pre_registration, max_tries=5),
        # Retain disabled recovery handlers to drain jobs queued by older workers.
        func(send_pre_registration_recovery_email, max_tries=3),
        scan_abandoned_pre_registrations,
    ]
    cron_jobs = [
        cron(dispatch_agent_operations, minute=set(range(60))),
        cron(reconcile_agent_purchases, minute=set(range(60))),
        # Recovery scanner (every minute): setups, deploys, previews, CF sync, and stale codegen runs
        # (see sweep_stale_codegen_runs invoked from run_recovery_scan).
        cron(run_recovery_scan, minute=set(range(60))),
        cron(dispatch_project_outbox, minute=set(range(60))),
        # Refresh Core-owned capability policy for existing projects after catalog changes.
        cron(reconcile_project_capabilities, minute=set(range(0, 60, 10))),
        # Membership drift repair (Workstream D): converges even if no Workspace event fired.
        cron(reconcile_all_project_memberships, minute={7}),
        # Cloud projection convergence (E4): repairs pending rows and out-of-date projections.
        cron(sync_all_membership_projections, minute=set(range(0, 60, 10))),
        # Repair the narrow presign-only upload shape in older generated Nuxt apps.
        cron(migrate_legacy_form_uploads, minute=set(range(5, 60, 10))),
        # Fleet monitor: health checks + auto-scaling every minute
        cron(monitor_worker_fleet, minute=set(range(60))),
        # Device heartbeat health monitor: flag stale-heartbeat devices offline every minute
        cron(monitor_device_health, minute=set(range(60))),
        # Domain registration completion + Cloudflare auto-renew reconciliation.
        cron(reconcile_domain_renewals, hour=set(range(0, 24, 6)), minute=40),
        # Daily usage snapshot + credit exhaustion alerts (midnight UTC)
        cron(daily_usage_snapshot, hour=0, minute=5),
        # Managed email domain overage -> Stripe invoice items (midnight UTC)
        cron(bill_email_overage, hour=0, minute=20),
        # Public drafts contain unauthenticated uploads and must quietly expire after 90 days.
        cron(cleanup_expired_guest_drafts, hour=1, minute=15),
        # Expired run work branches (p2hs/codegen/<run_id>); no-op unless the work branch is enabled.
        cron(cleanup_codegen_work_branches, hour=2, minute=30),
    ]
    on_startup = startup
    on_shutdown = shutdown
    max_jobs = _WORKER_SETTINGS.arq_max_jobs
    job_timeout = 600
    redis_settings = RedisSettings.from_dsn(_WORKER_SETTINGS.redis_url)

if __name__ == "__main__":
    log.info("starting arq worker")
    run_worker(WorkerSettings)