"""Admin HTTP API for the Identity Sync Center.

Endpoints (all require global admin):

    GET    /api/v1/identity-sync/presets           preset templates
    GET    /api/v1/identity-sync/targets           list targets
    POST   /api/v1/identity-sync/targets           create
    GET    /api/v1/identity-sync/targets/{id}      get one
    PATCH  /api/v1/identity-sync/targets/{id}      update
    DELETE /api/v1/identity-sync/targets/{id}      soft delete
    POST   /api/v1/identity-sync/targets/{id}/run  trigger sweep
    POST   /api/v1/identity-sync/targets/{id}/sync-person/{person_id}
                                                   single-person push
    POST   /api/v1/identity-sync/targets/{id}/preview
                                                   render mapping for one person
    GET    /api/v1/identity-sync/state             list per-person state
    GET    /api/v1/identity-sync/runs              list run history

The preview endpoint is the admin-UI mapping debugger — render a
single person's payload against draft mapping rules without persisting
or pushing. Eliminates "save → run → check Google → fix" cycle.
"""
from __future__ import annotations

from uuid import UUID

from fastapi import APIRouter, Body, Depends, Query, Request
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import (
    require_admin,
    require_admin_read,
    require_person_filter_lookup,
)
from app.core.response import success_response
from app.domains.identity_sync.destructive_confirmation import DestructivePlanService
from app.domains.identity_sync.models import IdentitySyncDestructivePlan
from app.domains.identity_sync.presets import list_presets
from app.domains.identity_sync.schemas import (
    DestructivePlanConfirmRequest,
    DestructivePlanConfirmResponse,
    DestructivePlanResponse,
    PresetResponse,
    PreviewRequest,
    RunResponse,
    StateResponse,
    TargetCreateRequest,
    TargetResponse,
    TargetUpdateRequest,
    TriggerResponse,
    TriggerRunRequest,
)
from app.domains.identity_sync.service import IdentitySyncService

router = APIRouter(
    prefix="/api/v1/identity-sync",
    tags=["identity-sync"],
)


def _serialize_target(row, recipient_map: dict[str, dict] | None = None) -> dict:
    data = TargetResponse.model_validate(row).model_dump(mode="json")
    ids = row.alert_recipient_ids or []
    data["alert_recipients"] = (
        [recipient_map[i] for i in ids if recipient_map and i in recipient_map]
        if recipient_map
        else []
    )
    return data


async def _resolve_recipient_map(db: AsyncSession, rows) -> dict[str, dict]:
    """Batch-resolve every target's alert recipients into id→brief dicts.

    One query for the whole page so the list view never N+1s. Persons
    that no longer exist are simply omitted (the UI then shows fewer
    chips than ids stored — surfacing the stale entry implicitly).
    """
    from sqlalchemy import select

    from app.identity.person_models import IdentityPerson

    ids: set[str] = set()
    for r in rows:
        for rid in (r.alert_recipient_ids or []):
            ids.add(str(rid))
    if not ids:
        return {}
    parsed: list[UUID] = []
    for rid in ids:
        try:
            parsed.append(UUID(rid))
        except ValueError:
            continue
    if not parsed:
        return {}
    persons = (
        await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(parsed))
        )
    ).scalars().all()
    return {
        str(p.id): {
            "id": str(p.id),
            "name": p.display_name or p.work_email,
            "email": p.work_email,
        }
        for p in persons
    }


@router.get("/presets", summary="List target presets")
async def list_target_presets(
    request: Request,
    _admin: AuthUser = Depends(require_admin_read()),
):
    items = [
        PresetResponse(
            key=p.key,
            label=p.label,
            description=p.description,
            kind=p.kind,
            mapping_rules=p.mapping_rules,
            filter_config=p.filter_config,
            schedule_config=p.schedule_config,
            runner_options=p.runner_options,
            requires_integration_kind=p.requires_integration_kind,
            requires_integration_provider=p.requires_integration_provider,
            write_mode=p.write_mode,
        ).model_dump()
        for p in list_presets()
    ]
    return success_response(data=items, request=request)


@router.get(
    "/mapping-metadata",
    summary="Authoritative source-field + transform catalog for the mapping builder",
)
async def get_mapping_metadata(
    request: Request,
    _admin: AuthUser = Depends(require_admin_read()),
):
    from app.domains.identity_sync.mapping import mapping_metadata

    return success_response(data=mapping_metadata(), request=request)


@router.get(
    "/overview",
    summary="Cockpit aggregation — upstream posture + one card per target",
)
async def get_overview(
    request: Request,
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Single round-trip behind the 数据同步 总览 tab: people / HRIS /
    data-quality posture plus per-target health cards (state counts,
    in-scope count, recent-run sparkline). Pure reads — zero
    downstream API calls.
    """
    svc = IdentitySyncService(db)
    return success_response(data=await svc.overview(), request=request)


_METRICS_WINDOWS = {"today": 1, "7d": 7, "30d": 30, "90d": 90}
_METRICS_TZ = "Asia/Shanghai"


@router.get(
    "/metrics",
    summary="Time-windowed sync analytics (监控看板) — KPIs, trend, failures",
)
async def get_metrics(
    request: Request,
    window: str = Query("7d", description="today | 7d | 30d | 90d"),
    target_id: UUID | None = Query(None, description="scope to one target"),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Aggregations behind the 监控看板 tab over ``[from, to)``: run KPI
    strip (success rate, person outcomes, avg/p95 duration), daily
    status/volume trend, per-target breakdown, trigger mix, the
    upstream→downstream data-volume overview, and recent failures. All
    pure reads — no downstream API calls.

    The window is anchored to ``Asia/Shanghai`` day boundaries so
    "今日 / 近 7 天" line up with the operating-timezone wall clock.
    """
    from datetime import datetime as _dt
    from datetime import time as _time
    from datetime import timedelta as _td
    from zoneinfo import ZoneInfo

    days = _METRICS_WINDOWS.get(window)
    if days is None:
        window = "7d"
        days = 7
    tz = ZoneInfo(_METRICS_TZ)
    now_local = _dt.now(tz)
    to_dt = now_local
    start_day = (now_local - _td(days=days - 1)).date()
    from_dt = _dt.combine(start_day, _time.min, tzinfo=tz)

    svc = IdentitySyncService(db)
    data = await svc.metrics(
        from_dt=from_dt,
        to_dt=to_dt,
        tz_name=_METRICS_TZ,
        target_id=target_id,
        window_label=window,
    )
    return success_response(data=data, request=request)


@router.get("/targets", summary="List sync targets")
async def list_targets(
    request: Request,
    kind: str | None = Query(default=None),
    enabled: bool | None = Query(default=None),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = IdentitySyncService(db)
    rows = await svc.list_targets(kind=kind, enabled=enabled)
    recipient_map = await _resolve_recipient_map(db, rows)
    return success_response(
        data=[_serialize_target(r, recipient_map) for r in rows], request=request,
    )


@router.get("/targets/{target_id}", summary="Get target by id")
async def get_target(
    target_id: UUID,
    request: Request,
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = IdentitySyncService(db)
    row = await svc.get_target(target_id)
    recipient_map = await _resolve_recipient_map(db, [row])
    return success_response(
        data=_serialize_target(row, recipient_map), request=request,
    )


@router.post("/targets", summary="Create a sync target")
async def create_target(
    body: TargetCreateRequest,
    request: Request,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = IdentitySyncService(db)
    row = await svc.create_target(
        name=body.name,
        slug=body.slug,
        description=body.description,
        kind=body.kind,
        integration_id=body.integration_id,
        enabled=body.enabled,
        on_offboard=body.on_offboard,
        write_mode=body.write_mode,
        mapping_config=body.mapping_config,
        filter_config=body.filter_config,
        schedule_config=body.schedule_config,
        runner_options=body.runner_options,
        alert_enabled=body.alert_enabled,
        alert_recipient_ids=body.alert_recipient_ids,
        actor_id=current_user.id,
    )
    await db.commit()
    recipient_map = await _resolve_recipient_map(db, [row])
    return success_response(
        data=_serialize_target(row, recipient_map), request=request,
    )


@router.patch("/targets/{target_id}", summary="Update a sync target")
async def update_target(
    target_id: UUID,
    body: TargetUpdateRequest,
    request: Request,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = IdentitySyncService(db)
    row = await svc.update_target(
        target_id,
        name=body.name,
        description=body.description,
        integration_id=body.integration_id,
        integration_id_set=body.integration_id_set,
        enabled=body.enabled,
        on_offboard=body.on_offboard,
        write_mode=body.write_mode,
        mapping_config=body.mapping_config,
        filter_config=body.filter_config,
        schedule_config=body.schedule_config,
        runner_options=body.runner_options,
        alert_enabled=body.alert_enabled,
        alert_recipient_ids=body.alert_recipient_ids,
        alert_recipient_ids_set="alert_recipient_ids" in body.model_fields_set,
        actor_id=current_user.id,
    )
    await db.commit()
    recipient_map = await _resolve_recipient_map(db, [row])
    return success_response(
        data=_serialize_target(row, recipient_map), request=request,
    )


@router.delete("/targets/{target_id}", summary="Soft-delete a sync target")
async def delete_target(
    target_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = IdentitySyncService(db)
    await svc.delete_target(target_id, actor_id=current_user.id)
    await db.commit()
    return success_response(data={"deleted": str(target_id)}, request=request)


@router.get(
    "/targets/{target_id}/dry-run",
    summary="Preview impact of a sync sweep without calling the runner",
)
async def dry_run(
    target_id: UUID,
    request: Request,
    sample_size: int = Query(
        default=20,
        ge=1,
        le=5000,
        description=(
            "Per-group cap on returned samples. The default of 20 is "
            "what the in-page Sync-Now confirm modal asks for (just "
            "enough to spot-check); the CSV-export path bumps this to "
            "5000 to capture the full impact set for offline review."
        ),
    ),
    full_reconcile: bool = Query(
        default=False,
        description=(
            "Preview the FULL reconcile (out-of-scope people fall to "
            "on_offboard). Default previews the push-only manual run, "
            "which never deactivates anyone."
        ),
    ),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Compute in-scope vs. out-of-scope split + on_offboard impact
    BEFORE calling the runner. Zero downstream API calls. Backs the
    "Sync now" confirmation dialog and the standalone preview button.
    """
    svc = IdentitySyncService(db)
    return success_response(
        data=await svc.dry_run(
            target_id, sample_size=sample_size, full_reconcile=full_reconcile,
        ),
        request=request,
    )


@router.get(
    "/targets/{target_id}/dry-run-feishu",
    summary="Full read-only sync plan for a feishu_contact target",
)
async def dry_run_feishu(
    target_id: UUID,
    request: Request,
    sample_size: int = Query(default=50, ge=1, le=2000),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Users (create/update/excluded) + departments (create/adopt/skip/
    soft-delete) plan, computed with live read-only Feishu reads. Backs
    the "同步前预览" panel. Zero writes. feishu_contact only.
    """
    svc = IdentitySyncService(db)
    return success_response(
        data=await svc.dry_run_feishu(target_id, sample_size=sample_size),
        request=request,
    )


@router.get(
    "/targets/{target_id}/dry-run-huilianyi",
    summary="Full read-only cutover plan for a huilianyi target",
)
async def dry_run_huilianyi(
    target_id: UUID,
    request: Request,
    sample_size: int = Query(default=50, ge=1, le=2000),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Users (create/update/email-conflict/resigned/unmapped-company/
    vendor-only) + departments (create/update/disable/untouched-legacy)
    reconciliation plan, computed against a live full-roster read of
    汇联易. Zero writes. huilianyi only — backs the 切换前对账 panel.
    """
    svc = IdentitySyncService(db)
    return success_response(
        data=await svc.dry_run_huilianyi(target_id, sample_size=sample_size),
        request=request,
    )


@router.post(
    "/targets/{target_id}/run",
    summary="Trigger a sync sweep now",
)
async def trigger_run(
    target_id: UUID,
    request: Request,
    body: TriggerRunRequest | None = Body(default=None),
    full_reconcile: bool | None = Query(
        default=None,
        description=(
            "Manual runs are push-only by default (create/update the "
            "in-scope set, never deactivate out-of-scope people). Set "
            "true to run the FULL reconcile — required for targets "
            "whose own trigger is manual/event_only, which otherwise "
            "have NO reconcile path and would retain leavers forever."
        ),
    ),
    force_reverify: bool | None = Query(
        default=None,
        description=(
            "Bypass the payload-hash skip AND the drift re-verify window "
            "(default 7 days): re-push EVERY in-scope person/dept now, "
            "re-asserting all Helix-managed fields against live vendor "
            "state. Use to force an immediate full correction instead of "
            "waiting for the next drift-reverify cycle."
        ),
    ),
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Kick off a sync sweep in the background and return its run_id
    immediately.

    The sweep runs off-request (retained background task) under the same
    per-target distributed lock the dispatcher uses — so a manual trigger
    still can't run concurrently with a scheduled tick — but the admin's
    connection lifetime no longer bounds it. A NetSuite reconcile that
    takes minutes previously died whenever the browser tab closed or an
    nginx proxy timeout elapsed (client-disconnect → request-task cancel →
    the run finalised ``cancelled`` mid-sweep). The UI polls the returned
    run_id (``GET /runs/{run_id}``) for progress and the final status.
    """
    svc = IdentitySyncService(db)
    effective_full_reconcile = body.full_reconcile if body is not None else bool(full_reconcile)
    effective_force_reverify = body.force_reverify if body is not None else bool(force_reverify)
    confirmation_token_hashes = [
        DestructivePlanService.hash_token(token)
        for token in (body.confirmation_tokens if body is not None else [])
    ]
    target = await svc.get_target(target_id)
    if not target.enabled:
        from app.core.exceptions import AppException
        raise AppException(
            code="IDENTITY_SYNC_DISABLED",
            message=f"target {target.slug!r} is disabled; enable it first",
            status_code=409,
        )
    run_id = await svc.start_manual_run_background(
        target,
        trigger_source=f"manual:{current_user.id}",
        push_only=False if effective_full_reconcile else None,
        force_reverify=effective_force_reverify,
        confirmation_token_hashes=confirmation_token_hashes,
        actor_user_id=current_user.id,
        trace_id=str(getattr(request.state, "trace_id", "identity-sync-manual")),
    )
    return success_response(
        data=TriggerResponse(
            target_id=target_id,
            run_id=run_id,
            status="running",
            summary=None,
        ).model_dump(mode="json"),
        request=request,
    )


@router.get(
    "/targets/{target_id}/destructive-plans",
    summary="List active mass-action plans awaiting review",
)
async def list_destructive_plans(
    target_id: UUID,
    request: Request,
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    await IdentitySyncService(db).get_target(target_id)
    rows = await DestructivePlanService(db).list_active(target_id)
    return success_response(
        data=[DestructivePlanResponse.model_validate(row).model_dump(mode="json") for row in rows],
        request=request,
    )


async def _confirm_destructive_plan(
    *, plan_id: UUID, payload: DestructivePlanConfirmRequest, request: Request,
    current_user: AuthUser, db: AsyncSession, reissue: bool,
):
    service = DestructivePlanService(db)
    token = await service.confirm(
        plan_id=plan_id,
        expected_fingerprint=payload.fingerprint,
        actor_user_id=current_user.id,
        trace_id=str(getattr(request.state, "trace_id", "identity-sync-confirm")),
        reissue=reissue,
    )
    plan = await db.get(IdentitySyncDestructivePlan, plan_id)
    if plan is None or plan.confirmation_expires_at is None:
        raise RuntimeError(
            "Confirmed identity-sync plan is missing its persisted expiry",
        )
    return success_response(
        data=DestructivePlanConfirmResponse(
            plan_id=plan_id,
            confirmation_token=token,
            confirmation_expires_at=plan.confirmation_expires_at,
        ).model_dump(mode="json"),
        request=request,
    )


@router.post("/destructive-plans/{plan_id}/confirm", summary="Confirm a reviewed plan")
async def confirm_destructive_plan(
    plan_id: UUID, payload: DestructivePlanConfirmRequest, request: Request,
    current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db),
):
    return await _confirm_destructive_plan(
        plan_id=plan_id, payload=payload, request=request,
        current_user=current_user, db=db, reissue=False,
    )


@router.post(
    "/destructive-plans/{plan_id}/reissue",
    summary="Invalidate a lost token and issue a replacement",
)
async def reissue_destructive_plan(
    plan_id: UUID, payload: DestructivePlanConfirmRequest, request: Request,
    current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db),
):
    return await _confirm_destructive_plan(
        plan_id=plan_id, payload=payload, request=request,
        current_user=current_user, db=db, reissue=True,
    )
@router.post(
    "/targets/{target_id}/sync-person/{person_id}",
    summary="Push a single person to a target now",
)
async def sync_one_person(
    target_id: UUID,
    person_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = IdentitySyncService(db)
    # ``force_in_scope=True`` — the admin explicitly named this
    # person; the filter is for batch sweeps, not for adjudicating
    # explicit pushes. Without this override, picking someone who
    # isn't in the filter would silently route to the deactivate
    # branch, which is confusing UX for "I just clicked Push".
    outcome = await svc.sync_person_to_target(
        target_id=target_id,
        person_id=person_id,
        trigger_source=f"manual:{current_user.id}",
        force_in_scope=True,
    )
    await db.commit()
    # ``detail`` carries the rendered request payload + vendor response
    # for the in-scope upsert path — the admin push modal renders the
    # full round-trip inline ("I see exactly what was sent and what
    # came back"), no second query needed.
    from app.domains.identity_sync.service import (
        _safe_error,
        mask_sensitive_display,
    )

    detail = outcome.detail or {}
    response_detail = {
        k: v for k, v in detail.items() if k != "request_payload"
    }
    return success_response(
        data={
            "person_id": str(person_id),
            "target_id": str(target_id),
            "status": outcome.status,
            "external_id": outcome.external_id,
            # Runner errors can embed vendor response bodies — scrub
            # credential shapes before echoing to the caller (storage
            # is already scrubbed; the HTTP echo must match).
            "error": _safe_error(outcome.error),
            "request_payload": mask_sensitive_display(
                detail.get("request_payload"),
            ),
            "response_detail": (
                mask_sensitive_display(response_detail) or None
            ),
        },
        request=request,
    )


@router.post(
    "/targets/{target_id}/preview",
    summary="Render mapping payload for one person without pushing",
)
async def preview_target_payload(
    target_id: UUID,
    body: PreviewRequest,
    request: Request,
    _admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.identity_sync.service import mask_sensitive_display

    svc = IdentitySyncService(db)
    # Fall back to the stored mapping_config when the body doesn't
    # carry one — the "Mapping Preview" drawer uses this read-only
    # mode to render the live saved config without re-typing it. The
    # editor still POSTs an explicit draft mapping_config.
    target = await svc.get_target(target_id)
    mapping_config = body.mapping_config or (target.mapping_config or {})
    runner_options = body.runner_options or (target.runner_options or {})
    result = await svc.preview_payload(
        person_id=body.person_id,
        mapping_config=mapping_config,
        runner_options=runner_options,
    )
    # Hash + payload for the UI to display. The preview embeds the raw
    # person ``context`` and ``rendered_payload`` (mobile_phone, etc.);
    # mask before echoing to the caller — same contract as the sibling
    # request_payload / snapshot / push-log endpoints in this router
    # (an admin loop must not be able to scrape plaintext PII here when
    # every other surface masks it). The hash is computed pre-mask, so
    # drift comparison in the UI is unaffected.
    return success_response(
        data=mask_sensitive_display(result), request=request,
    )


@router.get("/state", summary="List per-person sync state")
async def list_state(
    request: Request,
    target_id: UUID | None = Query(default=None),
    person_id: UUID | None = Query(default=None),
    status: str | None = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=200),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = IdentitySyncService(db)
    items, total = await svc.list_state(
        target_id=target_id,
        person_id=person_id,
        status=status,
        page=page,
        page_size=page_size,
    )
    serialized = [
        StateResponse.model_validate(it).model_dump(mode="json")
        for it in items
    ]
    return success_response(
        data=serialized,
        meta={
            "page": page,
            "page_size": page_size,
            "total": total,
        },
        request=request,
    )


@router.get(
    "/lookups/departments",
    summary="Department tree (source_id + name + path) for filter pickers",
)
async def lookup_departments(
    request: Request,
    _reader: AuthUser = Depends(require_person_filter_lookup()),
    db: AsyncSession = Depends(get_db),
):
    """Return every active department node with source_id + name +
    parent_source_id + org_path. Renders as a tree in the frontend
    filter editor — the admin picks names, we save source_ids.

    Cached client-side per page-load; ~500 rows is small enough that
    we don't paginate.
    """
    from sqlalchemy import select as sa_select

    from app.identity.person_models import IdentityOrgNode

    rows = (await db.execute(
        sa_select(IdentityOrgNode).where(
            IdentityOrgNode.org_node_type == "department",
            IdentityOrgNode.deleted_at.is_(None),
        ).order_by(IdentityOrgNode.org_path)
    )).scalars().all()
    return success_response(
        data=[
            {
                "source_id": r.org_node_source_id,
                "name": r.org_node_name,
                "parent_source_id": r.parent_org_node_source_id,
                "org_path": r.org_path,
                "level": r.org_level,
            }
            for r in rows
        ],
        request=request,
    )


@router.get(
    "/lookups/countries",
    summary="Distinct country codes from identity_persons (filter picker)",
)
async def lookup_countries(
    request: Request,
    _reader: AuthUser = Depends(require_person_filter_lookup()),
    db: AsyncSession = Depends(get_db),
):
    from sqlalchemy import distinct
    from sqlalchemy import select as sa_select

    from app.identity.person_models import IdentityPerson

    rows = (await db.execute(
        sa_select(distinct(IdentityPerson.country))
        .where(
            IdentityPerson.country.is_not(None),
            IdentityPerson.country != "",
            IdentityPerson.deleted_at.is_(None),
        )
        .order_by(IdentityPerson.country)
    )).scalars().all()
    return success_response(
        data=[c for c in rows if c],
        request=request,
    )


@router.get(
    "/lookups/entities",
    summary="Distinct legal entities (company_code + name) for filter picker",
)
async def lookup_entities(
    request: Request,
    _reader: AuthUser = Depends(require_person_filter_lookup()),
    db: AsyncSession = Depends(get_db),
):
    """Return every distinct ``(company_code, company_name)`` seen on
    active identity_persons. Feeds the "employing entity" multi-select
    in the filter editor — the admin picks entities, we save
    company_codes. Small set (~20 rows), no pagination.
    """
    from sqlalchemy import select as sa_select

    from app.identity.person_models import IdentityPerson

    rows = (await db.execute(
        sa_select(
            IdentityPerson.company_code,
            IdentityPerson.company_name,
        )
        .where(
            IdentityPerson.company_code.is_not(None),
            IdentityPerson.company_code != "",
            IdentityPerson.deleted_at.is_(None),
        )
        .distinct()
        .order_by(IdentityPerson.company_code)
    )).all()
    # Collapse to one row per code (a code may show minor name variants;
    # keep the first stable name).
    seen: dict[str, str] = {}
    for code, name in rows:
        if code and code not in seen:
            seen[code] = name or code
    return success_response(
        data=[{"company_code": c, "company_name": n} for c, n in seen.items()],
        request=request,
    )


@router.get(
    "/lookups/persons",
    summary="Search persons by email or name (filter picker)",
)
async def lookup_persons(
    request: Request,
    q: str = Query(default="", description="Substring match on email or name"),
    ids: str = Query(
        default="",
        description=(
            "Comma-separated person ids to resolve by exact id (for "
            "rehydrating labels of an already-saved person_ids filter). "
            "When given, ``q`` is ignored and only these rows return."
        ),
    ),
    limit: int = Query(default=20, ge=1, le=100),
    _reader: AuthUser = Depends(require_person_filter_lookup()),
    db: AsyncSession = Depends(get_db),
):
    from sqlalchemy import or_
    from sqlalchemy import select as sa_select

    from app.identity.person_models import IdentityPerson

    base = sa_select(IdentityPerson).where(IdentityPerson.deleted_at.is_(None))
    id_list = [s.strip() for s in ids.split(",") if s.strip()] if ids else []
    if id_list:
        # Exact-id rehydration: resolve up to 100 ids so the admin UI can
        # show "name (email)" for a saved person_ids filter instead of raw
        # UUIDs. Bounded by the same limit cap.
        base = base.where(IdentityPerson.id.in_(id_list[:100]))
        rows = (await db.execute(base)).scalars().all()
    else:
        if q:
            like = f"%{q.strip().lower()}%"
            base = base.where(
                or_(
                    IdentityPerson.work_email.ilike(like),
                    IdentityPerson.name_pinyin.ilike(like),
                    IdentityPerson.display_name.ilike(like),
                )
            )
        rows = (await db.execute(
            base.order_by(IdentityPerson.work_email).limit(limit),
        )).scalars().all()
    return success_response(
        data=[
            {
                "id": str(p.id),
                "email": p.work_email,
                "name": p.name_pinyin or p.display_name or p.work_email,
            }
            for p in rows
        ],
        request=request,
    )


@router.post(
    "/targets/{target_id}/vendor-snapshot/refresh",
    summary="Pull the downstream system's live roster into the snapshot",
)
async def refresh_vendor_snapshot(
    target_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Read-only on the vendor side; wipe-and-replace locally. Works on
    disabled targets by design — "look before you flip". Guarded by the
    same per-target lock the dispatcher uses so a snapshot pull can't
    interleave with a running sweep's rate-limit budget.
    """
    from app.core.distributed_lock import LockBusy, distributed_lock
    from app.core.exceptions import AppException

    svc = IdentitySyncService(db)
    target = await svc.get_target(target_id)
    lock_key = f"identity_sync:target:{target.slug}"
    try:
        async with distributed_lock(lock_key, ttl=600):
            result = await svc.refresh_vendor_snapshot(
                target_id=target_id, actor_id=current_user.id,
            )
    except LockBusy:
        raise AppException(
            code="IDENTITY_SYNC_TARGET_BUSY",
            message=(
                f"target {target.slug!r} is busy (sweep or snapshot in "
                "flight); retry when it finishes."
            ),
            status_code=409,
        ) from None
    return success_response(data=result, request=request)


@router.get(
    "/targets/{target_id}/vendor-records",
    summary="Page the stored vendor snapshot (对端数据)",
)
async def list_vendor_records(
    target_id: UUID,
    request: Request,
    record_type: str | None = Query(default=None),
    q: str | None = Query(default=None, max_length=128),
    matched: bool | None = Query(default=None),
    is_active: bool | None = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=200),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.identity_sync.service import mask_sensitive_display

    svc = IdentitySyncService(db)
    rows, total, fetched_at = await svc.list_vendor_records(
        target_id=target_id, record_type=record_type, q=q,
        matched=matched, is_active=is_active,
        page=page, page_size=page_size,
    )
    return success_response(
        data=[
            {
                "id": str(r.id),
                "record_type": r.record_type,
                "external_key": r.external_key,
                "display_name": r.display_name,
                "email": r.email,
                "vendor_status": r.vendor_status,
                "is_active": r.is_active,
                "matched_person_id": (
                    str(r.matched_person_id) if r.matched_person_id else None
                ),
                # Display-boundary masking — phones / id numbers in
                # the vendor row never reach the wire in clear text.
                "raw": mask_sensitive_display(r.raw),
                "fetched_at": r.fetched_at.isoformat() if r.fetched_at else None,
            }
            for r in rows
        ],
        meta={
            "page": page,
            "page_size": page_size,
            "total": total,
            "fetched_at": fetched_at.isoformat() if fetched_at else None,
        },
        request=request,
    )


@router.get(
    "/targets/{target_id}/drift",
    summary="Per-person diff: Helix render vs the stored vendor snapshot",
)
async def get_drift(
    target_id: UUID,
    request: Request,
    q: str | None = Query(default=None, max_length=128),
    only_drift: bool = Query(default=False),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=200),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """What a sweep WOULD send right now vs what the vendor holds —
    field-level verdicts (match / drift / not_comparable), plus
    vendor-missing and render-failed buckets. Zero vendor API calls;
    refresh the vendor snapshot first for a current comparison.
    """
    svc = IdentitySyncService(db)
    return success_response(
        data=await svc.compute_drift(
            target_id=target_id, q=q, only_drift=only_drift,
            page=page, page_size=page_size,
        ),
        request=request,
    )


def _serialize_run_item(r) -> dict:
    from app.domains.identity_sync.service import mask_sensitive_display

    return {
        "id": str(r.id),
        "run_id": str(r.run_id),
        "target_id": str(r.target_id),
        "item_type": r.item_type,
        "person_id": str(r.person_id) if r.person_id else None,
        "subject_display": r.subject_display,
        "external_key": r.external_key,
        "outcome": r.outcome,
        "request_payload": mask_sensitive_display(r.request_payload),
        "response_detail": mask_sensitive_display(r.response_detail),
        "error": r.error,
        "duration_ms": r.duration_ms,
        "created_at": r.created_at.isoformat() if r.created_at else None,
    }


@router.get(
    "/runs/{run_id}/items",
    summary="Per-run ledger — every operation this run performed",
)
async def list_run_items(
    run_id: UUID,
    request: Request,
    item_type: str | None = Query(default=None),
    outcome: str | None = Query(default=None),
    q: str | None = Query(default=None, max_length=128),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=200),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """One row per vendor-touching operation: subject, outcome, the
    rendered request payload as sent, and the vendor response detail /
    error. The per-run drill-down behind the Runs table's 明细 action.
    """
    svc = IdentitySyncService(db)
    rows, total = await svc.list_run_items(
        run_id=run_id, item_type=item_type, outcome=outcome, q=q,
        page=page, page_size=page_size,
    )
    return success_response(
        data=[_serialize_run_item(r) for r in rows],
        meta={"page": page, "page_size": page_size, "total": total},
        request=request,
    )


@router.get(
    "/items",
    summary="Cross-run ledger — per-person / per-target sync history",
)
async def list_items(
    request: Request,
    person_id: UUID | None = Query(default=None),
    target_id: UUID | None = Query(default=None),
    item_type: str | None = Query(default=None),
    outcome: str | None = Query(default=None),
    q: str | None = Query(default=None, max_length=128),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=200),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """The person-by-person cutover audit trail: "what has Helix ever
    sent for this person, to which targets, and what came back."
    Requires at least one of ``person_id`` / ``target_id`` so the
    endpoint can't be used as an unbounded table scan.
    """
    if person_id is None and target_id is None:
        from app.core.exceptions import AppException
        raise AppException(
            code="IDENTITY_SYNC_ITEMS_FILTER_REQUIRED",
            message="provide person_id and/or target_id",
            status_code=400,
        )
    svc = IdentitySyncService(db)
    rows, total = await svc.list_run_items(
        person_id=person_id, target_id=target_id,
        item_type=item_type, outcome=outcome, q=q,
        page=page, page_size=page_size,
    )
    return success_response(
        data=[_serialize_run_item(r) for r in rows],
        meta={"page": page, "page_size": page_size, "total": total},
        request=request,
    )


@router.get("/runs/{run_id}", summary="Get one sync run by id")
async def get_run(
    run_id: UUID,
    request: Request,
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Fetch a single run row — used by the UI to poll a background
    manual "Sync now" until it reaches a terminal status."""
    svc = IdentitySyncService(db)
    run = await svc.get_run(run_id)
    return success_response(
        data=RunResponse.model_validate(run).model_dump(mode="json"),
        request=request,
    )


@router.get("/runs", summary="List sync run history")
async def list_runs(
    request: Request,
    target_id: UUID | None = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=200),
    _admin: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = IdentitySyncService(db)
    rows, total = await svc.list_runs(
        target_id=target_id, page=page, page_size=page_size,
    )
    serialized = [
        RunResponse.model_validate(r).model_dump(mode="json")
        for r in rows
    ]
    return success_response(
        data=serialized,
        meta={
            "page": page,
            "page_size": page_size,
            "total": total,
        },
        request=request,
    )