"""One error string in, one stable cluster signature out.

The platform already had this idea in exactly one place —
:func:`app.notifications.service.normalize_error_signature`, which
collapses outbox ``last_error`` values so the admin sees "Feishu
open_id not found for <email>" once instead of forty times. That
function is scoped to notifications and stays there; this module is the
domain-agnostic version every source collector shares, because the
whole point of the exception ledger is that a Zoom 429 arriving through
a provisioning task and the same 429 arriving through an identity-sync
run item land on **one** row.

Two functions matter:

* :func:`canonical_signature` — the human-readable cluster label. Row
  identity (emails, UUIDs, ids, numbers, timestamps, quoted paths) is
  stripped; the sentence still reads as the error it came from, because
  an admin has to recognise it at a glance.
* :func:`signature_hash` — the 32-char digest actually stored and
  indexed. Signatures are prose and prose is long; a hash gives the
  unique constraint a fixed-width key without truncating the label.

**Scoping is part of the key, not part of the text.** The same
canonical sentence from two different integrations is two different
problems with two different fixes, so the hash mixes in the source
domain and the scope key (integration slug / handler key / target
slug). Getting this wrong in either direction is a real cost: too
coarse and "rate limited" from Zoom merges with "rate limited" from
Google, and a fix verified against one closes the other; too fine and
the ledger degenerates into the per-row list it was built to replace.
"""
from __future__ import annotations

import hashlib
import re

#: Order matters — the email pattern must run before the generic word
#: scrubbers, and the UUID pattern before the hex one (a UUID is hex
#: with dashes and would otherwise be shredded into five fragments).
_EMAIL_RE = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
_UUID_RE = re.compile(
    r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
    r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b",
)
#: Bare 32-char hex (Feishu open_id / union_id, request ids, our own
#: CHAR(32) ids) and anything longer that looks like an opaque token.
_HEX_RE = re.compile(r"\b[0-9a-fA-F]{16,}\b")
#: ISO-8601-ish timestamps, with or without timezone.
_TS_RE = re.compile(
    r"\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}(:\d{2})?(\.\d+)?"
    r"(Z|[+-]\d{2}:?\d{2})?",
)
_DATE_RE = re.compile(r"\b\d{4}-\d{2}-\d{2}\b")
#: Vendor ids that travel as ``ou_xxx`` / ``on_xxx`` / ``sk-xxx`` /
#: ``AKIA...`` shaped tokens. Deliberately narrow: an unanchored
#: alphanumeric rule would eat error vocabulary.
_TOKEN_RE = re.compile(
    r"\b(?:ou|on|oc|om|od|open|union|cli|sk|ak|AKIA|ASIA)"
    r"[-_][A-Za-z0-9_-]{6,}\b",
)
#: Quoted or bracketed payloads — a JSON body or a file path inside an
#: error message is per-row detail, never cluster identity.
_QUOTED_RE = re.compile(r"(['\"])(?:(?!\1).){12,}?\1")
_BRACED_RE = re.compile(r"\{[^{}]{12,}\}")
#: Numbers last, and NOT digit-adjacent to a letter: HTTP status codes
#: and vendor error codes ARE cluster identity ("429", "3412", "3780"),
#: so only multi-digit runs that stand alone as counts get masked.
#: The distinction is load-bearing — masking "429" would merge a rate
#: limit with a not-found, and masking nothing would split one cluster
#: into one row per affected employee count.
_COUNT_RE = re.compile(r"(?<![\w.-])\d{5,}(?![\w.-])")
_WS_RE = re.compile(r"\s+")

#: What a source hands us when the row genuinely carries no error text.
#: A real value, not an empty string: "the vendor said nothing" is a
#: cluster of its own and the commonest shape of a silent failure.
NO_ERROR_TEXT = "(no error text recorded)"

#: Hard cap on the stored label. Long enough to keep a vendor sentence
#: intact, short enough to render in a table cell without wrapping the
#: whole row.
_MAX_SIGNATURE_LEN = 200


def canonical_signature(error_text: str | None) -> str:
    """Collapse one error string into its cluster label.

    Returns :data:`NO_ERROR_TEXT` for ``None`` / blank input rather
    than an empty string, so callers never have to special-case the
    "failed with nothing to say" row — it clusters like any other.
    """
    if error_text is None or not error_text.strip():
        return NO_ERROR_TEXT
    text = error_text.strip()
    # Take a generous prefix before scrubbing: vendor stack traces can
    # be kilobytes, and the identifying sentence is always at the top.
    text = text[:1200]
    text = _TS_RE.sub("<ts>", text)
    text = _DATE_RE.sub("<date>", text)
    text = _EMAIL_RE.sub("<email>", text)
    text = _UUID_RE.sub("<id>", text)
    text = _TOKEN_RE.sub("<token>", text)
    text = _HEX_RE.sub("<hex>", text)
    text = _QUOTED_RE.sub("<value>", text)
    text = _BRACED_RE.sub("<payload>", text)
    text = _COUNT_RE.sub("<n>", text)
    text = _WS_RE.sub(" ", text).strip()
    if not text:
        return NO_ERROR_TEXT
    return text[:_MAX_SIGNATURE_LEN]


def signature_hash(
    *,
    source: str,
    scope_key: str | None,
    signature: str,
) -> str:
    """Stable 32-char digest for ``(source, scope, signature)``.

    ``scope_key`` is the narrowest deterministic thing that identifies
    *where* the failure happened — an integration slug, a cron handler
    key, a sync-target slug. ``None`` is allowed and means "the source
    itself is the scope"; it hashes distinctly from the empty string so
    a collector that starts supplying a scope later cannot silently
    merge its new rows into its old ones.
    """
    scope = "\x00" if scope_key is None else scope_key.strip().lower()
    payload = "\x1f".join((source.strip().lower(), scope, signature))
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:32]