"""催办节奏 (nag cadence) —— 同一件事对同一个人的第 N 次催办,不应该发生在第 N 天。

平台上「一直催到你按按钮为止」的推送有两个:资产盘点提醒和云订阅
Owner 半年确认。两个都是**每天一张卡、无限期**,而且各自都是深思熟虑之后
的选择 —— 云那条 2026-08-24 明确否掉了「14 天后转交 IT」的老做法,理由
是把九十来个别人的待办丢给 5 个管理员「produced a chase queue and moved
no decisions」。那个判断是对的。

但它当时比较的是「每天催」和「转交别人」,没有比较过第三种:**继续催,
但是越来越慢。** 生产数据说这第三种才是缺的那个 —— 最近 30 天资产盘点
提醒人均 **16.7 条**、云 Owner 确认人均 **7.6 条**。一个已经无视了 20 张
同样卡片的人,不会被第 21 张说动;那 21 张唯一确定做到的事,是训练他把
这个来源整个过滤掉 —— 连同将来某一张真正要紧的卡。

所以这里的阶梯是「不放弃、但递减」:头几天每天,然后 2 天、3 天、5 天,
之后每周一次,**永不停止**。开着的事一直开着(2026-08-24 的要求成立),
而第一个月的推送从 30 条降到 8 条,之后每月约 4 条。

三条设计约束,每一条都是防着一种把降噪做成事故的方式:

1. **第一次永远发。** 节流只作用于**重复**催办。一件事的第一次通知是
   信息,不是噪音,任何阶梯都不该延后它。
2. **只管周期性催办,不碰一次性事件。** 审批到了、卡片要签、账号建好了 ——
   这些是事件通知,发一次就完了,它们根本不进这套。把它们塞进来会变成
   「重要的东西被延后」,那比噪音坏得多。
3. **事情办完必须销账。** ``clear`` 不被调用的话,同一个人下次因为同一类
   事情被催时会继承上一次的计数,于是第一次就落进周更档 —— 一个「降噪」
   把新问题变成了几乎不通知。每个接入点都要在终态调用它。
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID

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

from app.notifications.models import NotificationNagState

logger = logging.getLogger(__name__)

# Gap required BEFORE the Nth reminder, in days, indexed by how many
# have already gone out. Sends land on day 0, 1, 2, 3, 5, 8, 13, then
# every ``TAIL_INTERVAL_DAYS``.
#
# The front of the ladder stays daily on purpose: most people answer in
# the first few days, and slowing those down would trade a real drop in
# response time for noise nobody was feeling yet. The saving comes
# entirely from the long tail, which is where the volume actually is.
DEFAULT_LADDER: tuple[int, ...] = (1, 1, 1, 2, 3, 5)
TAIL_INTERVAL_DAYS = 7

# A per-template override lives here rather than in each caller so the
# whole platform's nag policy is readable in one place.
LADDERS: dict[str, tuple[int, ...]] = {}
TAIL_OVERRIDES: dict[str, int] = {}

# Stands for "no per-template override" when asking the ladder a
# question about itself. Not a template key — spelling it as one
# invites a reader to go looking for the template.
_LADDER_DEFAULT_KEY = "\x00default"


def required_gap_days(template_key: str, sent_count: int) -> int:
    """Days that must pass after send #``sent_count`` before the next one."""
    ladder = LADDERS.get(template_key, DEFAULT_LADDER)
    if sent_count <= 0:
        return 0
    if sent_count <= len(ladder):
        return ladder[sent_count - 1]
    return TAIL_OVERRIDES.get(template_key, TAIL_INTERVAL_DAYS)


def projected_sends(template_key: str, days: int) -> int:
    """How many reminders the ladder produces over ``days`` — the number
    an admin actually wants to see next to a cadence setting, because
    "ladder (1,1,1,2,3,5) then weekly" means nothing to a reader and
    "8 in the first month" means everything.
    """
    elapsed = 0
    count = 0
    while True:
        gap = required_gap_days(template_key, count)
        if count and elapsed + gap > days:
            break
        elapsed += gap
        count += 1
        if count > 1000:  # pathological ladder guard
            break
    return count


async def _state(
    db: AsyncSession, *, recipient_id: UUID, template_key: str,
    subject_key: str,
) -> NotificationNagState | None:
    return (
        await db.execute(
            select(NotificationNagState).where(
                NotificationNagState.recipient_person_id == recipient_id,
                NotificationNagState.template_key == template_key,
                NotificationNagState.subject_key == subject_key,
            )
        )
    ).scalars().first()


async def taper_holds(
    db: AsyncSession,
    *,
    recipient_id: UUID,
    template_key: str,
    subject_key: str,
    now: datetime | None = None,
) -> bool:
    """``True`` when the TAPER (and only the taper) says "not yet".

    For callers that already enforce their own daily floor — a
    per-local-day dedup, a 20h minimum interval — this is the correct
    entry point rather than :func:`should_send`.

    The rule it encodes: **the ladder may only ever LENGTHEN an
    interval, never tighten one that already exists.** Layering a second
    wall-clock "24h minus slack" test on top of a caller's local-day
    boundary drops a reminder whenever the two disagree — yesterday's
    tick ran late, today's ran early — and a change made to reduce noise
    silently becomes a change to delivery. So while the ladder is still
    on its daily rungs this returns ``False`` and the caller's own floor
    governs unchanged; it only starts holding sends back once the ladder
    asks for more than a day.
    """
    st = await _state(
        db, recipient_id=recipient_id, template_key=template_key,
        subject_key=subject_key,
    )
    if st is None or st.sent_count <= 0 or st.last_sent_at is None:
        return False
    gap = required_gap_days(template_key, st.sent_count)
    if gap <= 1:
        return False
    ref = st.last_sent_at
    if ref.tzinfo is None:
        ref = ref.replace(tzinfo=UTC)
    deadline = ref + timedelta(days=gap) - timedelta(hours=2)
    return (now or datetime.now(UTC)) < deadline


async def should_send(
    db: AsyncSession,
    *,
    recipient_id: UUID,
    template_key: str,
    subject_key: str,
    now: datetime | None = None,
) -> bool:
    """Is this person due for another reminder about this thing?

    ``subject_key`` identifies the THING, not the message — an inventory
    cycle id, a confirmation-prompt id. Two different open asks nag
    independently, which is correct: tapering one must never mute the
    other.

    Returns ``True`` when nothing has been sent yet. See constraint (1)
    in the module docstring — a first notification is information.
    """
    st = await _state(
        db, recipient_id=recipient_id, template_key=template_key,
        subject_key=subject_key,
    )
    if st is None or st.sent_count <= 0 or st.last_sent_at is None:
        return True
    ref = st.last_sent_at
    if ref.tzinfo is None:
        ref = ref.replace(tzinfo=UTC)
    gap = required_gap_days(template_key, st.sent_count)
    # Sub-day tolerance: crons fire at a fixed wall-clock time, so
    # "exactly N days later" lands a few seconds early or late depending
    # on scheduler drift and DST. Without the slack a daily-gap reminder
    # skips a day roughly whenever the tick runs early, which reads as
    # the ladder being one step further along than it is.
    deadline = ref + timedelta(days=gap) - timedelta(hours=2)
    return (now or datetime.now(UTC)) >= deadline


async def record_sent(
    db: AsyncSession,
    *,
    recipient_id: UUID,
    template_key: str,
    subject_key: str,
    now: datetime | None = None,
) -> int:
    """Stamp a send and return the new count. Caller commits."""
    stamp = now or datetime.now(UTC)
    st = await _state(
        db, recipient_id=recipient_id, template_key=template_key,
        subject_key=subject_key,
    )
    if st is None:
        # SAVEPOINT around the INSERT. Today's two callers serialise
        # themselves (a cycle row-lock, a 20h floor), so the race is
        # narrow — but this is a shared primitive and the next caller
        # will not necessarily be serialised. A raw IntegrityError here
        # would poison the whole session, which in the inventory path
        # means every REMAINING recipient in the fan-out silently gets
        # nothing: a bookkeeping collision turning into lost
        # notifications. Losing the race is not an error condition —
        # somebody else recorded the same send, and re-reading gives the
        # right answer.
        from sqlalchemy.exc import IntegrityError

        try:
            async with db.begin_nested():
                st = NotificationNagState(
                    recipient_person_id=recipient_id,
                    template_key=template_key,
                    subject_key=subject_key,
                    first_sent_at=stamp,
                    sent_count=0,
                )
                db.add(st)
                await db.flush()
        except IntegrityError:
            st = await _state(
                db, recipient_id=recipient_id, template_key=template_key,
                subject_key=subject_key,
            )
            if st is None:  # pragma: no cover — unique index says otherwise
                raise
    st.last_sent_at = stamp
    st.sent_count = (st.sent_count or 0) + 1
    await db.flush()
    return st.sent_count


async def clear(
    db: AsyncSession,
    *,
    template_key: str,
    subject_key: str,
    recipient_id: UUID | None = None,
) -> int:
    """Forget the nag history for a thing that is now settled.

    Constraint (3): without this the next unrelated ask inherits the old
    count and starts on the weekly rung — a noise fix that turns a brand
    new problem into near-silence. Every caller must invoke it on the
    terminal transition (answered / cancelled / closed).

    ``recipient_id=None`` clears every recipient of that subject, which
    is what a cycle closing wants.
    """
    stmt = select(NotificationNagState).where(
        NotificationNagState.template_key == template_key,
        NotificationNagState.subject_key == subject_key,
    )
    if recipient_id is not None:
        stmt = stmt.where(
            NotificationNagState.recipient_person_id == recipient_id,
        )
    rows = (await db.execute(stmt)).scalars().all()
    for r in rows:
        await db.delete(r)
    if rows:
        await db.flush()
    return len(rows)


async def volume_report(
    db: AsyncSession, *, window_days: int = 30, limit: int = 25,
) -> dict[str, Any]:
    """How much this platform is actually pushing at people, by template.

    Exists because "don't send employees too much" is unfalsifiable
    without a number. Reports per-RECIPIENT averages rather than totals:
    a template that sent 500 cards to 500 people is fine and one that
    sent 500 to 30 people is not, and the total cannot tell them apart —
    which is exactly how a 16.7-per-person reminder sat unnoticed
    underneath a headline that looked like ordinary volume.

    Counts only pushes aimed at ACTIVE employees; admin alert fan-out is
    reported separately so operational paging never dilutes the number
    that is supposed to describe employee experience.
    """
    from sqlalchemy import text

    sql = text("""
        SELECT COALESCE(o.template_key, CONCAT('(untyped) ', LEFT(o.title, 40)))
                   AS k,
               COUNT(*) AS n,
               COUNT(DISTINCT o.recipient_person_id) AS ppl
        FROM notification_outbox o
        JOIN identity_persons p ON p.id = o.recipient_person_id
        WHERE o.created_at >= :since
          AND p.employment_status = 'active'
        GROUP BY 1
        ORDER BY n DESC
        LIMIT :lim
    """)
    since = datetime.now(UTC) - timedelta(days=window_days)
    rows = (
        await db.execute(sql, {"since": since, "lim": limit})
    ).all()
    templates = [
        {
            "template_key": r[0],
            "total": int(r[1]),
            "recipients": int(r[2]),
            "per_recipient": round(int(r[1]) / max(int(r[2]), 1), 1),
        }
        for r in rows
    ]

    top_sql = text("""
        SELECT p.id, p.display_name, p.work_email, COUNT(*) AS n
        FROM notification_outbox o
        JOIN identity_persons p ON p.id = o.recipient_person_id
        WHERE o.created_at >= :since
          AND p.employment_status = 'active'
        GROUP BY p.id, p.display_name, p.work_email
        ORDER BY n DESC
        LIMIT 10
    """)
    top = [
        {
            "person_id": str(r[0]),
            "display_name": r[1],
            "work_email": r[2],
            "total": int(r[3]),
        }
        for r in (await db.execute(top_sql, {"since": since})).all()
    ]
    return {
        "window_days": window_days,
        "templates": templates,
        "busiest_recipients": top,
        # The ladder, expressed the way a human reads it.
        "cadence_projection": {
            "first_month": projected_sends(_LADDER_DEFAULT_KEY, 30),
            "daily_equivalent": 30,
        },
    }