"""Celery application configuration for async document processing."""

import asyncio
import os

# Force httpx to not use HTTP/2 to avoid segfaults in forked processes
# This is a known issue with httpx and OpenAI SDK in Celery prefork workers
os.environ.setdefault("HTTPX_HTTP2", "0")

from celery import Celery
from celery.signals import worker_process_init

from src.core.config import settings


class ThreadSafeEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    """
    Custom event loop policy that automatically creates event loops for threads.

    The default asyncio policy raises RuntimeError when get_event_loop() is called
    from a thread without an event loop. This policy instead creates a new loop,
    which is needed for libraries like Docling that use internal threading.
    """

    def get_event_loop(self):
        """Get or create an event loop for the current thread."""
        try:
            return super().get_event_loop()
        except RuntimeError:
            # No event loop in this thread, create one
            loop = self.new_event_loop()
            self.set_event_loop(loop)
            return loop


@worker_process_init.connect
def setup_worker_event_loop(**kwargs):
    """
    Set up event loop handling for each forked worker process.

    When Celery uses prefork pool (default), child workers are created via fork().
    The parent process's event loop state is inherited but invalid in the child.
    This causes "There is no current event loop in thread 'MainThread'" errors
    when libraries like Docling try to use asyncio.

    We use a custom event loop policy that automatically creates loops for any
    thread that needs one, including internal threads spawned by libraries.
    """
    # Set custom policy that auto-creates loops for threads
    asyncio.set_event_loop_policy(ThreadSafeEventLoopPolicy())

    # Create a new event loop for the main thread
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

# Create Celery app instance
celery_app = Celery(
    "dsol_workers",
    broker=settings.redis_url,
    backend=settings.redis_url,
    include=[
        "src.workers.tasks",
        "src.workers.extraction_tasks",
        "src.workers.extraction_v2_tasks",
    ],
)

# Celery configuration
celery_app.conf.update(
    # Task execution settings
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    # Task result settings
    result_expires=3600,  # Results expire after 1 hour
    result_backend_transport_options={
        "master_name": "mymaster",
    },
    # Worker settings
    worker_prefetch_multiplier=1,  # Fetch one task at a time
    worker_max_tasks_per_child=1000,  # Restart worker after 1000 tasks
    # Use default queue for all tasks (simpler setup)
    # For production with separate scaling needs, add task_routes with -Q flag
    task_default_queue="celery",
    # Retry settings
    task_acks_late=True,  # Acknowledge task after completion
    task_reject_on_worker_lost=True,  # Requeue if worker crashes
)