import asyncio
import logging
import threading
from typing import Any, Dict, Optional

from app.services.ai.ai_gateway import AIGateway, create_ai_gateway

logger = logging.getLogger(__name__)

MIN_POLLING = 1.0
MAX_POLLING = 60.0


class GhostWorker:
    """Background worker for generating vector embeddings for document chunks."""

    def __init__(self) -> None:
        """Initialize the ghost worker with an AI Gateway."""
        self.gateway: AIGateway = create_ai_gateway()
        self.shutdown_event: asyncio.Event = asyncio.Event()
        self.wakeup_event: asyncio.Event = asyncio.Event()
        self.worker_loop: asyncio.AbstractEventLoop | None = None
        self.loop_thread_name: str | None = None
        self._pending_wakeup = False

    def _schedule_wakeup(self, reason: str) -> bool:
        """Schedule a wakeup on the active worker loop in a thread-safe way."""
        loop = self.worker_loop
        current_thread = threading.current_thread().name

        if loop is None:
            self._pending_wakeup = True
            logger.warning(
                "GhostWorker wakeup requested before worker loop is ready",
                extra={
                    "reason": reason,
                    "thread": current_thread,
                    "pending_wakeup": self._pending_wakeup,
                },
            )
            return False

        if loop.is_closed():
            logger.warning(
                "GhostWorker wakeup skipped because worker loop is closed",
                extra={
                    "reason": reason,
                    "thread": current_thread,
                    "loop_thread": self.loop_thread_name,
                },
            )
            return False

        loop.call_soon_threadsafe(self.wakeup_event.set)
        logger.info(
            "GhostWorker wakeup scheduled on active loop",
            extra={
                "reason": reason,
                "thread": current_thread,
                "loop_thread": self.loop_thread_name,
                "loop_id": id(loop),
            },
        )
        return True

    def trigger_wakeup(self) -> None:
        """Wake the worker so the main loop can re-check the queue.

        Returns:
            None.
        """
        logger.info(
            "GhostWorker wakeup triggered",
            extra={
                "worker_thread": threading.current_thread().name,
                "loop_ready": self.worker_loop is not None,
            },
        )
        self._schedule_wakeup("trigger_wakeup")

    async def run_ghost_worker_loop(self) -> None:
        """
        Main loop for the ghost worker.
        Uses adaptive polling to minimize database load when idle.
        """
        import time
        self.worker_loop = asyncio.get_running_loop()
        self.loop_thread_name = threading.current_thread().name
        logger.info(
            "GhostWorker loop starting",
            extra={
                "thread": self.loop_thread_name,
                "loop_id": id(self.worker_loop),
                "pending_wakeup": self._pending_wakeup,
            },
        )
        if self._pending_wakeup:
            self._pending_wakeup = False
            logger.info(
                "GhostWorker applying pending wakeup captured before loop readiness",
                extra={
                    "thread": self.loop_thread_name,
                    "loop_id": id(self.worker_loop),
                },
            )
            self.wakeup_event.set()
        interval = MIN_POLLING
        last_pulse = time.time()

        while not self.shutdown_event.is_set():
            try:
                # Pulse log every 5 minutes (300 seconds)
                current_time = time.time()
                if current_time - last_pulse >= 300:
                    logger.info("GhostWorker Pulse: Worker is active and polling.")
                    last_pulse = current_time

                self.wakeup_event.clear()

                # 1. Claim a document batch (max 20 chunks)
                batch_info = await self.claim_next_document_batch(limit=20)

                if batch_info:
                    logger.info(
                        "GhostWorker claimed document batch",
                        extra={
                            "file_hash": batch_info.get("file_hash"),
                            "project_id": batch_info.get("project_id"),
                            "user_id": batch_info.get("user_id"),
                            "chunk_count": len(batch_info.get("chunks", [])),
                        },
                    )
                    # 2. Integrity Check with Neo4j
                    file_hash = batch_info.get("file_hash", "")
                    if file_hash and not await self.audit_integrity(file_hash):
                        continue  # Already handled (downgraded)

                    # 3. Process each chunk with throttling
                    await self.process_batch(batch_info)
                    interval = MIN_POLLING  # Stay fast
                else:
                    # 4. Backoff if idle
                    interval = min(MAX_POLLING, interval * 2)

            except Exception as e:
                logger.error(f"Ghost Worker Error: {e}", exc_info=True)
                await asyncio.sleep(60)  # Panic cooldown

            if self.shutdown_event.is_set():
                break

            try:
                await asyncio.wait_for(self.wakeup_event.wait(), timeout=interval)
                logger.info(
                    "GhostWorker woke up early to re-check queue",
                    extra={
                        "worker_thread": threading.current_thread().name,
                        "interval": interval,
                    },
                )
            except asyncio.TimeoutError:
                pass

        logger.info("GhostWorker loop stopped cleanly.")

    async def claim_next_document_batch(
        self, limit: int = 20
    ) -> Optional[Dict[str, Any]]:
        """
        Claim a batch of pending chunks from the database.
        To be implemented with FOR UPDATE SKIP LOCKED.
        """
        from app.services.storage.supabase_client import claim_next_document_batch
        return await claim_next_document_batch(limit=limit)

    async def audit_integrity(self, file_hash: str) -> bool:
        """
        Audit data integrity between Postgres and Neo4j.
        Returns False if integrity is broken and status was downgraded.
        """
        import asyncio
        from app.services.storage.supabase_client import check_document_chunks_exist_by_hash
        from app.services.mcp_service import mark_literature_failed
        
        # We need project_id and user_id to correctly downgrade Neo4j node.
        # But this function only receives file_hash for now based on the previous plan.
        # To be safe, we will just use the hash and leave project_id empty 
        # (the fallback logic in mark_literature_failed will use the hash).
        
        has_chunks = await asyncio.to_thread(check_document_chunks_exist_by_hash, file_hash)
        
        if not has_chunks:
            logger.warning(f"Integrity check failed: No chunks found for file_hash {file_hash} in Postgres. Downgrading Neo4j node to FAILED.")
            try:
                await mark_literature_failed(
                    file_hash=file_hash,
                    project_id="",
                    user_id=None,
                    filename="Unknown (Recovered by GhostWorker)",
                    error="Data Integrity Failure: Document chunks missing in Postgres."
                )
            except Exception as e:
                logger.error(f"Error downgrading Neo4j node for {file_hash}: {e}")
            return False
            
        return True

    async def process_batch(self, batch_info: Dict[str, Any]) -> None:
        """
        Process embeddings for the claimed batch via AI Gateway.
        """
        file_hash = batch_info.get("file_hash")
        chunks = batch_info.get("chunks", [])
        logger.info(f"Processing embedding for {len(chunks)} chunks of file_hash: {file_hash}")

        engine = self.gateway.get_embedding_engine()
        
        processed_chunks = []

        for chunk in chunks:
            text_to_embed = chunk.get("content", "")
            if not text_to_embed.strip():
                continue

            try:
                # Run sync embed in a thread to prevent blocking the async loop
                embedding = await asyncio.to_thread(engine.embed, text_to_embed[:8000])
                
                processed_chunk = {
                    "id": chunk["id"],
                    "embedding": embedding,
                    "embedding_status": "COMPLETED"
                }
                processed_chunks.append(processed_chunk)
                
            except Exception as e:
                logger.error(f"Failed to embed chunk {chunk.get('id')}: {e}")
                processed_chunk = {
                    "id": chunk["id"],
                    "error_log": str(e),
                    "embedding_status": "FAILED"
                }
                processed_chunks.append(processed_chunk)
                
            # Throttling aktif (TPM safe)
            await asyncio.sleep(1.0)
            
        from app.services.storage.supabase_client import update_chunks_batch
        success = await update_chunks_batch(processed_chunks)
        if success:
            status_counts = {
                "COMPLETED": sum(
                    1 for chunk in processed_chunks if chunk.get("embedding_status") == "COMPLETED"
                ),
                "FAILED": sum(
                    1 for chunk in processed_chunks if chunk.get("embedding_status") == "FAILED"
                ),
            }
            logger.info(
                "Finished embedding loop and updated chunks for file_hash",
                extra={
                    "file_hash": file_hash,
                "updated_chunks": len(processed_chunks),
                "status_completed": status_counts["COMPLETED"],
                "status_failed": status_counts["FAILED"],
                "worker_thread": threading.current_thread().name,
            },
        )
        else:
            logger.error(f"Failed to update database for {len(processed_chunks)} chunks for file_hash: {file_hash}")

# Global instance
_ghost_worker_instance: Optional[GhostWorker] = None

def get_ghost_worker() -> GhostWorker:
    """Get or create the global GhostWorker instance."""
    global _ghost_worker_instance
    if _ghost_worker_instance is None:
        _ghost_worker_instance = GhostWorker()
    return _ghost_worker_instance
