"""Durable compile job manager — Postgres-backed.

Jobs survive server restarts. The store is the source of truth for job
status; callers that need a same-process fast path use the wrapper in
`compile_jobs.py` (`get_job_durable` consults a small process-local
cache before this module).

Database errors propagate. If Postgres is unhealthy a compile request
returns 5xx — not "submitted but invisible to the next process". The
operator sees the failure and fixes the deploy instead of running blind.

Public API:
- submit_job(subject_id) → CompileJob
- get_job(job_id) → CompileJob | None
- mark_running(job_id)
- mark_completed(job_id, memories_created, memories)
- mark_failed(job_id, error)
"""

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from dataclasses import dataclass, field
from enum import Enum
from typing import Any

import structlog
from sqlalchemy import delete, select, update

from server.db.engine import get_session_factory
from server.db.tables import CompileJobRow

logger = structlog.stdlib.get_logger()


class JobStatus(str, Enum):
    pending = "pending"
    running = "running"
    completed = "completed"
    failed = "failed"


@dataclass
class CompileJob:
    id: str
    subject_id: str
    tenant_id: str | None = None
    status: JobStatus = JobStatus.pending
    memories_created: int = 0
    memories: list[dict[str, Any]] = field(default_factory=list)
    error: str | None = None
    created_at: float = field(default_factory=lambda: datetime.now(timezone.utc).timestamp())
    completed_at: float | None = None


def _row_to_job(row: CompileJobRow) -> CompileJob:
    return CompileJob(
        id=row.id,
        subject_id=row.subject_id,
        tenant_id=row.tenant_id,
        status=JobStatus(row.status),
        memories_created=row.memories_created,
        memories=[],  # Not stored in DB to keep table lean
        error=row.error,
        created_at=row.created_at.timestamp() if row.created_at else 0,
        completed_at=row.completed_at.timestamp() if row.completed_at else None,
    )


async def submit_job(subject_id: str, tenant_id: str | None = None) -> CompileJob:
    """Create a durable compile job and persist to Postgres.

    Raises whatever SQLAlchemy raises on DB failure — we no longer
    return a "submitted" job that was never persisted. Callers must
    surface DB outages to the operator.
    """
    job_id = str(uuid.uuid4())[:8]
    now = datetime.now(timezone.utc)
    async with get_session_factory()() as session:
        row = CompileJobRow(
            id=job_id,
            subject_id=subject_id,
            tenant_id=tenant_id,
            status="pending",
            memories_created=0,
            created_at=now,
        )
        session.add(row)
        await session.commit()
    return CompileJob(
        id=job_id,
        subject_id=subject_id,
        tenant_id=tenant_id,
        created_at=now.timestamp(),
    )


async def get_job(job_id: str, tenant_id: str | None = None) -> CompileJob | None:
    """Retrieve a job by ID from Postgres.

    Returns None only when the job genuinely does not exist. DB errors
    propagate so a transient outage doesn't masquerade as "not found".
    """
    async with get_session_factory()() as session:
        stmt = select(CompileJobRow).where(CompileJobRow.id == job_id)
        if tenant_id is None:
            stmt = stmt.where(CompileJobRow.tenant_id.is_(None))
        else:
            stmt = stmt.where(CompileJobRow.tenant_id == tenant_id)
        result = await session.execute(stmt)
        row = result.scalar_one_or_none()
        if row is None:
            return None
        return _row_to_job(row)


async def mark_running(job_id: str) -> None:
    """Mark job as running. Raises on DB failure."""
    async with get_session_factory()() as session:
        await session.execute(
            update(CompileJobRow)
            .where(CompileJobRow.id == job_id)
            .values(status="running", started_at=datetime.now(timezone.utc))
        )
        await session.commit()


async def mark_completed(
    job_id: str, memories_created: int, memories: list[dict[str, Any]]
) -> None:
    """Mark job as completed with result count. Raises on DB failure."""
    async with get_session_factory()() as session:
        await session.execute(
            update(CompileJobRow)
            .where(CompileJobRow.id == job_id)
            .values(
                status="completed",
                memories_created=memories_created,
                completed_at=datetime.now(timezone.utc),
            )
        )
        await session.commit()


async def update_progress(job_id: str, memories_created: int) -> None:
    """Update the cumulative `memories_created` count on a running job.

    Called once per batch by the async drain loop (issue #134) so that
    operators polling `/v1/memories/compile/{job_id}` see growing
    progress while a large subject is being drained, rather than a
    static `0` until the job completes minutes or hours later.
    """
    async with get_session_factory()() as session:
        await session.execute(
            update(CompileJobRow)
            .where(CompileJobRow.id == job_id)
            .values(memories_created=memories_created)
        )
        await session.commit()


async def mark_failed(job_id: str, error: str) -> None:
    """Mark job as failed with error message. Raises on DB failure."""
    async with get_session_factory()() as session:
        await session.execute(
            update(CompileJobRow)
            .where(CompileJobRow.id == job_id)
            .values(
                status="failed",
                error=error,
                completed_at=datetime.now(timezone.utc),
            )
        )
        await session.commit()


async def list_jobs(
    status: str | None = None,
    subject_id: str | None = None,
    tenant_id: str | None = None,
    limit: int = 20,
    offset: int = 0,
) -> tuple[list[dict[str, Any]], int]:
    """List compile jobs for admin introspection.

    Returns (jobs_list, total_count) tuple.
    """
    async with get_session_factory()() as session:
        from sqlalchemy import func

        # Build base filter conditions
        conditions = []
        if status:
            conditions.append(CompileJobRow.status == status)
        if subject_id:
            conditions.append(CompileJobRow.subject_id == subject_id)
        if tenant_id is not None:
            conditions.append(CompileJobRow.tenant_id == tenant_id)

        # Get total count
        count_stmt = select(func.count()).select_from(CompileJobRow)
        for cond in conditions:
            count_stmt = count_stmt.where(cond)
        total = await session.scalar(count_stmt) or 0

        # Get paginated results
        stmt = select(CompileJobRow).order_by(CompileJobRow.created_at.desc())
        for cond in conditions:
            stmt = stmt.where(cond)
        stmt = stmt.offset(offset).limit(limit)

        result = await session.execute(stmt)
        rows = result.scalars().all()

        jobs = [
            {
                "job_id": row.id,
                "subject_id": row.subject_id,
                "tenant_id": row.tenant_id,
                "status": row.status,
                "memories_created": row.memories_created,
                "error": row.error,
                "created_at": row.created_at.isoformat() if row.created_at else None,
                "started_at": row.started_at.isoformat() if row.started_at else None,
                "completed_at": row.completed_at.isoformat() if row.completed_at else None,
            }
            for row in rows
        ]
        return jobs, total


async def cleanup_old_jobs(retention_hours: int = 168) -> int:
    """Delete completed/failed jobs older than retention window.

    Default: 7 days (168 hours). Returns count of deleted rows.

    Runs from a long-lived background loop on app startup; we swallow +
    log here because a transient DB blip should not crash the lifespan
    task. Errors surface as ERROR-level logs that the operator can find.
    """
    from datetime import timedelta

    cutoff = datetime.now(timezone.utc) - timedelta(hours=retention_hours)
    try:
        async with get_session_factory()() as session:
            result = await session.execute(
                delete(CompileJobRow)
                .where(CompileJobRow.status.in_(["completed", "failed"]))
                .where(CompileJobRow.created_at < cutoff)
            )
            await session.commit()
            count = result.rowcount  # type: ignore[attr-defined]
            if count:
                logger.info("compile_jobs_cleaned", deleted=count)
            return count
    except Exception:
        logger.error("compile_jobs_cleanup_failed", exc_info=True)
        return 0


# Statuses that are safe to delete on operator request — never wipe a job
# that the worker may still be holding (pending/running) or another caller
# is racing to mark.
TERMINAL_JOB_STATUSES = ("completed", "failed")


async def purge_jobs(
    status: str | None = None,
    subject_id: str | None = None,
    tenant_id: str | None = None,
) -> int:
    """Operator-triggered delete for terminal jobs matching the given filter.

    At least one filter (status, subject_id, tenant_id) must be supplied —
    an empty filter would wipe the entire history of completed/failed jobs
    in a single click. Status, when given, must be a terminal state.
    Returns the number of rows deleted.
    """
    if not (status or subject_id or tenant_id):
        raise ValueError("at least one filter is required")
    if status and status not in TERMINAL_JOB_STATUSES:
        raise ValueError(
            f"status must be one of {TERMINAL_JOB_STATUSES}; got {status!r}"
        )

    async with get_session_factory()() as session:
        stmt = delete(CompileJobRow).where(
            CompileJobRow.status.in_(
                [status] if status else list(TERMINAL_JOB_STATUSES)
            )
        )
        if subject_id:
            stmt = stmt.where(CompileJobRow.subject_id == subject_id)
        if tenant_id:
            stmt = stmt.where(CompileJobRow.tenant_id == tenant_id)
        result = await session.execute(stmt)
        await session.commit()
        count = result.rowcount or 0  # type: ignore[attr-defined]
        logger.info(
            "compile_jobs_purged",
            deleted=count,
            status=status,
            subject_id=subject_id,
            tenant_id=tenant_id,
        )
        return count


async def delete_job(job_id: str) -> bool:
    """Delete a single terminal compile job by ID.

    Returns True if the job was found and deleted, False if not found.
    Raises ValueError if the job is not in a terminal state (completed/failed).
    """
    async with get_session_factory()() as session:
        row = await session.get(CompileJobRow, job_id)
        if row is None:
            return False
        if row.status not in TERMINAL_JOB_STATUSES:
            raise ValueError(
                f"Cannot delete job {job_id!r} with status {row.status!r} — "
                "only terminal jobs (completed, failed) can be deleted"
            )
        subject_id = row.subject_id
        await session.delete(row)
        await session.commit()
        logger.info("compile_job_deleted", job_id=job_id, subject_id=subject_id)
        return True


async def reset_stuck_jobs(threshold_minutes: int = 30) -> int:
    """Reset jobs stuck in `running` state back to `pending`.

    A job is stuck when it has been `running` for longer than
    `threshold_minutes` — the worker that claimed it crashed or was
    restarted before it could mark the job completed or failed.
    Resetting to `pending` makes the next available worker pick it up
    and rerun it cleanly.

    Returns the count of jobs reset.
    """
    from datetime import timedelta

    cutoff = datetime.now(timezone.utc) - timedelta(minutes=threshold_minutes)

    async with get_session_factory()() as session:
        result = await session.execute(
            update(CompileJobRow)
            .where(CompileJobRow.status == "running")
            .where(CompileJobRow.started_at < cutoff)
            .values(status="pending", started_at=None)
        )
        await session.commit()
        count = result.rowcount or 0  # type: ignore[attr-defined]
        logger.info("compile_jobs_stuck_reset", reset=count, threshold_minutes=threshold_minutes)
        return count
