"""OnboardingService state machine — PR-1 (2026-06-01).

Covers:
  * ``start_process`` is idempotent (re-call same hire_date is a no-op
    on shape, updates audit trail)
  * ``start_process`` updates hire_date in-place + rescans
  * status transitions: scanning → preparing / ready / live based on
    blocker count + hire_date
  * blocker mutations: skip / mark_resolved recompute status
  * SLA evaluation: warn / urgent / breach thresholds past hire_date
  * cancel_process flips terminal
"""
from __future__ import annotations

import uuid
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.onboarding.models import (
    PROC_CANCELLED,
    PROC_LIVE,
    PROC_PREPARING,
    PROC_READY,
    RES_HARDWARE_ALLOCATION,
    RES_SOFTWARE_SEAT,
    RES_SSO_PROVISION,
    RES_WORK_EMAIL,
    RESO_PENDING,
    SLA_BREACH,
    SLA_URGENT,
    SLA_WARN,
    OnboardingBlocker,
)
from app.domains.onboarding.service import OnboardingService
from app.identity.person_models import IdentityPerson


def _today():
    return datetime.now(ZoneInfo("Asia/Shanghai")).date()


async def _mk_person(db: AsyncSession, *, hire_offset_days: int, work_email: str):
    pid = uuid.uuid4()
    p = IdentityPerson(
        id=pid,
        work_email=work_email,
        display_name=work_email.split("@")[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{pid.hex[:8]}",
        hire_date=_today() + timedelta(days=hire_offset_days),
        job_title="Engineer",
        company_name="PingCAP",
    )
    db.add(p)
    await db.flush()
    return p


@pytest.mark.asyncio
async def test_start_process_creates_row_and_scans(db_session: AsyncSession):
    p = await _mk_person(db_session, hire_offset_days=5, work_email="a@x.com")
    proc = await OnboardingService(db_session).start_process(
        p.id, p.hire_date, trace_id="t1",
    )
    assert proc.person_id == p.id
    assert proc.hire_date == p.hire_date
    # Bare new hire: account is fine (active + email) but every
    # provisioned dimension is a pending blocker → status preparing.
    assert proc.status == PROC_PREPARING
    assert proc.scanned_at is not None
    # Hard blockers exist for SSO, hardware (severity=blocker).
    res_types = {
        b.resource_type
        for b in (
            (await db_session.execute(
                __import__("sqlalchemy").select(OnboardingBlocker).where(
                    OnboardingBlocker.process_id == proc.id,
                    OnboardingBlocker.resolution == RESO_PENDING,
                )
            )).scalars().all()
        )
    }
    assert RES_SSO_PROVISION in res_types
    assert RES_HARDWARE_ALLOCATION in res_types
    # work_email is the LIVE-mailbox gate now (2026-07-06): a bare hire
    # has only the reserved SF address and no login identity, so the
    # mailbox isn't live → a pending work_email blocker exists. (Before
    # the fix work_email auto-satisfied on any ``x@y`` string, greening a
    # dimension for a mailbox that was never minted.)
    assert RES_WORK_EMAIL in res_types


@pytest.mark.asyncio
async def test_start_process_idempotent_same_date(db_session: AsyncSession):
    p = await _mk_person(db_session, hire_offset_days=5, work_email="b@x.com")
    svc = OnboardingService(db_session)
    p1 = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    p2 = await svc.start_process(p.id, p.hire_date, trace_id="t2")
    assert p1.id == p2.id
    assert p1.hire_date == p2.hire_date


@pytest.mark.asyncio
async def test_start_process_updates_hire_date(db_session: AsyncSession):
    p = await _mk_person(db_session, hire_offset_days=5, work_email="c@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    new_date = p.hire_date + timedelta(days=3)
    proc2 = await svc.start_process(p.id, new_date, trace_id="t2")
    assert proc2.id == proc.id
    assert proc2.hire_date == new_date


@pytest.mark.asyncio
async def test_status_promotes_to_live_when_hire_passed_and_clear(
    db_session: AsyncSession,
):
    p = await _mk_person(db_session, hire_offset_days=-2, work_email="d@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    # Resolve every pending blocker manually so the state machine can
    # promote past preparing.
    blockers = (
        (await db_session.execute(
            __import__("sqlalchemy").select(OnboardingBlocker).where(
                OnboardingBlocker.process_id == proc.id,
            )
        )).scalars().all()
    )
    for b in blockers:
        if b.resolution == RESO_PENDING:
            await svc.mark_blocker_resolved(b.id, trace_id="t1")
    await db_session.refresh(proc)
    # hire_date is in the past, no pending blockers → live.
    assert proc.status == PROC_LIVE
    assert proc.live_at is not None


@pytest.mark.asyncio
async def test_status_ready_when_clear_but_hire_future(
    db_session: AsyncSession,
):
    p = await _mk_person(db_session, hire_offset_days=5, work_email="e@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    blockers = (
        (await db_session.execute(
            __import__("sqlalchemy").select(OnboardingBlocker).where(
                OnboardingBlocker.process_id == proc.id,
            )
        )).scalars().all()
    )
    for b in blockers:
        if b.resolution == RESO_PENDING:
            await svc.mark_blocker_resolved(b.id, trace_id="t1")
    await db_session.refresh(proc)
    assert proc.status == PROC_READY


@pytest.mark.asyncio
async def test_skip_blocker_marks_resolution_skipped(db_session: AsyncSession):
    p = await _mk_person(db_session, hire_offset_days=5, work_email="f@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    blockers = (
        (await db_session.execute(
            __import__("sqlalchemy").select(OnboardingBlocker).where(
                OnboardingBlocker.process_id == proc.id,
                OnboardingBlocker.resource_type == RES_SOFTWARE_SEAT,
            )
        )).scalars().all()
    )
    assert len(blockers) == 1
    b = blockers[0]
    res = await svc.skip_blocker(
        b.id, reason="contractor", trace_id="t1",
    )
    assert res.resolution == "skipped"
    assert res.resolution_ref == "contractor"


@pytest.mark.asyncio
async def test_cancel_process_terminal(db_session: AsyncSession):
    p = await _mk_person(db_session, hire_offset_days=5, work_email="g@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    cancelled = await svc.cancel_process(
        proc.id, trace_id="t1", reason="offer rescinded",
    )
    assert cancelled.status == PROC_CANCELLED
    assert cancelled.cancelled_at is not None


@pytest.mark.asyncio
async def test_account_action_timeline_surfaces_failure_reason(
    db_session: AsyncSession,
):
    """The per-hire timeline unions person-scoped account audits with the
    process transitions, newest-first, and passes the failure reason
    through verbatim — so an admin can see WHY a hire has no account."""
    from app.audit.service import AuditService

    p = await _mk_person(db_session, hire_offset_days=1, work_email="tl@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    # Simulate the Google failure the cockpit's red chip can't explain.
    await AuditService(db_session).log(
        trace_id="t1",
        actor_type="system",
        action="google_workspace.account_create_failed",
        status="failed",
        resource_type="identity_person",
        resource_id=str(p.id),
        detail={"reason": "Google 429 rate limited", "domain": "pingcap.com"},
    )
    await db_session.flush()

    events = await svc.get_account_action_timeline(
        p.id, process_id=proc.id
    )
    actions = [e["action"] for e in events]
    # Both scopes present: the person-scoped failure and the
    # process-scoped start.
    assert "google_workspace.account_create_failed" in actions
    assert "onboarding.process_started" in actions
    failure = next(
        e for e in events
        if e["action"] == "google_workspace.account_create_failed"
    )
    assert failure["status"] == "failed"
    assert failure["detail"]["reason"] == "Google 429 rate limited"

    # process_id-less call returns ONLY the person-scoped account events.
    person_only = await svc.get_account_action_timeline(p.id)
    assert "onboarding.process_started" not in [
        e["action"] for e in person_only
    ]
    assert "google_workspace.account_create_failed" in [
        e["action"] for e in person_only
    ]


@pytest.mark.asyncio
async def test_account_action_timeline_surfaces_distinct_sf_snapshots(
    db_session: AsyncSession,
):
    from app.domains.hr.models import HrSyncBatch, HrSyncStagedEvent

    p = await _mk_person(db_session, hire_offset_days=1, work_email="unknown@pingcap.com")
    p.hris_person_id = "P10020"
    p.source_system = "successfactors"
    p.hris_instance_id = uuid.uuid4()
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    batch = HrSyncBatch(
        triggered_by="system:scheduler:sf-eu-prod", trigger_source="scheduled",
        hris_instance_id=p.hris_instance_id,
        sync_mode="incremental", start_date=p.hire_date, end_date=p.hire_date,
        status="confirmed", events_staged=3,
    )
    db_session.add(batch)
    await db_session.flush()
    original = {
        "display_name": "Unknown", "email_name": None,
        "work_email": "otso.juntunen@gmail.com", "employee_no": "P10020",
        "hire_date": str(p.hire_date),
    }
    corrected = {**original, "display_name": "Otso Juntunen", "email_name": "otso.juntenen"}
    db_session.add_all([
        HrSyncStagedEvent(batch_id=batch.id, event_type="onboard", user_id="P10020", person_data=original, employment_data={}, action="create", status="confirmed"),
        HrSyncStagedEvent(batch_id=batch.id, event_type="onboard", user_id="P10020", person_data=original, employment_data={}, action="create", status="confirmed"),
        HrSyncStagedEvent(batch_id=batch.id, event_type="onboard", user_id="P10020", person_data=corrected, employment_data={}, action="skip", status="confirmed"),
    ])
    await db_session.flush()

    events = await svc.get_account_action_timeline(p.id, process_id=proc.id)
    sf_events = [e for e in events if e["action"] == "hris.sync_snapshot_received"]
    assert len(sf_events) == 2
    warning = next(e for e in sf_events if e["status"] == "warning")
    assert warning["detail"]["display_name"] == "Unknown"
    assert warning["detail"]["risk_flags"] == ["placeholder_display_name"]
    healthy = next(e for e in sf_events if e["status"] == "success")
    assert healthy["detail"]["email_name"] == "otso.juntenen"


@pytest.mark.asyncio
async def test_evaluate_sla_warn_today_with_blockers(db_session: AsyncSession):
    p = await _mk_person(db_session, hire_offset_days=0, work_email="h@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    # Hire date is today; pending blockers persist → 0h past →
    # SLA = warn.
    level = await svc.evaluate_sla(proc)
    assert level == SLA_WARN


@pytest.mark.asyncio
async def test_evaluate_sla_urgent_after_24h(db_session: AsyncSession):
    p = await _mk_person(db_session, hire_offset_days=-1, work_email="hu@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    # 24h past → urgent.
    level = await svc.evaluate_sla(proc)
    assert level == SLA_URGENT


@pytest.mark.asyncio
async def test_evaluate_sla_breach_after_72h(db_session: AsyncSession):
    p = await _mk_person(db_session, hire_offset_days=-5, work_email="i@x.com")
    svc = OnboardingService(db_session)
    proc = await svc.start_process(p.id, p.hire_date, trace_id="t1")
    level = await svc.evaluate_sla(proc)
    # 5 days past hire_date with pending blockers → breach.
    assert level == SLA_BREACH