"""The notification engine (M5) — the bell. Design copied wholesale from
review-learnings §4 / review-patterns-ref §2.6: three kinds (mention beats
thread_reply beats nothing), participation-as-subscription, the 160-char
stripped preview, the denormalized context_label, resolved-state propagation.

Port note (plan §6.4, ref §2.6's own port note): the review app created all
of this in security-definer DB triggers because untrusted clients wrote
straight to the DB. Here no client writes directly to the DB, so these are
plain pure-ish functions called IN THE SAME TRANSACTION as the triggering
write — never call db.commit() in this module; the caller's existing commit
(services.conversation.create_comment / services.board.patch_ball /
services.touches.pass_ball) covers these inserts too. mark_read/mark_all_read
are the one exception (they ARE the triggering write) and commit themselves.

Errors raise NotificationError carrying the structured shape ({error_code,
summary, detail}) the router serializes — same error language as the rest
of the app (services.board.BoardError / services.conversation.ConversationError).
"""
from __future__ import annotations

import re
import uuid

from sqlalchemy import func, select, update
from sqlalchemy.orm import Session

from .. import events
from ..models import Ball, Comment, Court, Notification, Player
from . import projection as projection_service
from .perspective import MatchTopology, PerspectiveLens

# The three kinds (ref §1.6: convention-only, no DB CHECK — reproduced as-is).
KIND_MENTION = "mention"
KIND_THREAD_REPLY = "thread_reply"
KIND_ACTION_TO = "action_to"

# source_type values (mirrors services.conversation.PARENT_TYPE_BALL — kept
# as its own literal here rather than imported, to avoid a conversation ->
# notifications -> conversation import cycle; see hook site comments).
SOURCE_COMMENT = "comment"
SOURCE_BALL = "ball"

# The GET /api/notifications page size. A module constant (not a bare 200
# inline) so tests can monkeypatch it down without seeding 200+ rows.
LIST_LIMIT = 200

# Shared preview-stripping idiom, transcribed verbatim from review-patterns-
# ref §2.6 (reused identically at every trigger site there; one copy here).
_MD_IMAGE_RE = re.compile(r"!\[[^\]]*\]\([^)]*\)")
_WHITESPACE_RE = re.compile(r"\s+")

# A bare "@username" mention (no email). conversation.parse_mentions already
# extracts full-email mentions (passed in as `mentioned_emails`); this is the
# OTHER half the task asks for — "also accept @username mentions if the
# email doesn't resolve". The character class stops at a second '@', so
# "@erin@brentwood.example" naturally yields the candidate "erin" too (a
# second path to the same Player, harmless — _resolve_mentioned_players
# de-dupes by id before creating anything).
MENTION_USERNAME_RE = re.compile(r"@([A-Za-z0-9_.\-]+)")


class NotificationError(Exception):
    """Structured service failure; router turns it into HTTP JSON (mirrors
    services.board.BoardError / services.conversation.ConversationError)."""

    def __init__(self, status_code: int, error_code: str, summary: str, detail: str = ""):
        super().__init__(summary)
        self.status_code = status_code
        self.error_code = error_code
        self.summary = summary
        self.detail = detail


# ------------------------------------------------------------------ helpers


def _uuid(v) -> uuid.UUID:
    return v if isinstance(v, uuid.UUID) else uuid.UUID(str(v))


def _preview(content: str | None) -> str:
    """160-char preview: markdown image tags stripped, whitespace collapsed
    — the exact idiom review-patterns-ref §2.6 quotes at every trigger site
    (mention helper, both thread-participant triggers, the action-to
    trigger)."""
    text_ = _MD_IMAGE_RE.sub("", content or "")
    text_ = _WHITESPACE_RE.sub(" ", text_).strip()
    return text_[:160]


def _ball_link(ball_id, comment_id=None) -> str:
    """Deep link: /?focus=ball:<ball_id>[&comment=<comment_id>]. The frontend
    appends its own `_t=<timestamp>` cache-buster on click (ref §3.6) — never
    stored here, so the identical notification re-opens every time it's
    clicked, not just the first."""
    link = f"/?focus=ball:{ball_id}"
    if comment_id is not None:
        link += f"&comment={comment_id}"
    return link


def _username_candidates(body: str) -> list[str]:
    """Bare @username tokens in a comment body, order-preserving de-dup.
    Common trailing punctuation stripped so "@rian," / "@rian." still
    resolve — the char class already excludes '@','!','?' etc but includes
    '.'/'-' (valid username chars), so a sentence-ending period needs an
    explicit rstrip."""
    if not body:
        return []
    raw = MENTION_USERNAME_RE.findall(body)
    cleaned = [t.rstrip(".,!?;:") for t in raw]
    return list(dict.fromkeys(t for t in cleaned if t))


def _resolve_mentioned_players(db: Session, body: str, mentioned_emails: list[str]) -> list[Player]:
    """Union of: emails resolved case-insensitively against Player.email
    (the mentioned_emails argument — already parsed by
    services.conversation.parse_mentions, a full-email match only), and bare
    @username tokens resolved case-insensitively against Player.username
    (this module's own, looser scan — "also accept @username mentions if
    the email doesn't resolve — check both"). De-duped by player id: the
    same person matched via both paths (e.g. a "@user@domain.com" mention
    where the username scan also independently captures the local part)
    must only ever get ONE notification.
    """
    players: dict = {}

    if mentioned_emails:
        lowered = [e.lower() for e in mentioned_emails]
        rows = (
            db.execute(select(Player).where(func.lower(Player.email).in_(lowered)))
            .scalars()
            .all()
        )
        for p in rows:
            players[p.id] = p

    usernames = _username_candidates(body)
    if usernames:
        lowered_u = [u.lower() for u in usernames]
        rows = (
            db.execute(select(Player).where(func.lower(Player.username).in_(lowered_u)))
            .scalars()
            .all()
        )
        for p in rows:
            players[p.id] = p

    return list(players.values())


def _sse_summary(n: Notification) -> dict:
    """JSON-primitive summary for the SSE payload. events.publish() hands
    this straight to routers.events' plain `json.dumps` — no Pydantic in
    between — so every UUID/datetime must already be a string here, same
    convention as services.touches._publish / services.conversation's
    publish() calls."""
    return {
        "id": str(n.id),
        "recipient_id": str(n.recipient_id),
        "actor_id": str(n.actor_id) if n.actor_id else None,
        "kind": n.kind,
        "category": n.category,
        "context_label": n.context_label,
        "body": n.body,
        "link": n.link,
        "read": n.read,
        "resolved": n.resolved,
        "source_type": n.source_type,
        "source_id": str(n.source_id),
        "ball_id": str(n.ball_id),
        "created_at": n.created_at.isoformat(),
    }


def _create(
    db: Session,
    *,
    recipient: Player,
    actor: Player | None,
    kind: str,
    ball: Ball,
    body: str,
    link: str,
    source_type: str,
    source_id,
) -> Notification:
    notification = Notification(
        recipient_id=recipient.id,
        actor_id=actor.id if actor else None,
        kind=kind,
        category=ball.category,
        context_label=ball.title,
        body=body,
        link=link,
        source_type=source_type,
        source_id=source_id,
        ball_id=ball.id,
    )
    db.add(notification)
    db.flush()  # id + created_at for the SSE payload below; caller commits.
    court = db.get(Court, ball.court_id)  # identity-map hit at every call site
    events.publish(
        {
            "type": "notification.created",
            "recipient_id": str(recipient.id),
            # match_id lets the SSE boundary project the actor through the
            # recipient's lens (services.projection.project_event) — and
            # deliver to the RECIPIENT'S stream only.
            "match_id": str(court.match_id) if court else None,
            "notification": _sse_summary(notification),
        }
    )
    return notification


# ------------------------------------------------------------------ writes
# (called from other services' transactions — see module docstring)


def notify_comment_created(
    db: Session,
    ball: Ball,
    comment: Comment,
    author: Player | None,
    mentioned_emails: list[str],
) -> None:
    """Comment-authored notifications. Called by
    services.conversation.create_comment right after its db.flush() gives
    the comment an id, before its own db.commit() — same transaction as the
    triggering write (plan §6.4).

    Two kinds, mention beats thread_reply (review-learnings §4's dedup rule):
      (a) mention        every Player resolved from mentioned_emails (by
                         email) or a bare @username token in the body (by
                         username) — see _resolve_mentioned_players. SELF-
                         MENTIONS ALLOWED (ref §2.6.1's un-block, ported:
                         "rian wants self-pings... as reminders").
      (b) thread_reply   ball.author + every distinct PRIOR commenter on
                         this ball, EXCLUDING the new comment's author and
                         anyone already notified via (a) for this comment.

    M6 perspective closes over the bell too (§4.3 — a notification is a
    read surface; its 160-char preview must never leak what the timeline
    would collapse):
      - a MENTION only resolves to players VISIBLE to the author — typing
        "@zeina" in a raw request cannot reach a player the author's game
        does not contain (the composer autocomplete never offers one);
      - a THREAD_REPLY only reaches participants who can SEE this comment
        (node_visible on its acting_node_id) — Tingang chatter on Rian's
        ball never pings Rian; he gets the collapsed line when he opens
        the timeline, exactly like the plan's founding promise.
    Both checks use the topology as of NOW (write time): notifications
    are point-in-time sends, not re-projected history.
    """
    court = db.get(Court, ball.court_id)
    topology = MatchTopology(db, court.match_id) if court else None

    mentioned_players = _resolve_mentioned_players(db, comment.body, mentioned_emails)
    if topology is not None:
        author_lens = PerspectiveLens(topology, author.id if author else None)
        mentioned_players = [
            p for p in mentioned_players if author_lens.player_visible(p.id)
        ]
    mentioned_ids = {p.id for p in mentioned_players}

    preview = _preview(comment.body)
    link = _ball_link(ball.id, comment.id)

    for player in mentioned_players:
        _create(
            db,
            recipient=player,
            actor=author,
            kind=KIND_MENTION,
            ball=ball,
            body=preview,
            link=link,
            source_type=SOURCE_COMMENT,
            source_id=comment.id,
        )

    # Participation = subscription: the ball's author plus every distinct
    # PRIOR commenter (not this new one — Comment.id != comment.id).
    participant_ids: set = set()
    if ball.author_id is not None:
        participant_ids.add(ball.author_id)
    prior_authors = (
        db.execute(
            select(Comment.author_id)
            .where(
                Comment.parent_type == "ball",
                Comment.parent_id == ball.id,
                Comment.id != comment.id,
                Comment.author_id.is_not(None),
            )
            .distinct()
        )
        .scalars()
        .all()
    )
    participant_ids.update(prior_authors)

    # Dedup rule: never the new comment's own author; never anyone who
    # already got a mention out of this same comment.
    if author is not None:
        participant_ids.discard(author.id)
    participant_ids -= mentioned_ids

    # M6 scoping (docstring): only participants who can see this comment's
    # node get the preview.
    if topology is not None:
        participant_ids = {
            pid
            for pid in participant_ids
            if topology.node_visible(pid, comment.acting_node_id)
        }

    if not participant_ids:
        return
    participants = db.execute(select(Player).where(Player.id.in_(participant_ids))).scalars().all()
    for player in participants:
        _create(
            db,
            recipient=player,
            actor=author,
            kind=KIND_THREAD_REPLY,
            ball=ball,
            body=preview,
            link=link,
            source_type=SOURCE_COMMENT,
            source_id=comment.id,
        )


def notify_action_to_changed(
    db: Session,
    ball: Ball,
    actor: Player | None,
    old_action_to: list,
    new_action_to: list,
) -> None:
    """action_to notifications for NEWLY ADDED assignees only (array diff vs
    old — ref §2.6.4's array-diff idiom), excluding the actor (you never
    need telling that you assigned yourself, or that your own automation
    moved the ball). Called at every site action_to can change: the
    explicit PATCH (services.board.patch_ball), the touch automations
    (spike / revision-required restore / reopen restore in
    services.touches.apply_status_automations), and the explicit pass
    endpoint (services.touches.pass_ball).
    """
    old_ids = {_uuid(p) for p in (old_action_to or [])}
    new_ids = {_uuid(p) for p in (new_action_to or [])}
    newly_added = new_ids - old_ids
    if actor is not None:
        newly_added.discard(actor.id)
    if not newly_added:
        return

    preview = _preview(ball.body)
    link = _ball_link(ball.id)
    recipients = db.execute(select(Player).where(Player.id.in_(newly_added))).scalars().all()
    for player in recipients:
        _create(
            db,
            recipient=player,
            actor=actor,
            kind=KIND_ACTION_TO,
            ball=ball,
            body=preview,
            link=link,
            source_type=SOURCE_BALL,
            source_id=ball.id,
        )


def propagate_resolved(db: Session, ball: Ball, resolved: bool) -> None:
    """Mirror `resolved` onto every notification this ball produced — ONE
    UPDATE keyed on the denormalized ball_id (no link-parsing, no join
    through comments). review-learnings §4: "a notification's lifecycle is
    tied to the underlying work item, not just to whether it's been read."
    Reopen calls this with resolved=False to flip it back."""
    db.execute(update(Notification).where(Notification.ball_id == ball.id).values(resolved=resolved))


# ------------------------------------------------------------------- reads


def _item_dict(n: Notification) -> dict:
    """NotificationOut shape as a plain dict — the projection below must
    never write a rep()'d actor onto a session-attached row."""
    return {
        "id": n.id,
        "recipient_id": n.recipient_id,
        "actor_id": n.actor_id,
        "kind": n.kind,
        "category": n.category,
        "context_label": n.context_label,
        "body": n.body,
        "link": n.link,
        "read": n.read,
        "resolved": n.resolved,
        "source_type": n.source_type,
        "source_id": n.source_id,
        "ball_id": n.ball_id,
        "created_at": n.created_at,
    }


def list_for_recipient(db: Session, recipient_id) -> dict:
    """The viewer's notifications, newest first, capped at LIST_LIMIT.
    unread_count/unresolved_count are TRUE TOTALS, not capped by the page
    size — ref §3.6: two independent counts that move separately and both
    drive the bell badge, so they must stay correct even past the listing
    cap.

    M6: each item's ACTOR renders through rep() for the recipient (§4.3 —
    "notification actors render through rep()"): invisible actors become
    their boundary lead or actor_id None + actor_label. Topologies are
    loaded once per MATCH represented in the page (in practice one), never
    per item."""
    items = (
        db.execute(
            select(Notification)
            .where(Notification.recipient_id == recipient_id)
            .order_by(Notification.created_at.desc())
            .limit(LIST_LIMIT)
        )
        .scalars()
        .all()
    )

    # ball -> match map for the page, one grouped query; then one lens per
    # match for this recipient.
    ball_ids = {n.ball_id for n in items}
    match_by_ball: dict = {}
    if ball_ids:
        rows = db.execute(
            select(Ball.id, Court.match_id)
            .join(Court, Ball.court_id == Court.id)
            .where(Ball.id.in_(ball_ids))
        ).all()
        match_by_ball = dict(rows)
    lens_by_match: dict = {}

    projected_items = []
    for n in items:
        item = _item_dict(n)
        match_id = match_by_ball.get(n.ball_id)
        if match_id is not None and n.actor_id is not None:
            lens = lens_by_match.get(match_id)
            if lens is None:
                lens = PerspectiveLens(MatchTopology(db, match_id), recipient_id)
                lens_by_match[match_id] = lens
            item = projection_service.project_notification_actor(lens, item)
        projected_items.append(item)
    unread_count = db.execute(
        select(func.count())
        .select_from(Notification)
        .where(Notification.recipient_id == recipient_id, Notification.read.is_(False))
    ).scalar_one()
    unresolved_count = db.execute(
        select(func.count())
        .select_from(Notification)
        .where(Notification.recipient_id == recipient_id, Notification.resolved.is_(False))
    ).scalar_one()
    return {
        "items": projected_items,
        "unread_count": unread_count,
        "unresolved_count": unresolved_count,
    }


def mark_read(db: Session, notification_id, actor: Player | None) -> Notification:
    """Mark one notification read. Gated by RECIPIENT, not authorship —
    only the notification's own recipient may mark it read. Anyone else's
    id (or no signed-in player at all) is reported as NOT_FOUND rather than
    FORBIDDEN, so the endpoint never confirms that a given id belongs to
    someone else (mirrors why board/conversation 404 first on lookup)."""
    notification = db.get(Notification, notification_id)
    if notification is None or actor is None or notification.recipient_id != actor.id:
        raise NotificationError(
            404, "NOTIFICATION_NOT_FOUND", f"No notification {notification_id}"
        )
    if not notification.read:
        notification.read = True
        db.commit()
        db.refresh(notification)
    return notification


def mark_all_read(db: Session, actor: Player | None) -> int:
    """Mark every unread notification for the session player read. No
    Player row yet -> nothing is theirs -> 0, not an error (mirrors GET's
    empty-state tolerance, not board/conversation's 404-on-lookup shape,
    since there's no single id being looked up here)."""
    if actor is None:
        return 0
    result = db.execute(
        update(Notification)
        .where(Notification.recipient_id == actor.id, Notification.read.is_(False))
        .values(read=True)
    )
    db.commit()
    return result.rowcount or 0