"""Incremental sync of Feishu helpdesk tickets into the mirror table.

Strategy (per enabled ``helpdesk``/``feishu`` integration):

  * **Watermark.** ``MAX(update_time_ms)`` already stored for the
    integration, minus a 5-minute overlap buffer (Feishu ``updated_at`` is
    not strictly monotonic across the page boundary). First run (no rows)
    pulls a bounded ``INITIAL_LOOKBACK_DAYS`` window so the table backfills
    without an unbounded all-time scan.
  * **Pull.** ``iter_tickets`` over the ``[watermark, now]`` window.
  * **Upsert.** ``INSERT … ON DUPLICATE KEY UPDATE`` keyed on
    ``ticket_id`` — idempotent, so a re-run over an overlapping window is a
    no-op on unchanged rows.
  * **Attribution.** ``guest_email`` is resolved to a Helix
    ``identity_persons`` id (case-insensitive ``work_email``) for the
    department dimension; unresolved guests keep ``guest_person_id`` NULL.
"""
from __future__ import annotations

import json
import logging
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import func, select
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.feishu_helpdesk.models import FeishuHelpdeskTicket
from app.identity.person_models import IdentityPerson
from app.integrations import service as integration_service
from app.integrations.drivers.feishu_helpdesk import FeishuHelpdeskDriver
from app.integrations.models import Integration

logger = logging.getLogger(__name__)

INITIAL_LOOKBACK_DAYS = 365
_OVERLAP_MS = 5 * 60 * 1000
# Cap raw payload size before persisting (defence against an unexpectedly
# huge ticket blob bloating the row).
_RAW_CAP = 16384


def _ms_to_dt(value: Any) -> datetime | None:
    if not isinstance(value, (int, float)) or value <= 0:
        return None
    return datetime.fromtimestamp(value / 1000, tz=UTC)


def _int_or_none(value: Any) -> int | None:
    """API uses -1 for not-applicable / unfinished — normalise to NULL."""
    if not isinstance(value, (int, float)):
        return None
    ivalue = int(value)
    return ivalue if ivalue >= 0 else None


def _float_or_none(value: Any) -> float | None:
    if not isinstance(value, (int, float)):
        return None
    fvalue = float(value)
    return fvalue if fvalue >= 0 else None


def _person_block(obj: Any) -> tuple[str | None, str | None, str | None]:
    """Extract (open_id/id, name, email) from a Feishu agent/guest block."""
    if not isinstance(obj, dict):
        return None, None, None
    oid = obj.get("id") or obj.get("open_id") or obj.get("user_id")
    name = obj.get("name")
    email = obj.get("email") or obj.get("enterprise_email")
    return (
        str(oid) if oid else None,
        str(name) if name else None,
        (str(email).strip().lower() if email else None),
    )


def _ticket_to_row(
    t: dict[str, Any], *, integration_id: str,
) -> dict[str, Any]:
    guest_oid, guest_name, guest_email = _person_block(t.get("guest"))
    owner_oid, owner_name, owner_email = _person_block(t.get("agent_owner"))
    update_ms = t.get("updated_at") if isinstance(t.get("updated_at"), int) else None
    # Cap the persisted raw blob — an unexpectedly huge ticket payload
    # (long custom-field bodies / attachment metadata) shouldn't bloat the
    # row. Keep the full dict when it serialises under the cap, else store a
    # truncation marker carrying the id so forensics still trace the ticket.
    raw: Any = t
    try:
        if len(json.dumps(t, ensure_ascii=False)) > _RAW_CAP:
            raw = {"_truncated": True, "ticket_id": t.get("ticket_id") or t.get("id")}
    except (TypeError, ValueError):
        raw = {"_unserializable": True, "ticket_id": t.get("ticket_id") or t.get("id")}
    return {
        "ticket_id": str(t.get("ticket_id") or t.get("id")),
        "integration_id": integration_id,
        "helpdesk_id": str(t.get("helpdesk_id") or ""),
        "helpdesk_name": t.get("helpdesk_name"),
        "status": _int_or_none(t.get("status")),
        "channel": _int_or_none(t.get("channel")),
        "ticket_type": _int_or_none(t.get("ticket_type")),
        "solve": _int_or_none(t.get("solve")),
        "score": _int_or_none(t.get("score")),
        "language": t.get("language"),
        "guest_open_id": guest_oid,
        "guest_name": guest_name,
        "guest_email": guest_email,
        "agent_owner_open_id": owner_oid,
        "agent_owner_name": owner_name,
        "agent_owner_email": owner_email,
        "agent_group": t.get("actual_agent_group"),
        "first_response_seconds": _int_or_none(t.get("agent_first_response_duration")),
        "resolution_seconds": _int_or_none(t.get("agent_resolution_time")),
        "agent_service_minutes": _float_or_none(t.get("agent_service_duration")),
        "bot_service_minutes": _float_or_none(t.get("bot_service_duration")),
        "processing_seconds": _int_or_none(t.get("actual_processing_time")),
        "created_at_feishu": _ms_to_dt(t.get("created_at")),
        "updated_at_feishu": _ms_to_dt(t.get("updated_at")),
        "closed_at_feishu": _ms_to_dt(t.get("closed_at")),
        "update_time_ms": update_ms,
        "raw": raw,
        "synced_at": datetime.now(UTC),
    }


async def _resolve_people(
    db: AsyncSession, emails: set[str],
) -> dict[str, Any]:
    """Map lowercased work-email → IdentityPerson.id."""
    emails = {e for e in emails if e}
    if not emails:
        return {}
    rows = (
        await db.execute(
            select(IdentityPerson.id, IdentityPerson.work_email).where(
                func.lower(IdentityPerson.work_email).in_(emails),
            )
        )
    ).all()
    return {(email or "").strip().lower(): pid for pid, email in rows}


async def _sync_one(
    db: AsyncSession, row: Integration, *, now_ms: int,
) -> dict[str, Any]:
    driver = await integration_service.build_driver_by_id(row.id, session=db)
    if not isinstance(driver, FeishuHelpdeskDriver):
        return {"integration": row.slug, "skipped": "not_a_helpdesk_driver"}
    client = driver.make_client()

    watermark = (
        await db.execute(
            select(func.max(FeishuHelpdeskTicket.update_time_ms)).where(
                FeishuHelpdeskTicket.integration_id == row.id,
            )
        )
    ).scalar()
    if watermark is None:
        start_ms = now_ms - INITIAL_LOOKBACK_DAYS * 86400 * 1000
        mode = "initial"
    else:
        start_ms = int(watermark) - _OVERLAP_MS
        mode = "incremental"

    tickets = await client.iter_tickets(
        update_time_start_ms=start_ms, update_time_end_ms=now_ms,
    )
    if not tickets:
        return {
            "integration": row.slug, "mode": mode,
            "fetched": 0, "upserted": 0,
        }

    parsed = [_ticket_to_row(t, integration_id=str(row.id)) for t in tickets]
    parsed = [p for p in parsed if p["ticket_id"] and p["ticket_id"] != "None"]

    people = await _resolve_people(
        db, {p["guest_email"] for p in parsed if p["guest_email"]},
    )
    for p in parsed:
        p["guest_person_id"] = people.get(p["guest_email"])

    # Columns the upsert overwrites on conflict (everything except the PK
    # and the create-only audit timestamp).
    update_cols = [
        c for c in parsed[0]
        if c not in ("ticket_id", "integration_id")
    ]
    stmt = mysql_insert(FeishuHelpdeskTicket).values(parsed)
    stmt = stmt.on_duplicate_key_update(
        {c: stmt.inserted[c] for c in update_cols},
    )
    await db.execute(stmt)
    await db.commit()

    return {
        "integration": row.slug,
        "helpdesk_id": client.helpdesk_id,
        "mode": mode,
        "fetched": len(tickets),
        "upserted": len(parsed),
    }


async def sync_all_helpdesks(db: AsyncSession) -> dict[str, Any]:
    """Sync every enabled ``helpdesk``/``feishu`` integration.

    Returns a summary dict suitable for ``last_result_summary``. One
    helpdesk failing does not abort the others — each is isolated and its
    error recorded in the summary, but the handler re-raises if EVERY
    helpdesk failed so the scheduled-job run lands ``status=failed``.
    """
    rows = (
        await db.execute(
            select(Integration).where(
                Integration.kind == "helpdesk",
                Integration.provider == "feishu",
                Integration.enabled.is_(True),
            )
        )
    ).scalars().all()

    now_ms = int(datetime.now(UTC).timestamp() * 1000)
    results: list[dict[str, Any]] = []
    errors = 0
    for row in rows:
        try:
            results.append(await _sync_one(db, row, now_ms=now_ms))
        except Exception as e:  # noqa: BLE001
            await db.rollback()
            errors += 1
            logger.exception(
                "feishu helpdesk sync failed for integration %s", row.slug,
            )
            results.append({"integration": row.slug, "error": f"{type(e).__name__}: {e}"})

    total_upserted = sum(r.get("upserted", 0) for r in results)
    summary = {
        "helpdesks": len(rows),
        "upserted": total_upserted,
        "errors": errors,
        "details": results,
    }
    if rows and errors == len(rows):
        # Every helpdesk failed — surface as a hard failure so the run is
        # recorded failed and the failure-notifier fires (handlers must
        # raise, not return status=failed — see feedback_handler_status_must_raise).
        raise RuntimeError(f"feishu helpdesk sync: all {errors} helpdesk(s) failed")
    return summary