"""Tests for the background job queue.

Covers:
- enqueue_job inserts a pending row
- claim_next_job atomically claims the oldest pending row
- complete_job marks a job done
- fail_job retries then permanently fails
- push/stream endpoint enqueues intel jobs (not asyncio.create_task)
"""
from __future__ import annotations

import hashlib
from datetime import datetime, timezone

import msgpack
import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from muse.core.mpack import MuseWireFrameWriter, grpc_frame
from musehub.db import musehub_models as db
from musehub.services.musehub_jobs import (
    claim_next_job,
    complete_job,
    enqueue_job,
    fail_job,
)
from musehub.types.json_types import StrDict
from tests.factories import create_repo as factory_create_repo

_fw = MuseWireFrameWriter()


def _utc_now() -> datetime:
    return datetime.now(tz=timezone.utc)


def _mp(data: _MsgpackInput) -> bytes:
    return msgpack.packb(data, use_bin_type=True)


def _wrap(ft: str, data: object) -> bytes:
    return grpc_frame(_fw.wrap(frame_type=ft, payload=_mp(data)))


# ── enqueue ────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_enqueue_job_creates_pending_row(db_session: AsyncSession) -> None:
    """enqueue_job must insert a row with status=pending."""
    repo = await factory_create_repo(db_session, slug="job-enqueue-test")
    await db_session.commit()

    job_id = await enqueue_job(db_session, repo.repo_id, "symbol_index", {"head": "abc123"})
    await db_session.commit()

    row = await db_session.get(db.MusehubBackgroundJob, job_id)
    assert row is not None
    assert row.status == "pending"
    assert row.job_type == "symbol_index"
    assert row.payload == {"head": "abc123"}
    assert row.attempt == 0


# ── claim ──────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_claim_next_job_returns_oldest_pending(db_session: AsyncSession) -> None:
    """claim_next_job must return the oldest pending row and mark it running."""
    repo = await factory_create_repo(db_session, slug="job-claim-test")
    await db_session.commit()

    id1 = await enqueue_job(db_session, repo.repo_id, "gc", {})
    id2 = await enqueue_job(db_session, repo.repo_id, "symbol_index", {"head": "xyz"})
    await db_session.commit()

    job = await claim_next_job(db_session)
    await db_session.commit()

    assert job is not None
    assert job.job_id == id1  # oldest first
    assert job.status == "running"
    assert job.claimed_at is not None
    assert job.attempt == 1


@pytest.mark.asyncio
async def test_claim_next_job_returns_none_when_queue_empty(db_session: AsyncSession) -> None:
    """claim_next_job must return None when no pending jobs exist."""
    job = await claim_next_job(db_session)
    assert job is None


# ── complete ───────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_complete_job_sets_done_status(db_session: AsyncSession) -> None:
    """complete_job must set status=done and done_at."""
    repo = await factory_create_repo(db_session, slug="job-complete-test")
    await db_session.commit()

    job_id = await enqueue_job(db_session, repo.repo_id, "gc", {})
    await db_session.commit()

    job = await claim_next_job(db_session)
    await db_session.commit()
    assert job is not None

    await complete_job(db_session, job_id)
    await db_session.commit()

    await db_session.refresh(job)
    assert job.status == "done"
    assert job.done_at is not None


# ── fail / retry ───────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_fail_job_retries_on_first_failure(db_session: AsyncSession) -> None:
    """fail_job must reset status=pending when attempts remain."""
    repo = await factory_create_repo(db_session, slug="job-fail-retry-test")
    await db_session.commit()

    job_id = await enqueue_job(db_session, repo.repo_id, "symbol_index", {"head": "aaa"})
    await db_session.commit()

    job = await claim_next_job(db_session)
    await db_session.commit()
    assert job is not None
    assert job.attempt == 1  # first attempt

    await fail_job(db_session, job_id, "transient error")
    await db_session.commit()

    await db_session.refresh(job)
    assert job.status == "pending"  # retryable
    assert job.error is not None


@pytest.mark.asyncio
async def test_fail_job_permanently_fails_after_max_attempts(db_session: AsyncSession) -> None:
    """fail_job must set status=failed after MAX_ATTEMPTS failures."""
    from musehub.services.musehub_jobs import _MAX_ATTEMPTS

    repo = await factory_create_repo(db_session, slug="job-fail-perm-test")
    await db_session.commit()

    job_id = await enqueue_job(db_session, repo.repo_id, "gc", {})
    await db_session.commit()

    # Exhaust all attempts.
    for _ in range(_MAX_ATTEMPTS):
        job = await claim_next_job(db_session)
        await db_session.commit()
        assert job is not None
        await fail_job(db_session, job_id, "boom")
        await db_session.commit()

    row = await db_session.get(db.MusehubBackgroundJob, job_id)
    assert row is not None
    assert row.status == "failed"
    assert row.done_at is not None


# ── push endpoint integration ──────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_push_enqueues_intel_jobs(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """A successful push must call enqueue_push_intel after a successful push.

    The autouse _stub_push_background_tasks fixture replaces enqueue_push_intel
    with a spy that records calls.  This test verifies the push route delegates
    to enqueue_push_intel — confirming it does NOT use asyncio.create_task.

    Uses the MWP streaming push format: POST /{owner}/{slug}/push/stream with
    concatenated msgpack frames (H → C → E).
    """
    import musehub.services.musehub_jobs as _jobs
    from musehub.models.wire import (
        SFRAME_COMMIT_PACK,
        SFRAME_END,
        SFRAME_HEADER,
        SFRAME_RESULT,
    )
    from muse.core.mpack import GRPC_CONTENT_TYPE
    from musehub.core.genesis import compute_branch_id
    from musehub.db.musehub_models import MusehubBranch

    repo = await factory_create_repo(
        db_session, slug="job-push-enqueue-test", owner="test-user-wire"
    )
    branch = MusehubBranch(
        branch_id=compute_branch_id(repo.repo_id, "main"),
        repo_id=repo.repo_id,
        name="main",
    )
    db_session.add(branch)
    await db_session.commit()

    commit_id = "sha256:" + hashlib.sha256(b"single-commit-stream").hexdigest()
    snap_id = "sha256:" + hashlib.sha256(b"single-snap-stream").hexdigest()

    commit = {
        "commit_id": commit_id,
        "parent_ids": [],
        "snapshot_id": snap_id,
        "branch": "main",
        "message": "test commit",
        "author": "test-user-wire",
        "committed_at": _utc_now().isoformat(),
        "signature": "",
        "signer_key_id": "",
        "agent_id": "",
        "model_id": "",
        "metadata": {},
    }
    snapshot = {
        "snapshot_id": snap_id,
        "manifest": {"file.py": snap_id},
        "committed_at": _utc_now().isoformat(),
    }

    body = (
        _wrap(SFRAME_HEADER, {"t": SFRAME_HEADER, "branch": "main", "force": False, "have": [],
              "head": commit_id, "n_objects": 0, "n_commits": 1})
        + _wrap(SFRAME_COMMIT_PACK, {"t": SFRAME_COMMIT_PACK, "commits": [commit], "snapshots": [snapshot]})
        + _wrap(SFRAME_END, {"t": SFRAME_END})
    )

    resp = await client.post(
        f"/{repo.owner}/{repo.slug}/push/stream",
        content=body,
        headers={**wire_headers, "Content-Type": GRPC_CONTENT_TYPE},
    )
    assert resp.status_code == 200

    # Verify enqueue_push_intel was called via the spy.
    assert _jobs._test_enqueued_calls, (
        "enqueue_push_intel was not called after a successful push"
    )