"""MCP service layer for PDF ingestion and Supabase storage.

This module contains the synchronous ingestion pipeline wrappers plus async helpers
for running long-running work in background tasks while publishing progress events.
"""

import asyncio
import hashlib
import logging
import os
import re
import tempfile
import time
import uuid
from collections.abc import Awaitable, Callable
from typing import Any

from fastapi import HTTPException, status

from app.core.config import get_settings
from app.core.progress import progress_manager

logger = logging.getLogger(__name__)

_SAFE_STORAGE_CHARS_RE = re.compile(r"[^A-Za-z0-9._-]+")


def normalize_ingestion_project_id(project_id: str | None) -> str:
    """Normalize the canonical project boundary for ingestion services."""
    normalized = (project_id or "").strip()
    return normalized or "workspace-dev-01"


def _to_safe_storage_filename(filename: str) -> str:
    """Return a conservative, URL/path-safe filename for Supabase Storage.

    Supabase Storage paths are URL-addressable; characters like ':' and various
    unicode punctuation can trigger 400s depending on client/server encoding.
    """
    name = (filename or "upload.pdf").strip()
    name = name.replace("\\", "/").split("/")[-1]  # basename
    name = _SAFE_STORAGE_CHARS_RE.sub("_", name)
    name = re.sub(r"_+", "_", name).strip("._-")
    if not name.lower().endswith(".pdf"):
        name = f"{name}.pdf"
    return name or "upload.pdf"


def _compute_md5(data: bytes) -> str:
    """Compute MD5 hash of raw bytes for idempotency checks.

    Args:
        data: Raw file bytes.

    Returns:
        Hex-encoded MD5 hash string.
    """
    return hashlib.md5(data).hexdigest()


def _to_safe_storage_path_segment(value: str) -> str:
    """Sanitize an arbitrary string for use as a Supabase Storage path segment."""

    segment = (value or "").strip()
    segment = segment.replace("\\", "/").split("/")[-1]
    segment = _SAFE_STORAGE_CHARS_RE.sub("_", segment)
    segment = re.sub(r"_+", "_", segment).strip("._-")
    return segment or "unknown"


async def upload_to_supabase_storage_with_path(
    pdf_bytes: bytes,
    filename: str,
    user_id: str | None = None,
    project_id: str | None = None,
) -> tuple[str, str]:
    """Upload PDF bytes to Supabase Storage bucket.

    Runs the synchronous Supabase SDK upload in a thread executor to avoid blocking
    the async event loop.

    Args:
        pdf_bytes: Raw PDF file content.
        filename: Target filename in the storage bucket.

    Returns:
        tuple[str, str]: (public_url, storage_path)

    Raises:
        HTTPException: When upload fails.
    """

    project_id = normalize_ingestion_project_id(project_id)
    settings = get_settings()

    def _sync_upload() -> tuple[str, str]:
        from supabase import create_client

        client = create_client(
            settings.supabase_url,
            settings.supabase_service_role_key,
        )
        bucket = client.storage.from_(settings.supabase_bucket)

        safe_name = _to_safe_storage_filename(filename)
        if user_id and project_id:
            safe_user = _to_safe_storage_path_segment(user_id)
            safe_project = _to_safe_storage_path_segment(project_id)
            storage_path = f"{safe_user}/{safe_project}/{safe_name}"
        else:
            storage_path = safe_name

        try:
            bucket.upload(
                path=storage_path,
                file=pdf_bytes,
                # storage3 builds request headers from file_options; header values must be str/bytes
                file_options={"content-type": "application/pdf", "upsert": "true"},
            )
        except Exception as exc:
            error_msg = str(exc)
            if "Duplicate" not in error_msg and "already exists" not in error_msg:
                logger.info(
                    "Supabase storage upload failed",
                    extra={"pdf_filename": filename, "storage_path": storage_path, "error": error_msg},
                )
                raise HTTPException(
                    status_code=status.HTTP_502_BAD_GATEWAY,
                    detail=f"Storage upload failed: {error_msg}",
                ) from exc
            logger.info(
                "File already exists in storage, skipping upload",
                extra={"pdf_filename": filename, "storage_path": storage_path},
            )

        public_url = bucket.get_public_url(storage_path)
        logger.info(
            "PDF uploaded to Supabase Storage",
            extra={"pdf_filename": filename, "storage_path": storage_path, "url": public_url},
        )
        return public_url, storage_path

    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, _sync_upload)


async def upload_to_supabase_storage(
    pdf_bytes: bytes,
    filename: str,
    user_id: str | None = None,
    project_id: str | None = None,
) -> str:
    """Upload PDF bytes to Supabase Storage bucket.

    Args:
        pdf_bytes: Raw PDF file content.
        filename: Target filename in the storage bucket.

    Returns:
        str: Public URL of the uploaded file.
    """

    public_url, _ = await upload_to_supabase_storage_with_path(
        pdf_bytes,
        filename,
        user_id=user_id,
        project_id=project_id,
    )
    return public_url


ProgressCallback = Callable[[str, int, str | None], Awaitable[None]]


async def ingest_pdf_to_neo4j(
    pdf_bytes: bytes,
    filename: str,
    project_id: str,
    user_id: str | None = None,
    progress_callback: ProgressCallback | None = None,
    storage_path: str | None = None,
    storage_url: str | None = None,
) -> dict[str, Any]:
    """Process PDF through ingestion pipeline and store in Neo4j.

    Writes bytes to a temporary file (required by hybrid_parser), then runs the
    extraction pipeline in executor threads. Progress can be reported via an
    optional async callback.

    Args:
        pdf_bytes: Raw PDF file content.
        filename: Original filename for metadata.
        project_id: Project identifier for Neo4j scoping.
        user_id: Optional user id for composite isolation.
        progress_callback: Optional callback to publish coarse-grained progress.

    Returns:
        dict[str, Any]: Dictionary with node_id, title, and pillar texts.

    Raises:
        HTTPException: When parsing or extraction fails.
    """

    from app.core.database import _driver
    from app.services.ingestion import (
        compute_composite_id,
        create_gateway_llm_client,
        hybrid_parser,
        process_io_bound_pdf,
    )

    settings = get_settings()
    tmp_path = None
    file_hash = _compute_md5(pdf_bytes)

    async def _emit(stage: str, progress: int, message: str) -> None:
        if progress_callback is not None:
            await progress_callback(stage, progress, message)

    def _sync_idempotency_lookup() -> dict[str, Any] | None:
        if _driver is None:
            return None

        composite_id = None
        if user_id:
            composite_id = compute_composite_id(user_id=user_id, project_id=project_id, file_hash=file_hash)

        with _driver.session() as session:
            if composite_id:
                rec = session.run(
                    "MATCH (n:Literature {composite_id: $cid}) "
                    "RETURN elementId(n) AS nid, n.status AS status, n.title AS title, "
                    "n.text_background AS bg, n.text_methodology AS met, "
                    "n.text_gap AS gap, n.text_objective AS obj, "
                    "n.text_futurework AS fut, coalesce(n.edges_generated, false) AS eg LIMIT 1",
                    cid=composite_id,
                ).single()
                # Only treat as idempotent when the node is completed (or legacy nodes without status).
                if rec is not None and (rec["status"] in (None, "COMPLETED")):
                    return {
                        "is_cached": True,
                        "edges_completed": rec["eg"],
                        "node_id": rec["nid"],
                        "title": rec["title"] or "",
                        "background": rec["bg"] or "",
                        "methodology": rec["met"] or "",
                        "gap": rec["gap"] or "",
                        "objective": rec["obj"] or "",
                        "futurework": rec["fut"] or "",
                    }
                return None
            else:
                rec = session.run(
                    "MATCH (n:Literature {hash: $h}) "
                    "RETURN elementId(n) AS nid, n.status AS status, n.title AS title, "
                    "n.text_background AS bg, n.text_methodology AS met, "
                    "n.text_gap AS gap, n.text_objective AS obj, "
                    "n.text_futurework AS fut, coalesce(n.edges_generated, false) AS eg LIMIT 1",
                    h=file_hash,
                ).single()
                if rec is None or (rec["status"] not in (None, "COMPLETED")):
                    return None
                return {
                    "is_cached": True,
                    "edges_completed": rec["eg"],
                    "node_id": rec["nid"],
                    "title": rec["title"] or "",
                    "background": rec["bg"] or "",
                    "methodology": rec["met"] or "",
                    "gap": rec["gap"] or "",
                    "objective": rec["obj"] or "",
                    "futurework": rec["fut"] or "",
                }

    _LOCK_STALE_SECONDS = 30 * 60
    _LOCK_POLL_SECONDS = 2

    def _sync_get_node_state() -> dict[str, Any] | None:
        if _driver is None:
            return None

        composite_id = None
        if user_id:
            composite_id = compute_composite_id(user_id=user_id, project_id=project_id, file_hash=file_hash)

        with _driver.session() as session:
            if composite_id:
                rec = session.run(
                    "MATCH (n:Literature {composite_id: $cid}) "
                    "RETURN elementId(n) AS nid, n.status AS status, n.status_updated_at AS ts LIMIT 1",
                    cid=composite_id,
                ).single()
            else:
                rec = session.run(
                    "MATCH (n:Literature {hash: $h}) "
                    "RETURN elementId(n) AS nid, n.status AS status, n.status_updated_at AS ts LIMIT 1",
                    h=file_hash,
                ).single()

        if rec is None:
            return None
        return {"node_id": rec["nid"], "status": rec["status"], "status_updated_at": rec["ts"]}

    def _sync_create_processing_node() -> None:
        if _driver is None:
            return

        composite_id = None
        if user_id:
            composite_id = compute_composite_id(user_id=user_id, project_id=project_id, file_hash=file_hash)

        with _driver.session() as session:
            if composite_id:
                session.run(
                    "MERGE (n:Literature {composite_id: $cid}) "
                    "ON CREATE SET n.hash=$h, n.user_id=$user_id, n.project_id=$project_id, n.filename=$filename "
                    "SET n.status='PROCESSING', n.status_updated_at=timestamp()",
                    cid=composite_id,
                    h=file_hash,
                    user_id=user_id,
                    project_id=project_id,
                    filename=filename,
                )
            else:
                session.run(
                    "MERGE (n:Literature {hash: $h}) "
                    "SET n.project_id=$project_id, n.filename=$filename, n.status='PROCESSING', n.status_updated_at=timestamp()",
                    h=file_hash,
                    project_id=project_id,
                    filename=filename,
                )

    def _sync_set_processing_node() -> None:
        if _driver is None:
            return

        composite_id = None
        if user_id:
            composite_id = compute_composite_id(user_id=user_id, project_id=project_id, file_hash=file_hash)

        with _driver.session() as session:
            if composite_id:
                session.run(
                    "MATCH (n:Literature {composite_id: $cid}) "
                    "SET n.status='PROCESSING', n.status_updated_at=timestamp()",
                    cid=composite_id,
                )
            else:
                session.run(
                    "MATCH (n:Literature {hash: $h}) "
                    "SET n.status='PROCESSING', n.status_updated_at=timestamp()",
                    h=file_hash,
                )

    async def _acquire_or_wait_processing_lock() -> None:
        if _driver is None:
            return

        loop = asyncio.get_event_loop()

        while True:
            state = await loop.run_in_executor(None, _sync_get_node_state)
            if state is None:
                await loop.run_in_executor(None, _sync_create_processing_node)
                return

            status_val = state.get("status")
            # Legacy nodes without status are treated as completed for compatibility.
            if status_val in (None, "COMPLETED"):
                return

            ts = state.get("status_updated_at")
            now_ms = int(time.time() * 1000)
            stale_ms = _LOCK_STALE_SECONDS * 1000
            is_fresh_processing = status_val == "PROCESSING" and isinstance(ts, (int, float)) and (now_ms - int(ts)) < stale_ms

            if is_fresh_processing:
                await _emit("INITIALIZING", 10, "Waiting for existing processing job")
                await asyncio.sleep(_LOCK_POLL_SECONDS)
                continue

            await loop.run_in_executor(None, _sync_set_processing_node)
            return

    def _sync_parse_pdf(path: str) -> str:
        return hybrid_parser.parse_pdf(path)

    def _sync_run_pipeline(raw_markdown: str) -> str:
        metrics = {
            "file_name": filename,
            "file_hash": file_hash,
            "status": "Pending",
            "total_pages": 0,
            "time_parsing": 0.0,
            "time_retrieval": 0.0,
            "time_llm": 0.0,
            "time_edge_gen": 0.0,
            "time_db": 0.0,
            "time_supabase": 0.0,
            "time_total": 0.0,
            "token_embed_in": 0,
            "token_embed_out": 0,
            "token_llm_in": 0,
            "token_llm_out": 0,
            "error_msg": "",
        }

        llm_client = create_gateway_llm_client()
        from app.services.storage.supabase_client import start_uploader_worker
        upload_queue, _ = start_uploader_worker()

        t_start = time.time()
        node_id, result_metrics = process_io_bound_pdf(
            pdf_path=tmp_path,
            raw_markdown=raw_markdown,
            metrics=metrics,
            t_start_total=t_start,
            driver=_driver,
            llm_client=llm_client,
            upload_queue=upload_queue,
            project_id=project_id,
            user_id=user_id,
            storage_path=storage_path,
            storage_url=storage_url,
            original_filename=filename,
        )

        if node_id is None:
            error_msg = (result_metrics or {}).get("error_msg", "Unknown error")
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=f"Ingestion failed: {error_msg}",
            )

        return node_id

    def _sync_fetch_pillars(node_id: str) -> dict[str, Any]:
        if _driver is None:
            return {"title": "", "background": "", "methodology": "", "gap": "", "objective": "", "futurework": ""}

        with _driver.session() as session:
            rec = session.run(
                "MATCH (n:Literature) WHERE elementId(n) = $nid "
                "RETURN n.title AS title, n.text_background AS bg, "
                "n.text_methodology AS met, n.text_gap AS gap, "
                "n.text_objective AS obj, n.text_futurework AS fut",
                nid=node_id,
            ).single()

        return {
            "title": (rec["title"] if rec else "") or "",
            "background": (rec["bg"] if rec else "") or "",
            "methodology": (rec["met"] if rec else "") or "",
            "gap": (rec["gap"] if rec else "") or "",
            "objective": (rec["obj"] if rec else "") or "",
            "futurework": (rec["fut"] if rec else "") or "",
        }

    try:
        await _emit("INITIALIZING", 5, "Acquiring processing lock")

        existing = await asyncio.get_event_loop().run_in_executor(None, _sync_idempotency_lookup)
        if existing is not None:
            logger.info(
                "Node already exists, skipping ingestion",
                extra={"file_hash": file_hash, "pdf_filename": filename, "project_id": project_id, "user_id": user_id},
            )
            await _emit("COMPLETED", 100, "Already ingested")
            return existing

        await _acquire_or_wait_processing_lock()
        await _emit("PARSING_PDF", 30, "Parsing PDF")

        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp_file:
            tmp_file.write(pdf_bytes)
            tmp_path = tmp_file.name

        raw_markdown = await asyncio.get_event_loop().run_in_executor(None, _sync_parse_pdf, tmp_path)
        if not raw_markdown:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail="Failed to extract text from PDF",
            )

        await _emit("AI_EXTRACTION", 50, "Running extraction pipeline")
        node_id = await asyncio.get_event_loop().run_in_executor(None, _sync_run_pipeline, raw_markdown)

        await _emit("EMBEDDING_PILLARS", 75, "Embedding and storing pillars")
        pillars = await asyncio.get_event_loop().run_in_executor(None, _sync_fetch_pillars, node_id)

        logger.info(
            "PDF ingestion completed",
            extra={"pdf_filename": filename, "node_id": node_id, "project_id": project_id, "user_id": user_id},
        )

        return {"node_id": node_id, **pillars}

    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)


async def delete_from_supabase_storage(storage_path: str) -> None:
    """Best-effort delete a file from Supabase Storage.

    Args:
        storage_path: Path inside the bucket.

    Raises:
        HTTPException: When deletion fails.
    """

    settings = get_settings()

    def _sync_delete() -> None:
        from supabase import create_client

        client = create_client(
            settings.supabase_url,
            settings.supabase_service_role_key,
        )
        bucket = client.storage.from_(settings.supabase_bucket)
        bucket.remove([storage_path])

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, _sync_delete)


async def run_pdf_job_background(
    job_id: str,
    pdf_bytes: bytes,
    filename: str,
    project_id: str,
    user_id: str,
) -> None:
    """Run upload+ingestion in background and publish progress to ProgressManager.

    Args:
        job_id: Progress job id.
        pdf_bytes: Raw PDF bytes.
        filename: Original filename.
        project_id: Project identifier.
        user_id: Supabase user id.
    """

    project_id = normalize_ingestion_project_id(project_id)
    storage_path: str | None = None
    storage_url: str | None = None
    completed = False
    file_hash = _compute_md5(pdf_bytes)

    await progress_manager.init_job(job_id, owner_user_id=user_id)

    async def _publish(stage: str, progress: int, message: str | None) -> None:
        await progress_manager.update(job_id, stage=stage, progress=progress, message=message)

    async def _log_progress_db_correlation(phase: str, *, error: str | None = None) -> None:
        from app.services.storage.supabase_client import get_document_chunk_status_summary

        try:
            progress_snapshot = await progress_manager.snapshot(job_id)
        except KeyError:
            progress_snapshot = None

        db_summary = await asyncio.to_thread(
            get_document_chunk_status_summary,
            file_hash,
            user_id,
            project_id,
        )
        logger.info(
            "Progress and document chunk status correlation",
            extra={
                "phase": phase,
                "job_id": job_id,
                "project_id": project_id,
                "user_id": user_id,
                "file_hash": file_hash,
                "progress_snapshot": progress_snapshot.model_dump() if progress_snapshot else None,
                "document_chunk_summary": db_summary,
                "error": error,
            },
        )

    try:
        await _publish("INITIALIZING", 0, "Job accepted")

        await _publish("STORING_BINARY", 10, "Uploading PDF to storage")
        storage_url, storage_path = await upload_to_supabase_storage_with_path(
            pdf_bytes,
            filename,
            user_id=user_id,
            project_id=project_id,
        )
        await _publish("STORING_BINARY", 20, "PDF stored")

        result = await ingest_pdf_to_neo4j(
            pdf_bytes=pdf_bytes,
            filename=filename,
            project_id=project_id,
            user_id=user_id,
            progress_callback=_publish,
            storage_path=storage_path,
            storage_url=storage_url,
        )

        node_id = result.get("node_id")
        
        # TLM.4: Orchestrator Gate - Cek apakah relasi sudah pernah digenerate di masa lalu
        edges_completed = False
        if node_id:
            try:
                # Perlu mengecek status terbaru ke DB, karena 'existing' hanya didefinisikan 
                # di dalam process_pdf_parallel yang melempar job ini. 
                # Jika dokumen benar-benar baru, otomatis edges_completed = False.
                from app.core.database import _driver
                if _driver:
                    with _driver.session() as session:
                        rec = session.run(
                            "MATCH (n:Literature) WHERE elementId(n) = $nid "
                            "RETURN coalesce(n.edges_generated, false) AS eg LIMIT 1",
                            nid=node_id
                        ).single()
                        if rec:
                            edges_completed = rec["eg"]
            except Exception as e:
                logger.warning(f"Gagal memverifikasi status edges_generated untuk node {node_id}: {e}")
        
        if node_id and not edges_completed:
            await _publish("GENERATING_EDGES", 85, "Mencari koneksi dan reasoning AI")
            
            from app.core.database import _driver
            from app.services.ingestion import create_gateway_llm_client, worker_edge_generator

            llm_client = create_gateway_llm_client()

            await asyncio.get_event_loop().run_in_executor(
                None,
                worker_edge_generator,
                node_id,
                _driver,
                llm_client
            )
        elif node_id and edges_completed:
            await _publish("GENERATING_EDGES", 85, "Koneksi dan reasoning AI sudah tersedia (Skipped)")

        await _publish("FINALIZING_GRAPH", 90, "Finalizing")

        from app.services.storage.supabase_client import start_uploader_worker
        upload_queue, _ = start_uploader_worker()
        if upload_queue:
            await asyncio.to_thread(upload_queue.join)

        from app.core.database import _driver
        from app.services.pagerank import calculate_visual_pagerank

        try:
            pagerank_result = await asyncio.get_event_loop().run_in_executor(
                None,
                calculate_visual_pagerank,
                _driver,
                project_id,
                user_id,
            )
        except Exception:
            logger.error(
                "PageRank finalization failed",
                extra={
                    "job_id": job_id,
                    "project_id": project_id,
                    "user_id": user_id,
                },
                exc_info=True,
            )
            raise
        logger.info(
            "Visual PageRank integrated into background PDF job",
            extra={
                "job_id": job_id,
                "project_id": project_id,
                "user_id": user_id,
                "pagerank_result": pagerank_result,
            },
        )

        await mark_literature_completed(
            file_hash=file_hash,
            project_id=project_id,
            user_id=user_id,
        )

        await _log_progress_db_correlation("FINALIZING_GRAPH_COMPLETE")
        await progress_manager.complete(
            job_id,
            result={
                "node_id": result.get("node_id", ""),
                "storage_url": storage_url or "",
                "title": result.get("title", ""),
            },
        )
        completed = True

    except HTTPException as exc:
        logger.error(
            "Background PDF job failed",
            extra={"job_id": job_id, "pdf_filename": filename, "project_id": project_id, "error": str(exc.detail)},
            exc_info=True,
        )
        try:
            await mark_literature_failed(
                file_hash=file_hash,
                project_id=project_id,
                user_id=user_id,
                filename=filename,
                error=str(exc.detail),
            )
        except Exception:
            logger.error(
                "Failed to mark Neo4j node as FAILED",
                extra={"job_id": job_id, "project_id": project_id},
                exc_info=True,
            )

        await _log_progress_db_correlation("FAILED_HTTP_EXCEPTION", error=str(exc.detail))
        await progress_manager.fail(job_id, error=str(exc.detail))

    except Exception as exc:
        logger.error(
            "Background PDF job failed (unexpected)",
            extra={"job_id": job_id, "pdf_filename": filename, "project_id": project_id, "error": type(exc).__name__},
            exc_info=True,
        )
        try:
            await mark_literature_failed(
                file_hash=file_hash,
                project_id=project_id,
                user_id=user_id,
                filename=filename,
                error=str(exc),
            )
        except Exception:
            logger.error(
                "Failed to mark Neo4j node as FAILED",
                extra={"job_id": job_id, "project_id": project_id},
                exc_info=True,
            )

        await _log_progress_db_correlation("FAILED_UNEXPECTED", error=str(exc))
        await progress_manager.fail(job_id, error=str(exc))

    finally:
        if not completed and storage_path:
            try:
                await delete_from_supabase_storage(storage_path)
                logger.info(
                    "Rolled back storage file after failure",
                    extra={"job_id": job_id, "storage_path": storage_path},
                )
            except Exception:
                logger.error(
                    "Failed to rollback storage file after failure",
                    extra={"job_id": job_id, "storage_path": storage_path},
                    exc_info=True,
                )


async def mark_literature_failed(
    file_hash: str,
    project_id: str,
    user_id: str | None,
    filename: str,
    error: str,
) -> None:
    """Best-effort update Neo4j node status to FAILED.

    This prevents stuck PROCESSING nodes and supports self-healing on retry.
    """

    from app.core.database import _driver
    from app.services.ingestion import compute_composite_id

    if _driver is None:
        return

    composite_id = None
    if user_id:
        composite_id = compute_composite_id(user_id=user_id, project_id=project_id, file_hash=file_hash)

    def _sync_mark_failed() -> None:
        with _driver.session() as session:
            if composite_id:
                session.run(
                    "MATCH (n:Literature {composite_id: $cid}) "
                    "SET n.status='FAILED', n.status_updated_at=timestamp(), n.error_msg=$error, n.filename=coalesce(n.filename,$filename) "
                    "RETURN 1",
                    cid=composite_id,
                    error=error,
                    filename=filename,
                )
            else:
                session.run(
                    "MATCH (n:Literature {hash: $h}) "
                    "SET n.status='FAILED', n.status_updated_at=timestamp(), n.error_msg=$error, n.filename=coalesce(n.filename,$filename) "
                    "RETURN 1",
                    h=file_hash,
                    error=error,
                    filename=filename,
                )

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, _sync_mark_failed)


async def mark_literature_completed(
    file_hash: str,
    project_id: str,
    user_id: str | None,
) -> None:
    """Best-effort update Neo4j node status to COMPLETED."""
    from app.core.database import _driver
    from app.services.ingestion import compute_composite_id

    if _driver is None:
        return

    composite_id = None
    if user_id:
        composite_id = compute_composite_id(user_id=user_id, project_id=project_id, file_hash=file_hash)

    def _sync_mark_completed() -> None:
        with _driver.session() as session:
            if composite_id:
                session.run(
                    "MATCH (n:Literature {composite_id: $cid}) "
                    "SET n.status='COMPLETED', n.status_updated_at=timestamp() "
                    "RETURN 1",
                    cid=composite_id,
                )
            else:
                session.run(
                    "MATCH (n:Literature {hash: $h}) "
                    "SET n.status='COMPLETED', n.status_updated_at=timestamp() "
                    "RETURN 1",
                    h=file_hash,
                )

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, _sync_mark_completed)


def generate_job_id() -> str:
    """Generate a job id for progress tracking."""

    return uuid.uuid4().hex


async def process_pdf_parallel(
    pdf_bytes: bytes,
    filename: str,
    project_id: str,
    user_id: str | None = None,
) -> dict[str, Any]:
    """Orchestrate parallel PDF upload and ingestion.

    Runs Supabase Storage upload and Neo4j ingestion concurrently
    using asyncio.gather. Both operations receive the same PDF bytes
    from memory without any redundant network downloads.

    Args:
        pdf_bytes: Raw PDF file content from client upload.
        filename: Original filename.
        project_id: Project identifier for Neo4j scoping.

    Returns:
        Combined result with storage URL and pillar data.
    """
    project_id = normalize_ingestion_project_id(project_id)

    # Run storage first to avoid doing expensive ingestion if we cannot persist the PDF.
    # This also prevents "partial success" where Neo4j nodes are created even though
    # the API returns 502 to the client.
    storage_url = await upload_to_supabase_storage(
        pdf_bytes,
        filename,
        user_id=user_id,
        project_id=project_id,
    )
    pillar_data = await ingest_pdf_to_neo4j(
        pdf_bytes,
        filename,
        project_id,
        user_id=user_id,
    )

    return {
        "status": "success",
        "node_id": pillar_data.get("node_id", ""),
        "storage_url": storage_url,
        "title": pillar_data.get("title", ""),
        "pillars": {
            "background": pillar_data.get("background", ""),
            "methodology": pillar_data.get("methodology", ""),
            "gap": pillar_data.get("gap", ""),
            "objective": pillar_data.get("objective", ""),
            "futurework": pillar_data.get("futurework", ""),
        },
    }
