"""The sync brain — coordinates scrapers, repos, watermarks, and the SyncRun log.

Each public ``sync_*`` function follows the same shape:

    read watermark -> scrape -> upsert -> (advance watermark) -> log SyncRun

Two flavors of "incremental" live here, per Task 2.7:

* **Time-window incremental** (messages only). The message endpoint pages
  backward over a ``(start, end]`` timestamp window, so a real incremental query
  is possible. ``sync_wechat_messages`` reads ``last_timestamp`` for
  ``(message, wechat, "<wechat_id>:<friend_id>")``, queries
  ``[last_timestamp, now]`` via :func:`scrape_wechat_messages`, upserts each
  record, **best-effort downloads media** for media-bearing messages, and
  advances the watermark to ``max(timestamp)``. A re-run with no new messages
  therefore re-queries the empty window and fetches nothing — the watermark
  advanced past the previously-seen rows.
* **Full re-scrape + idempotent upsert** (every other entity). The list
  endpoints (friends / accounts / groups / voice_calls / departments) have no
  ``since`` param, so "incremental" here = re-scrape everything and rely on the
  repos' idempotent Core upsert (same PK twice -> one refreshed row). The
  watermark records ``last_synced_at`` only — no cursor advance, because there
  is no cursor to advance.

Failure policy: a scraper that raises is caught, recorded as a ``SyncRun`` row
with ``status='failed'`` and the error string, and a failed :class:`SyncResult`
is returned — the orchestrator does NOT re-raise, so a single broken entity
does not abort a multi-entity batch. The caller decides whether to surface /
retry. Media-download failures are best-effort: a media error is logged and the
sync continues (the message is still upserted; the watermark still advances).

Transactions: the orchestrator never commits. Every write is flushed into the
caller's ``AsyncSession``; the caller owns the commit / rollback. This keeps a
multi-entity batch atomic at the caller's discretion and lets a test roll a
failed advance back by simply not committing. Exception:
:func:`sync_call_recordings` commits its own ``SyncRun`` because
:func:`download_call_recordings` commits per batch internally, so the run
bookkeeping must land in the same transaction boundary.
"""

from __future__ import annotations

import dataclasses
import datetime as dt
import logging
from collections.abc import AsyncIterator, Callable
from typing import Any

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import Department, SyncRun, WechatAccount, WhatsappAccount
from app.scraper.call import scrape_phone_calls, scrape_sms
from app.scraper.media import (
    download_call_recordings,
    download_message_images,
    download_message_media,
    download_voice_clip_audio,
)
from app.scraper.voice_clip import scrape_voice_clips
from app.scraper.wechat import (
    scrape_departments,
    scrape_wechat_accounts,
    scrape_wechat_friends,
    scrape_wechat_groups,
    scrape_wechat_messages,
    scrape_wechat_voice_calls,
)
from app.scraper.whatsapp import (
    scrape_wa_voice_calls,
    scrape_whatsapp_accounts,
    scrape_whatsapp_friends,
    scrape_whatsapp_groups,
    scrape_whatsapp_messages,
)
from app.session.yunke_client import YunkeClient
from app.storage.object_store import ObjectStore
from app.storage.repos import (
    upsert_chat_group,
    upsert_customer,
    upsert_department,
    upsert_messages,
    upsert_phone_call,
    upsert_sms,
    upsert_voice_call,
    upsert_voice_clip,
    upsert_wechat_account,
    upsert_whatsapp_account,
)
from app.sync.watermarks import get_last_timestamp, set_watermark

logger = logging.getLogger(__name__)

__all__ = [
    "ENTITY_CALL_RECORDINGS",
    "ENTITY_MESSAGE_IMAGES",
    "ENTITY_VOICE_CLIP_AUDIO",
    "SyncResult",
    "sync_all_wechat",
    "sync_call_recordings",
    "sync_departments",
    "sync_message_images",
    "sync_phone_calls",
    "sync_sms",
    "sync_voice_clip_audio",
    "sync_voice_clips",
    "sync_wa_voice_calls",
    "sync_wechat_accounts",
    "sync_wechat_friends",
    "sync_wechat_groups",
    "sync_wechat_messages",
    "sync_wechat_voice_calls",
    "sync_whatsapp_accounts",
    "sync_whatsapp_friends",
    "sync_whatsapp_groups",
    "sync_whatsapp_messages",
]

# ---------------------------------------------------------------------------
# Stable SyncRun / SyncState keys (entity_type / channel labels)
# ---------------------------------------------------------------------------

# entity_type labels — kept as module constants so the SyncRun / SyncState keys
# are stable strings shared between the orchestrator and any dashboard / ops
# consumer (e.g. Task 6.4 monitoring).
ENTITY_DEPARTMENT = "department"
ENTITY_WECHAT_ACCOUNT = "wechat_account"
ENTITY_WECHAT_FRIEND = "wechat_friend"
ENTITY_WECHAT_GROUP = "wechat_group"
ENTITY_WECHAT_VOICE_CALL = "wechat_voice_call"
ENTITY_WA_VOICE_CALL = "wa_voice_call"
# M3 (WhatsApp accounts + friends). The WA account scraper wraps
# ``businessSalesList`` (WhatsappAccount table); the WA friend scraper wraps
# ``businessFriendList`` (Customer table, channel="whatsapp"). Both share the
# org's department tree with WeChat (sync_departments seeds it).
ENTITY_WHATSAPP_ACCOUNT = "whatsapp_account"
ENTITY_WHATSAPP_FRIEND = "whatsapp_friend"
# WhatsApp chat-groups share the ``ChatGroup`` table with wechat groups (the
# ``channel`` column distinguishes them). The scraper wraps
# ``businessFriendList`` with the group-type switch (UNVERIFIED param — see
# :func:`app.scraper.whatsapp.scrape_whatsapp_groups`); records land with
# ``channel='whatsapp'``.
ENTITY_WHATSAPP_GROUP = "whatsapp_group"
ENTITY_MESSAGE = "message"
# M4 (/pc/ CRM module) — phone calls + SMS share no channel column with the
# wechat-family entities; they track their own SyncRun / SyncState rows.
ENTITY_PHONE_CALL = "phone_call"
ENTITY_SMS = "sms"
# Recording-localization pass: not a scrape — reads already-mirrored
# ``phone_call`` rows and downloads their OSS ``record_file`` into the local
# object store (records a SyncRun with this entity_type; no SyncState row
# because the phone_call table itself is the cursor).
ENTITY_CALL_RECORDINGS = "call_recordings"
# 长语音 (voice_clip) — per-employee scrape of ``/wechatweb/call/list``. The
# sync walks every distinct user_id across both staff-account tables and stamps
# the channel locally (WechatAccount -> wechat, WhatsappAccount -> whatsapp).
ENTITY_VOICE_CLIP = "voice_clip"
# Voice-clip audio localization pass: not a scrape — reads already-mirrored
# ``voice_clip`` rows and downloads their OSS ``oss_file_name`` into the local
# object store (records a SyncRun with this entity_type; no SyncState row
# because the voice_clip table itself is the cursor). Mirrors call_recordings.
ENTITY_VOICE_CLIP_AUDIO = "voice_clip_audio"
# Message image / media localization pass: not a scrape — reads already-mirrored
# ``message`` rows whose ``file_url`` / ``file_th_url`` / ``file_hd_url`` point
# at OSS and have no ``media_hash`` yet, downloads each into the local object
# store (records a SyncRun with this entity_type; no SyncState row because the
# message table itself is the cursor). Mirrors call_recordings / voice_clip_audio.
ENTITY_MESSAGE_IMAGES = "message_images"

CHANNEL_WECHAT = "wechat"
CHANNEL_WHATSAPP = "whatsapp"


# ---------------------------------------------------------------------------
# SyncResult — small per-run outcome returned to the caller
# ---------------------------------------------------------------------------


@dataclasses.dataclass(frozen=True)
class SyncResult:
    """Outcome of one sync run (one entity / one conversation).

    Mirrors the essential fields of the persisted ``SyncRun`` row so callers can
    react to a run without re-querying the DB. ``status`` is one of:

    * ``"success"`` — the scrape + upsert completed (the SyncRun row is
      persisted with ``status='success'``).
    * ``"failed"`` — the scrape/upsert raised; ``error`` carries the
      exception text (the SyncRun row is persisted with ``status='failed'``).
    * ``"skipped"`` — a precondition was unmet (e.g. no resolvable root
      ``department_id`` for an endpoint that requires one); ``error`` carries
      the reason. NO SyncRun row is recorded (a skip is neither success nor
      failure), so a skipped entity does not advance the watermark.
    """

    entity: str
    status: str
    fetched: int
    upserted: int
    channel: str | None = None
    scope_id: str | None = None
    error: str | None = None


# ---------------------------------------------------------------------------
# Time helpers
# ---------------------------------------------------------------------------


def _now() -> dt.datetime:
    """Aware UTC now (never naive). Used for SyncRun timestamps."""
    return dt.datetime.now(dt.UTC)


def _now_ms() -> int:
    """Current time as ms-epoch (the message scraper's timestamp unit)."""
    return int(_now().timestamp() * 1000)


# ---------------------------------------------------------------------------
# SyncRun bookkeeping
# ---------------------------------------------------------------------------


async def _start_run(
    session: AsyncSession,
    entity_type: str,
    *,
    channel: str | None = None,
    scope_id: str | None = None,
) -> SyncRun:
    """Open a ``running`` SyncRun row and flush it (assigning run_id)."""
    run = SyncRun(
        entity_type=entity_type,
        channel=channel,
        scope_id=scope_id,
        status="running",
        started_at=_now(),
    )
    session.add(run)
    await session.flush()
    return run


async def _finish_run(
    session: AsyncSession,
    run: SyncRun,
    *,
    status: str,
    fetched: int,
    upserted: int,
    error: str | None = None,
) -> None:
    """Stamp a SyncRun row with final status / counts / finished_at and flush."""
    run.status = status
    run.records_fetched = fetched
    run.records_upserted = upserted
    run.error = error
    run.finished_at = _now()
    await session.flush()


# ---------------------------------------------------------------------------
# Messages — time-window incremental (the workhorse)
# ---------------------------------------------------------------------------


def _message_scope(wechat_id: str, friend_id: str) -> str:
    """Scope-id key for a 1:1 (or room) WeChat conversation."""
    return f"{wechat_id}:{friend_id}"


async def sync_wechat_messages(
    session: AsyncSession,
    client: YunkeClient,
    store: ObjectStore,
    wechat_id: str,
    friend_id: str,
    *,
    types: str = "",
) -> SyncResult:
    """Incremental sync of one WeChat conversation's messages.

    Reads ``last_timestamp`` for ``(message, wechat, "<wechat_id>:<friend_id>")``
    (default ``0`` on first sync), queries ``[last_timestamp, now]`` via the
    scraper, upserts every yielded message, best-effort downloads media for
    each, and advances the watermark to ``max(timestamp)``.

    Returns a :class:`SyncResult`. On a scraper failure, records a
    ``status='failed'`` SyncRun with the error and returns a failed result
    (does NOT re-raise).
    """
    scope_id = _message_scope(wechat_id, friend_id)
    last_ts = await get_last_timestamp(
        session,
        ENTITY_MESSAGE,
        channel=CHANNEL_WECHAT,
        scope_id=scope_id,
        default=0,
    )
    end = _now_ms()

    run = await _start_run(
        session,
        ENTITY_MESSAGE,
        channel=CHANNEL_WECHAT,
        scope_id=scope_id,
    )

    # --- Scrape (may raise) ---
    try:
        msgs: list[dict[str, Any]] = [
            msg
            async for msg in scrape_wechat_messages(
                client, wechat_id, friend_id, last_ts, end, types=types
            )
        ]
    except Exception as exc:
        logger.exception(
            "scrape_wechat_messages failed for %s", scope_id
        )
        error = repr(exc)
        await _finish_run(
            session, run, status="failed", fetched=0, upserted=0, error=error
        )
        return SyncResult(
            entity=ENTITY_MESSAGE,
            status="failed",
            fetched=0,
            upserted=0,
            channel=CHANNEL_WECHAT,
            scope_id=scope_id,
            error=error,
        )

    # --- Upsert (idempotent; same msg_id twice -> one refreshed row) ---
    upserted = await upsert_messages(session, msgs) if msgs else 0

    # --- Best-effort media download (never aborts the sync) ---
    for msg in msgs:
        try:
            await download_message_media(client, msg, store, session)
        except Exception:
            logger.warning(
                "media download failed for msg %s; sync continues",
                msg.get("msg_id"),
                exc_info=True,
            )

    # --- Advance watermark to max(timestamp) seen in this batch ---
    timestamps = [m.get("timestamp") for m in msgs if m.get("timestamp") is not None]
    if timestamps:
        await set_watermark(
            session,
            ENTITY_MESSAGE,
            channel=CHANNEL_WECHAT,
            scope_id=scope_id,
            last_timestamp=max(timestamps),
        )

    await _finish_run(
        session, run, status="success", fetched=len(msgs), upserted=upserted
    )
    return SyncResult(
        entity=ENTITY_MESSAGE,
        status="success",
        fetched=len(msgs),
        upserted=upserted,
        channel=CHANNEL_WECHAT,
        scope_id=scope_id,
    )


# ---------------------------------------------------------------------------
# Full re-scrape entities — idempotent upsert, no since-param
# ---------------------------------------------------------------------------


# Type aliases for readability (a scrape factory takes the client captured at
# call time and returns an async generator of records; an upsert takes the
# session + one record).
_ScrapeFactory = Callable[[], AsyncIterator[dict[str, Any]]]
_UpsertOne = Callable[[AsyncSession, dict[str, Any]], Any]


async def _sync_full_rescrape(
    session: AsyncSession,
    *,
    entity_type: str,
    channel: str | None,
    scope_id: str | None,
    scrape: _ScrapeFactory,
    upsert: _UpsertOne,
) -> SyncResult:
    """Shared body for the "re-scrape all + idempotent upsert" entities.

    These list endpoints have no ``since`` param, so "incremental" = re-scrape
    everything and rely on the repos' idempotent Core upsert (same PK twice ->
    one refreshed row). The watermark records ``last_synced_at`` only; no
    version/timestamp cursor is advanced because there is no upstream window to
    close. A scraper failure is recorded as a failed SyncRun and a failed
    SyncResult is returned (no re-raise).
    """
    run = await _start_run(
        session, entity_type, channel=channel, scope_id=scope_id
    )

    try:
        recs: list[dict[str, Any]] = [rec async for rec in scrape()]
    except Exception as exc:
        logger.exception("scrape failed for %s", entity_type)
        error = repr(exc)
        await _finish_run(
            session, run, status="failed", fetched=0, upserted=0, error=error
        )
        return SyncResult(
            entity=entity_type,
            status="failed",
            fetched=0,
            upserted=0,
            channel=channel,
            scope_id=scope_id,
            error=error,
        )

    for rec in recs:
        await upsert(session, rec)

    # Stamp the watermark's last_synced_at (no cursor advance — see docstring).
    await set_watermark(
        session, entity_type, channel=channel, scope_id=scope_id
    )

    await _finish_run(
        session, run, status="success", fetched=len(recs), upserted=len(recs)
    )
    return SyncResult(
        entity=entity_type,
        status="success",
        fetched=len(recs),
        upserted=len(recs),
        channel=channel,
        scope_id=scope_id,
    )


async def sync_departments(
    session: AsyncSession, client: YunkeClient
) -> SyncResult:
    """Full re-scrape of the department tree and idempotent upsert."""
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_DEPARTMENT,
        channel=None,
        scope_id=None,
        scrape=lambda: scrape_departments(client),
        upsert=upsert_department,
    )


async def _find_root_department_id(
    session: AsyncSession,
) -> str | None:
    """Resolve the company-root department id from the mirrored ``Department`` table.

    The ``wxUser/list`` form body REQUIRES ``departmentId`` (the company-root
    dept id) — empty fails with ``code -90000``. The root is a node whose
    ``parent_id`` is falsy (``None`` / ``""`` / ``"0"`` — the yunke tree
    convention). If several roots exist, the first by ``dept_id`` is returned
    (with a warning) so a multi-root org still scrapes. Returns ``None`` when
    no root is present so the caller can skip gracefully rather than send a
    known-bad empty filter.
    """
    stmt = (
        sa.select(Department)
        .where(
            sa.or_(
                Department.parent_id.is_(None),
                Department.parent_id == "",
                Department.parent_id == "0",
            )
        )
        .order_by(Department.dept_id.asc())
        .limit(1)
    )
    row = (await session.execute(stmt)).scalars().first()
    return row.dept_id if row is not None else None


async def sync_wechat_accounts(
    session: AsyncSession,
    client: YunkeClient,
    *,
    department_id: str | None = None,
) -> SyncResult:
    """Full re-scrape of all WeChat staff accounts (+ embedded staff rows).

    ``department_id`` is the company-root dept id required by the
    ``wxUser/list`` form body. When omitted, the orchestrator resolves it from
    the mirrored ``Department`` table (a root node whose ``parent_id`` is
    falsy) — which is why :func:`sync_departments` must run first. If no root
    can be resolved, the scrape is SKIPPED with a logged warning (returns a
    ``status='skipped'`` :class:`SyncResult` and records no ``SyncRun``) so a
    missing org tree does not abort a multi-entity batch.
    """
    resolved = department_id
    if resolved is None:
        resolved = await _find_root_department_id(session)
    if not resolved:
        logger.warning(
            "sync_wechat_accounts: no root department_id resolved"
            " (sync departments first?); skipping wxUser/list scrape"
        )
        return SyncResult(
            entity=ENTITY_WECHAT_ACCOUNT,
            status="skipped",
            fetched=0,
            upserted=0,
            error="no root department_id available",
        )
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WECHAT_ACCOUNT,
        channel=None,
        scope_id=None,
        scrape=lambda: scrape_wechat_accounts(client, department_id=resolved),
        upsert=upsert_wechat_account,
    )


async def sync_wechat_friends(
    session: AsyncSession, client: YunkeClient
) -> SyncResult:
    """Full re-scrape of all WeChat friends (customers) and idempotent upsert."""
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WECHAT_FRIEND,
        channel=CHANNEL_WECHAT,
        scope_id=None,
        scrape=lambda: scrape_wechat_friends(client),
        upsert=upsert_customer,
    )


async def sync_wechat_groups(
    session: AsyncSession, client: YunkeClient
) -> SyncResult:
    """Full re-scrape of all WeChat chat-groups (+members) and idempotent upsert."""
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WECHAT_GROUP,
        channel=CHANNEL_WECHAT,
        scope_id=None,
        scrape=lambda: scrape_wechat_groups(client),
        upsert=upsert_chat_group,
    )


async def sync_wechat_voice_calls(
    session: AsyncSession,
    client: YunkeClient,
    *,
    department_id: str | None = None,
) -> SyncResult:
    """Full re-scrape of all WeChat voice-call records and idempotent upsert.

    ``department_id`` is the company-root dept id REQUIRED by the
    ``queryWeChatVoiceList`` form body — without it (and the accompanying
    ``callSource`` / ``isRoom`` / ``startTime`` / ``endTime`` filters) the
    endpoint returns 200 with an EMPTY ``data.data`` array (0 rows; live
    diagnosis 2026-08-07). When omitted, the orchestrator resolves it from
    the mirrored ``Department`` table via :func:`_find_root_department_id`
    — which is why :func:`sync_departments` must run first. If no root can
    be resolved, the scrape is SKIPPED with a logged warning (returns a
    ``status='skipped'`` :class:`SyncResult` and records no ``SyncRun``) so
    a missing org tree does not abort a multi-entity batch.
    """
    resolved = department_id
    if resolved is None:
        resolved = await _find_root_department_id(session)
    if not resolved:
        logger.warning(
            "sync_wechat_voice_calls: no root department_id resolved"
            " (sync departments first?); skipping queryWeChatVoiceList scrape"
        )
        return SyncResult(
            entity=ENTITY_WECHAT_VOICE_CALL,
            status="skipped",
            fetched=0,
            upserted=0,
            error="no root department_id available",
        )
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WECHAT_VOICE_CALL,
        channel=None,
        scope_id=None,
        scrape=lambda: scrape_wechat_voice_calls(client, department_id=resolved),
        upsert=upsert_voice_call,
    )


async def sync_wa_voice_calls(
    session: AsyncSession,
    client: YunkeClient,
    *,
    department_id: str | None = None,
) -> SyncResult:
    """Full re-scrape of all WhatsApp voice-call records and idempotent upsert.

    Mirrors :func:`sync_wechat_voice_calls`: the WA variant of the unified
    ``queryWeChatVoiceList`` endpoint (``callSource=4``) has no ``since``
    param, so "incremental" = re-scrape everything and rely on
    :func:`upsert_voice_call`'s idempotent Core upsert (same ``call_id``
    twice -> one refreshed row). The ``channel="wa_voice_call"`` value the
    scraper stamps distinguishes WA rows from WeChat rows in the shared
    ``PhoneCall`` table; the SyncRun row records ``entity_type='wa_voice_call'``.

    ``department_id`` resolution mirrors :func:`sync_wechat_accounts`: the
    ``queryWeChatVoiceList`` form body REQUIRES the company-root
    ``departmentId`` (live diagnosis 2026-08-07 — without it the endpoint
    returns 200 + empty ``data.data``, 0 rows). When omitted, the
    orchestrator resolves it from the mirrored ``Department`` table; if no
    root can be resolved, the scrape is SKIPPED with a logged warning and a
    ``status='skipped'`` :class:`SyncResult`.
    """
    resolved = department_id
    if resolved is None:
        resolved = await _find_root_department_id(session)
    if not resolved:
        logger.warning(
            "sync_wa_voice_calls: no root department_id resolved"
            " (sync departments first?); skipping queryWeChatVoiceList scrape"
        )
        return SyncResult(
            entity=ENTITY_WA_VOICE_CALL,
            status="skipped",
            fetched=0,
            upserted=0,
            error="no root department_id available",
        )
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WA_VOICE_CALL,
        channel=None,
        scope_id=None,
        scrape=lambda: scrape_wa_voice_calls(client, department_id=resolved),
        upsert=upsert_voice_call,
    )


async def sync_whatsapp_accounts(
    session: AsyncSession,
    client: YunkeClient,
) -> SyncResult:
    """Full re-scrape of all WhatsApp staff accounts (+ embedded staff rows).

    Mirrors :func:`sync_wechat_accounts` but the WA ``businessSalesList``
    endpoint needs NO ``department_id`` (its body is just the page slice), so
    there is no root-department resolution / skip path here. Idempotent upsert
    via :func:`upsert_whatsapp_account` (same ``uid`` twice -> one refreshed
    row; the companion :func:`upsert_staff_from_account` covers the embedded
    staff row automatically).
    """
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WHATSAPP_ACCOUNT,
        channel=CHANNEL_WHATSAPP,
        scope_id=None,
        scrape=lambda: scrape_whatsapp_accounts(client),
        upsert=upsert_whatsapp_account,
    )


async def sync_whatsapp_friends(
    session: AsyncSession,
    client: YunkeClient,
    *,
    department_id: str | None = None,
) -> SyncResult:
    """Full re-scrape of all WhatsApp friends (customers) and idempotent upsert.

    Mirrors :func:`sync_wechat_accounts`'s department-resolution path: the WA
    ``businessFriendList`` form body REQUIRES ``departmentId`` (the company-root
    dept id) or it returns empty (live payload carries ``departmentId`` +
    ``type``). When omitted, the orchestrator resolves it from the mirrored
    ``Department`` table via :func:`_find_root_department_id` — which is why
    :func:`sync_departments` must run first. If no root can be resolved, the
    scrape is SKIPPED with a logged warning (returns a ``status='skipped'``
    :class:`SyncResult` and records no ``SyncRun``) so a missing org tree does
    not abort a multi-entity batch.

    Records land in the shared :class:`Customer` table with
    ``channel='whatsapp'`` (stamped by :func:`_normalize_whatsapp_friend`),
    so :func:`upsert_customer` is reused directly — exactly the same pattern as
    :func:`sync_wechat_friends`.
    """
    resolved = department_id
    if resolved is None:
        resolved = await _find_root_department_id(session)
    if not resolved:
        logger.warning(
            "sync_whatsapp_friends: no root department_id resolved"
            " (sync departments first?); skipping businessFriendList scrape"
        )
        return SyncResult(
            entity=ENTITY_WHATSAPP_FRIEND,
            status="skipped",
            fetched=0,
            upserted=0,
            error="no root department_id available",
        )
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WHATSAPP_FRIEND,
        channel=CHANNEL_WHATSAPP,
        scope_id=None,
        scrape=lambda: scrape_whatsapp_friends(client, department_id=resolved),
        upsert=upsert_customer,
    )


async def sync_whatsapp_groups(
    session: AsyncSession,
    client: YunkeClient,
    *,
    department_id: str | None = None,
) -> SyncResult:
    """Full re-scrape of all WhatsApp chat-groups and idempotent upsert.

    Mirrors :func:`sync_wechat_groups` (the WA analogue) but with the WA
    department-resolution path borrowed from :func:`sync_whatsapp_friends`:
    the WA ``businessFriendList`` body REQUIRES ``departmentId`` (the
    company-root dept id) or it returns empty. When omitted, the orchestrator
    resolves it from the mirrored ``Department`` table via
    :func:`_find_root_department_id` — which is why :func:`sync_departments`
    must run first. If no root can be resolved, the scrape is SKIPPED with a
    logged warning (returns a ``status='skipped'`` :class:`SyncResult` and
    records no ``SyncRun``) so a missing org tree does not abort a multi-entity
    batch.

    Records land in the shared :class:`ChatGroup` table with
    ``channel='whatsapp'`` (stamped by :func:`_normalize_whatsapp_group`), so
    :func:`upsert_chat_group` is reused directly — exactly the same pattern as
    :func:`sync_wechat_groups`. The ``channel`` column distinguishes WA groups
    from WeChat groups.

    .. note::

       UNVERIFIED (2026-08-11, risk-control): the WA group ``type="1"`` body
       switch is inferred, not captured. See
       :func:`app.scraper.whatsapp.scrape_whatsapp_groups`. If a post-risk-
       control smoke returns 0 rows, the param name is the likely culprit.
    """
    resolved = department_id
    if resolved is None:
        resolved = await _find_root_department_id(session)
    if not resolved:
        logger.warning(
            "sync_whatsapp_groups: no root department_id resolved"
            " (sync departments first?); skipping businessFriendList scrape"
        )
        return SyncResult(
            entity=ENTITY_WHATSAPP_GROUP,
            status="skipped",
            fetched=0,
            upserted=0,
            error="no root department_id available",
        )
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_WHATSAPP_GROUP,
        channel=CHANNEL_WHATSAPP,
        scope_id=None,
        scrape=lambda: scrape_whatsapp_groups(client, department_id=resolved),
        upsert=upsert_chat_group,
    )


async def sync_whatsapp_messages(
    session: AsyncSession,
    client: YunkeClient,
    uid: str,
    friend_uid: str,
) -> SyncResult:
    """Incremental sync of one WhatsApp conversation's messages.

    Mirrors :func:`sync_wechat_messages` for the WA channel: reads
    ``last_timestamp`` for ``(message, whatsapp, "<uid>:<friend_uid>")``
    (default ``0`` on first sync), queries ``[last_timestamp, now]`` via
    :func:`scrape_whatsapp_messages`, upserts every yielded message into the
    shared ``message`` table (``channel='whatsapp'`` — stamped by the scraper's
    normalizer), and advances the watermark to ``max(timestamp)``. A re-run
    with no new messages re-queries the empty window and fetches nothing —
    the watermark advanced past the previously-seen rows.

    Per Phase 1 this entrypoint does NOT do best-effort media download (unlike
    :func:`sync_wechat_messages`); WA media localization is deferred to
    Phase 2. The message record itself is still upserted (so the conversation
    history is mirrored) — only the media fetch is skipped.

    Returns a :class:`SyncResult`. On a scraper failure, records a
    ``status='failed'`` SyncRun with the error and returns a failed result
    (does NOT re-raise), so a single broken conversation does NOT abort a
    multi-conversation batch.
    """
    scope_id = f"{uid}:{friend_uid}"
    last_ts = await get_last_timestamp(
        session,
        ENTITY_MESSAGE,
        channel=CHANNEL_WHATSAPP,
        scope_id=scope_id,
        default=0,
    )
    end = _now_ms()

    run = await _start_run(
        session,
        ENTITY_MESSAGE,
        channel=CHANNEL_WHATSAPP,
        scope_id=scope_id,
    )

    # --- Scrape (may raise) ---
    try:
        msgs: list[dict[str, Any]] = [
            msg
            async for msg in scrape_whatsapp_messages(
                client, uid, friend_uid, last_ts, end
            )
        ]
    except Exception as exc:
        logger.exception(
            "scrape_whatsapp_messages failed for %s", scope_id
        )
        error = repr(exc)
        await _finish_run(
            session, run, status="failed", fetched=0, upserted=0, error=error
        )
        return SyncResult(
            entity=ENTITY_MESSAGE,
            status="failed",
            fetched=0,
            upserted=0,
            channel=CHANNEL_WHATSAPP,
            scope_id=scope_id,
            error=error,
        )

    # --- Upsert (idempotent; same msg_id twice -> one refreshed row) ---
    upserted = await upsert_messages(session, msgs) if msgs else 0

    # NOTE: no media download for WA (Phase 2; records-only for now).

    # --- Advance watermark to max(timestamp) seen in this batch ---
    timestamps = [
        m.get("timestamp") for m in msgs if m.get("timestamp") is not None
    ]
    if timestamps:
        await set_watermark(
            session,
            ENTITY_MESSAGE,
            channel=CHANNEL_WHATSAPP,
            scope_id=scope_id,
            last_timestamp=max(timestamps),
        )

    await _finish_run(
        session, run, status="success", fetched=len(msgs), upserted=upserted
    )
    return SyncResult(
        entity=ENTITY_MESSAGE,
        status="success",
        fetched=len(msgs),
        upserted=upserted,
        channel=CHANNEL_WHATSAPP,
        scope_id=scope_id,
    )


async def sync_call_recordings(
    session: AsyncSession, client: YunkeClient, store: ObjectStore
) -> SyncResult:
    """Download localized copies of call recordings for already-mirrored calls.

    Not a scrape: reads local ``phone_call`` rows whose ``record_file`` (OSS
    URL) has no ``record_media_hash`` yet, downloads each via the public OSS
    URL, and links it. Records a ``SyncRun`` (``entity_type='call_recordings'``);
    ``fetched`` = calls needing download, ``upserted`` = calls localized.

    Unlike the other ``sync_*`` functions (which defer the commit to the
    caller), this one commits its own ``SyncRun`` row: the underlying
    :func:`download_call_recordings` commits per batch as it advances, so the
    run bookkeeping must land in the same transaction boundary to stay
    consistent with what was actually persisted. Because it commits internally,
    it must NOT be added to a batched runner (:func:`_run_all` /
    :func:`sync_all_wechat`) without reconsidering per-batch atomicity — a
    batched wrapper can no longer treat the whole entity set as one rollback
    unit once this func has already persisted its own rows.

    Failure policy mirrors every other ``sync_*``: a downloader exception is
    recorded as a ``status='failed'`` ``SyncRun`` and a failed
    :class:`SyncResult` is returned (no re-raise), so a broken recording pass
    does NOT abort a multi-entity batch.
    """
    run = await _start_run(session, ENTITY_CALL_RECORDINGS)
    try:
        attempted, localized = await download_call_recordings(
            client, store, session
        )
        await _finish_run(
            session, run, status="success", fetched=attempted, upserted=localized
        )
        await session.commit()
        return SyncResult(
            entity=ENTITY_CALL_RECORDINGS,
            status="success",
            fetched=attempted,
            upserted=localized,
        )
    except Exception as exc:
        logger.exception("sync_call_recordings failed")
        await _finish_run(
            session, run, status="failed", fetched=0, upserted=0, error=repr(exc)
        )
        await session.commit()
        return SyncResult(
            entity=ENTITY_CALL_RECORDINGS,
            status="failed",
            fetched=0,
            upserted=0,
            error=repr(exc),
        )


async def sync_phone_calls(
    session: AsyncSession, client: YunkeClient
) -> SyncResult:
    """Full re-scrape of all phone-call records (``/pc/callLog/list``) and
    idempotent upsert into :class:`PhoneCall` (``channel="phone_call"``).

    Mirrors :func:`sync_wechat_voice_calls`: the callLog endpoint has no
    ``since`` param, so "incremental" = re-scrape everything and rely on the
    repo's idempotent Core upsert (same ``call_id`` twice -> one refreshed
    row). Records a SyncRun row (``entity_type='phone_call'``) and stamps the
    watermark's ``last_synced_at`` (no cursor advance — there is no upstream
    window to close). A scraper failure is recorded as a failed SyncRun and a
    failed SyncResult is returned (no re-raise), so a phone-call scrape that
    500s does NOT abort a larger batch.
    """
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_PHONE_CALL,
        channel=None,
        scope_id=None,
        scrape=lambda: scrape_phone_calls(client),
        upsert=upsert_phone_call,
    )


async def sync_sms(
    session: AsyncSession, client: YunkeClient
) -> SyncResult:
    """Full re-scrape of all SMS records (``/pc/customer/smsLogsList``) and
    idempotent upsert into :class:`Sms`.

    Mirrors :func:`sync_phone_calls`: the smsLogsList endpoint has no
    ``since`` param, so "incremental" = re-scrape everything and rely on the
    repo's idempotent Core upsert (same ``sms_id`` twice -> one refreshed
    row). Records a SyncRun row (``entity_type='sms'``) and stamps the
    watermark's ``last_synced_at``. A scraper failure is recorded as a failed
    SyncRun + SyncResult without re-raising.
    """
    return await _sync_full_rescrape(
        session,
        entity_type=ENTITY_SMS,
        channel=None,
        scope_id=None,
        scrape=lambda: scrape_sms(client),
        upsert=upsert_sms,
    )


# ---------------------------------------------------------------------------
# VoiceClip (长语音) — per-employee scrape across both staff-account tables
# ---------------------------------------------------------------------------


async def _collect_voice_clip_staff(
    session: AsyncSession,
) -> list[tuple[str, str]]:
    """Return ``(user_id, channel)`` pairs for voice-clip scraping.

    One pair per DISTINCT ``user_id`` across both staff-account tables. The
    ``call/list`` endpoint is scoped per employee (``userId`` request field)
    and is shared by WeChat + WhatsApp staff (distinct ``userId`` ranges);
    each employee is scraped exactly once. WeChat accounts take precedence on
    collision (deterministic — in practice the two ``userId`` ranges are
    disjoint, so collisions are not expected).

    The returned ``channel`` is the LOCAL messaging-channel label stamped on
    every scraped record: ``wechat`` for ids found in :class:`WechatAccount`,
    ``whatsapp`` for ids found only in :class:`WhatsappAccount`. The endpoint
    response does NOT carry the channel, so it must be resolved here from the
    owning account table.
    """
    pairs: list[tuple[str, str]] = []
    seen: set[str] = set()

    wechat_rows = (
        await session.execute(
            sa.select(WechatAccount.user_id)
            .where(WechatAccount.user_id.is_not(None))
            .where(WechatAccount.user_id != "")
            .distinct()
        )
    ).all()
    for (uid,) in wechat_rows:
        if uid and uid not in seen:
            seen.add(uid)
            pairs.append((uid, CHANNEL_WECHAT))

    wa_rows = (
        await session.execute(
            sa.select(WhatsappAccount.user_id)
            .where(WhatsappAccount.user_id.is_not(None))
            .where(WhatsappAccount.user_id != "")
            .distinct()
        )
    ).all()
    for (uid,) in wa_rows:
        if uid and uid not in seen:
            seen.add(uid)
            pairs.append((uid, CHANNEL_WHATSAPP))

    return pairs


async def sync_voice_clips(
    session: AsyncSession, client: YunkeClient
) -> SyncResult:
    """Full re-scrape of all 长语音 (voice_clip) records across all staff.

    Walks every distinct ``user_id`` in the local staff-account tables
    (:class:`WechatAccount` + :class:`WhatsappAccount`) and, per employee,
    scrapes ``POST /wechatweb/call/list`` (body ``{pageNum, pageSize,
    userId}``), upserting each record into :class:`VoiceClip`. The channel is
    stamped locally (``wechat`` / ``whatsapp``) based on which account table
    owns the ``user_id`` — the shared endpoint does not carry it.

    Mirrors :func:`sync_wechat_voice_calls`'s failure policy + SyncRun
    bookkeeping, but iterates MULTIPLE per-employee scrape streams (rather than
    one department-scoped stream). A scraper failure on one employee is caught,
    recorded as ``status='failed'`` for the whole entity, and a failed
    :class:`SyncResult` is returned (no re-raise) — the caller decides whether
    to surface / retry.

    When no staff ``user_id`` is present (e.g. accounts not yet synced), the
    scrape is SKIPPED with a logged warning and a ``status='skipped'``
    :class:`SyncResult` (no ``SyncRun`` recorded), so a fresh DB does not abort
    a larger batch.
    """
    pairs = await _collect_voice_clip_staff(session)
    if not pairs:
        logger.warning(
            "sync_voice_clips: no staff user_ids found in wechat_account /"
            " whatsapp_account (sync accounts first?); skipping call/list scrape"
        )
        return SyncResult(
            entity=ENTITY_VOICE_CLIP,
            status="skipped",
            fetched=0,
            upserted=0,
            error="no staff user_ids available",
        )

    run = await _start_run(session, ENTITY_VOICE_CLIP)

    fetched = 0
    upserted = 0
    try:
        for uid, channel in pairs:
            async for rec in scrape_voice_clips(client, uid, channel=channel):
                await upsert_voice_clip(session, rec)
                fetched += 1
                upserted += 1
    except Exception as exc:
        logger.exception("scrape_voice_clips failed")
        error = repr(exc)
        await _finish_run(
            session, run, status="failed",
            fetched=fetched, upserted=upserted, error=error,
        )
        return SyncResult(
            entity=ENTITY_VOICE_CLIP,
            status="failed",
            fetched=fetched,
            upserted=upserted,
            error=error,
        )

    await set_watermark(session, ENTITY_VOICE_CLIP)
    await _finish_run(
        session, run, status="success", fetched=fetched, upserted=upserted
    )
    return SyncResult(
        entity=ENTITY_VOICE_CLIP,
        status="success",
        fetched=fetched,
        upserted=upserted,
    )


async def sync_voice_clip_audio(
    session: AsyncSession, client: YunkeClient, store: ObjectStore
) -> SyncResult:
    """Download localized copies of voice_clip audio for already-mirrored clips.

    Not a scrape: reads local ``voice_clip`` rows whose ``oss_file_name`` (OSS
    URL) has no ``media_hash`` yet, downloads each via the public OSS URL, and
    links it. Records a ``SyncRun`` (``entity_type='voice_clip_audio'``);
    ``fetched`` = clips needing download, ``upserted`` = clips localized.

    Mirrors :func:`sync_call_recordings` exactly (same commit semantics):
    commits its own ``SyncRun`` row because :func:`download_voice_clip_audio`
    commits per batch internally, so the run bookkeeping must land in the same
    transaction boundary to stay consistent with what was actually persisted.

    Failure policy mirrors every other ``sync_*``: a downloader exception is
    recorded as a ``status='failed'`` ``SyncRun`` and a failed
    :class:`SyncResult` is returned (no re-raise), so a broken audio pass does
    NOT abort a multi-entity batch.
    """
    run = await _start_run(session, ENTITY_VOICE_CLIP_AUDIO)
    try:
        attempted, localized = await download_voice_clip_audio(
            client, store, session
        )
        await _finish_run(
            session, run, status="success", fetched=attempted, upserted=localized
        )
        await session.commit()
        return SyncResult(
            entity=ENTITY_VOICE_CLIP_AUDIO,
            status="success",
            fetched=attempted,
            upserted=localized,
        )
    except Exception as exc:
        logger.exception("sync_voice_clip_audio failed")
        await _finish_run(
            session, run, status="failed", fetched=0, upserted=0, error=repr(exc)
        )
        await session.commit()
        return SyncResult(
            entity=ENTITY_VOICE_CLIP_AUDIO,
            status="failed",
            fetched=0,
            upserted=0,
            error=repr(exc),
        )


async def sync_message_images(
    session: AsyncSession, client: YunkeClient, store: ObjectStore
) -> SyncResult:
    """Download localized copies of chat-message media for already-mirrored
    messages.

    Not a scrape: reads local ``message`` rows whose ``file_url`` /
    ``file_th_url`` / ``file_hd_url`` point at OSS, have ``type`` in
    ('image', 'video', 'file'), are on the wechat / whatsapp channel, and have
    no ``media_hash`` yet; downloads each and links it. Records a ``SyncRun``
    (``entity_type='message_images'``); ``fetched`` = messages needing
    download, ``upserted`` = messages localized.

    Mirrors :func:`sync_call_recordings` / :func:`sync_voice_clip_audio`
    exactly (same commit semantics): commits its own ``SyncRun`` row because
    :func:`download_message_images` commits per batch internally, so the run
    bookkeeping must land in the same transaction boundary to stay consistent
    with what was actually persisted.

    Failure policy mirrors every other ``sync_*``: a downloader exception is
    recorded as a ``status='failed'`` ``SyncRun`` and a failed
    :class:`SyncResult` is returned (no re-raise), so a broken image pass does
    NOT abort a multi-entity batch.
    """
    run = await _start_run(session, ENTITY_MESSAGE_IMAGES)
    try:
        attempted, localized = await download_message_images(
            client, store, session
        )
        await _finish_run(
            session, run, status="success", fetched=attempted, upserted=localized
        )
        await session.commit()
        return SyncResult(
            entity=ENTITY_MESSAGE_IMAGES,
            status="success",
            fetched=attempted,
            upserted=localized,
        )
    except Exception as exc:
        logger.exception("sync_message_images failed")
        await _finish_run(
            session, run, status="failed", fetched=0, upserted=0, error=repr(exc)
        )
        await session.commit()
        return SyncResult(
            entity=ENTITY_MESSAGE_IMAGES,
            status="failed",
            fetched=0,
            upserted=0,
            error=repr(exc),
        )


# ---------------------------------------------------------------------------
# Coordinator
# ---------------------------------------------------------------------------


async def sync_all_wechat(
    session: AsyncSession,
    client: YunkeClient,
    store: ObjectStore,
    *,
    conversations: list[tuple[str, str]] | None = None,
) -> list[SyncResult]:
    """Run the full WeChat sync sequence: org/entities first, then messages.

    Order: departments → accounts → friends → groups → voice_calls →
    wa_voice_calls, then the per-conversation message sync for every
    ``(wechat_id, friend_id)`` in ``conversations``. Each step returns its own
    :class:`SyncResult` so a single failed entity does NOT abort the batch
    (the failure is recorded and the next entity proceeds). Returns the
    results in execution order.

    ``conversations`` is the explicit "which direct / room chats to pull
    messages for" list. Full auto-discovery from the talkers endpoint is a
    LATER concern (Task 2.5 wires the scraper; wiring it into the coordinator
    is intentionally deferred to keep this entrypoint bounded and testable).
    When ``conversations`` is ``None`` / empty, only the six full-re-scrape
    entities run.
    """
    results: list[SyncResult] = [
        await sync_departments(session, client),
        await sync_wechat_accounts(session, client),
        await sync_wechat_friends(session, client),
        await sync_wechat_groups(session, client),
        await sync_wechat_voice_calls(session, client),
        await sync_wa_voice_calls(session, client),
    ]
    for wechat_id, friend_id in conversations or []:
        results.append(
            await sync_wechat_messages(
                session, client, store, wechat_id, friend_id
            )
        )
    return results