"""年度报告 (Helix Wrapped) — the once-a-year year-end Feishu push.

Guardrails (this is the ONE deliberate employee-facing push exception, approved
explicitly; the standing rule is no new employee pushes):
  * once a year — idempotent via a ``PlatformSetting`` marker
    ``wrapped.year_end_push.<year>`` so a manual re-run never double-pushes;
  * opt-out — respects the ``appearance.wrapped_push_opt_out`` preference;
  * anti-storm — a per-run recipient cap (config ``max_recipients``), and a
    per-person try/except so one Feishu failure never aborts the batch.

Feishu-only (no email — the report is a fun in-app surface, not a record). The
card deep-links to ``/me/wrapped/<year>`` (resolved absolute by the notifier).
"""
from __future__ import annotations

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

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)

_TEMPLATE_KEY = "wrapped_year_end"
_DEFAULT_CAP = 5000  # belt-and-suspenders; org is ~500


def _marker_key(year: int) -> str:
    return f"wrapped.year_end_push.{year}"


async def run_year_end_push(
    db: AsyncSession, *, year: int, max_recipients: int = _DEFAULT_CAP,
) -> dict[str, Any]:
    from app.domains.feed.models import PREF_WRAPPED_PUSH_OPT_OUT, IdentityPersonPreference
    from app.domains.me_universe.wrapped import build_my_wrapped, is_wrapped_meaningful
    from app.identity.person_models import IdentityPerson
    from app.integrations.models import PlatformSetting
    from app.notifications import fanout
    from app.notifications.models import NotificationOutbox
    from app.notifications.service import NotificationEnvelope

    marker = await db.get(PlatformSetting, _marker_key(year))
    if marker is not None and marker.value:
        return {"skipped": "already_pushed", "year": year}

    # Per-person idempotency, independent of the run-level marker above.
    # ``dispatch`` has no cross-call dedup — it inserts a fresh outbox row
    # every time — so the "leave the marker unset on a partial-failure run
    # so a re-run retries" strategy below would otherwise re-push everyone
    # who already succeeded. A wrapped row for ``year`` is any outbox row
    # for this template created on/after Jan-1 of ``year`` (the push fires
    # Dec-31, so this cleanly separates this year's send from prior years').
    # Naive UTC bound: ``created_at`` is stored naive ([[feedback_mysql_datetime_naive_on_fresh_load]]).
    year_start = datetime(year, 1, 1)
    already_pushed = {
        str(pid)
        for (pid,) in (await db.execute(
            select(NotificationOutbox.recipient_person_id).where(
                NotificationOutbox.template_key == _TEMPLATE_KEY,
                NotificationOutbox.created_at >= year_start,
            )
        )).all()
        if pid is not None
    }

    # Opt-outs: filter the truthy ones in Python — a JSON-column ``== True``
    # comparison diverges between MySQL and TiDB ([[json_extract_is_true_divergence]]).
    opt_out_rows = (await db.execute(
        select(
            IdentityPersonPreference.person_id,
            IdentityPersonPreference.pref_value,
        ).where(IdentityPersonPreference.pref_key == PREF_WRAPPED_PUSH_OPT_OUT)
    )).all()
    # ``bool(val)`` (not ``is True``) fails SAFE: any truthy stored form
    # (True / 1 / "true") counts as opted out, so we never push to someone
    # who asked not to be pushed even if a legacy/odd value snuck in.
    opted_out = {str(pid) for pid, val in opt_out_rows if bool(val)}

    persons = (await db.execute(
        select(IdentityPerson).where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
            IdentityPerson.work_email.is_not(None),
        )
    )).scalars().all()

    # Collected here, sent by ``enqueue_then_drain`` AFTER the loop. Two
    # reasons, both of which bit this function as written (#2126):
    #
    # * every ``dispatch`` inserted an outbox row (FK → identity_persons)
    #   and made a Feishu call inside ONE never-committed transaction, so
    #   ~500 person-row locks were held across ~500 HTTP round-trips —
    #   the login brown-out of [[reference_outbox_fk_lock_login_outage]],
    #   on Dec 31;
    # * a crash at recipient 300 rolled back all 300 outbox rows while
    #   the cards had ALREADY been sent — and ``already_pushed`` (which
    #   reads those very rows) is what makes this a once-a-year push, so
    #   the re-run sent 300 people a second card.
    # Company headcount is the same for everyone — compute it once
    # instead of once per recipient (#2193-3).
    headcount = int((await db.execute(
        select(func.count()).select_from(IdentityPerson).where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
    )).scalar() or 0)
    envelopes: list[NotificationEnvelope] = []
    # ``queued`` counts envelopes collected (what ``max_recipients`` caps);
    # ``sent`` below is what the fan-out actually DELIVERED. They were one
    # variable, which made the cap check read as if it counted sends.
    queued = 0
    failed = 0
    skipped_opt_out = 0
    skipped_already = 0
    skipped_empty = 0
    cap_hit = False
    for person in persons:
        if str(person.id) in opted_out:
            skipped_opt_out += 1
            continue
        if str(person.id) in already_pushed:
            skipped_already += 1
            continue
        if queued >= max_recipients:
            cap_hit = True
            break
        # Meaningful-content guard: build the person's wrapped payload and skip
        # anyone whose year is empty/degenerate (no devices, quiz, badges,
        # tickets, or any platform activity). Without this a near-year-end hire
        # gets a card claiming "你留下了不少足迹" that deep-links to an all-zeros
        # /me/wrapped/<year>. Mirrors the cloud-efficiency digest's "don't send
        # when the payload isn't meaningful" guard (handlers.py). A build error
        # is treated like a dispatch failure (leave the marker unset → retry),
        # never as license to push a card we couldn't vouch for.
        try:
            wrapped = await build_my_wrapped(
                db, person, year, headcount=headcount,
            )
        except Exception:
            failed += 1
            logger.exception("wrapped.year_end_push: build for %s failed", person.id)
            continue
        if not is_wrapped_meaningful(wrapped):
            skipped_empty += 1
            continue
        # Title/body come from the registered template so the card speaks the
        # recipient's language; the envelope values below are only the
        # no-template fallback. ``context`` must carry every placeholder the
        # template interpolates — the bell re-renders from it later.
        envelopes.append(NotificationEnvelope(
            recipient=person,
            title=f"🎉 你的 {year} 年度报告来啦",
            body_text=(
                f"{year} 这一年你在 Helix 留下了不少足迹 —— 点开看看你的"
                "年度回顾吧(设备 / 安全答题 / 成就徽章…)。"
            ),
            channels=["feishu"],
            template_key=_TEMPLATE_KEY,
            context={"year": year},
            url=f"/me/wrapped/{year}",
        ))
        queued += 1

    # Durable two-phase fan-out on a lock-free session: every card is
    # committed as a PENDING outbox row BEFORE anything is sent, then
    # delivered one commit at a time. A crash now leaves the un-sent rows
    # for the retry worker instead of erasing the record of cards that
    # already went out.
    result = await fanout.enqueue_then_drain(
        db, envelopes, label="wrapped.year_end_push",
    )
    sent = result["delivered"]
    # An enqueue that never landed, or a send that failed, both mean this
    # recipient has no durable record — the run is not clean and the
    # marker must stay unset so the next tick retries just them.
    # ``failed`` covers phase-1 losses too now, so the open-coded
    # ``len(envelopes) - enqueued`` term this line used to carry would
    # double-count them.
    failed += result["failed"]

    # Mark done ONLY on a fully-clean run (every eligible recipient pushed,
    # no failures). If Feishu was down for the whole/partial run, or the
    # per-run cap truncated the batch, leave the marker UNSET so a re-run
    # retries — otherwise the once-a-year push could silently no-op with no
    # recovery path (review finding). The re-run is safe to re-drive the
    # whole roster because the ``already_pushed`` set above skips anyone who
    # already got a card this year, so only the un-pushed tail is retried.
    completed_clean = failed == 0 and not cap_hit
    if completed_clean:
        if marker is None:
            db.add(PlatformSetting(key=_marker_key(year), value=True))
        else:
            marker.value = True
        await db.flush()

    return {
        "year": year,
        "marked_done": completed_clean,
        "cap_hit": cap_hit,
        "sent": sent,
        "failed": failed,
        "skipped_opt_out": skipped_opt_out,
        "skipped_already": skipped_already,
        "skipped_empty": skipped_empty,
        "generated_at": datetime.now(UTC).isoformat(),
    }


__all__ = ["run_year_end_push"]