"""Bidirectional translation between Google Cloud Identity dynamic-group
**membership queries** (CEL strings) and Helix's **structured rule**
model, plus a local evaluator for the expected-vs-actual reconcile.

Google's dynamic groups compute membership from a CEL query such as::

    user.custom_schemas.Department_Information.L1_Dept=='Engineering Group'
    || user.custom_schemas.Department_Information.L1_Dept=='Product'
    || user.emails.exists(email, email.address.startsWith('lanxi.xiong@pingcap.com'))

HR must never hand-write CEL. Instead the UI edits a :class:`StructuredRule`
(one custom-schema field + a set of matched values + a few explicitly
included people + an optional negation) and Helix compiles it back to the
exact CEL string Google expects. We also parse the existing 46 production
groups back into structured rules on import; anything that doesn't fit the
structured shape (``&&``, ``user.suspended``, ``org_unit_id``, mixed
fields, …) is marked non-parseable and shown read-only ("managed in
Google").

The grammar we support (top-level ``||`` only):

* ``user.custom_schemas.<Schema>.<Field>=='<value>'``  (all terms same field)
* ``user.emails.exists(email, email.address.startsWith('<prefix>'))``
* optional ``!( ... )`` wrapping the whole thing → ``negate``

Anything else → ``parseable=False``.
"""
from __future__ import annotations

import re
from dataclasses import dataclass
from dataclasses import field as dc_field

# Fields the structured builder understands. ``<Schema>.<Field>`` as it
# appears in the CEL path ``user.custom_schemas.<Schema>.<Field>``.
KNOWN_FIELDS: tuple[str, ...] = (
    "Department_Information.L1_Dept",
    "Department_Information.L2_Dept",
    "Department_Information.L3_Dept",
    "Department_Information.L4_Dept",
    "Employee_Details.Office",
    "Employee_Details.Country",
    "Employee_Details.JobFunction",
    "Employee_Details.JobFamily",
    "Employee_Details.Location",
)

_SCHEMA_EQ_RE = re.compile(
    r"^user\.custom_schemas\.(\w+)\.(\w+)\s*==\s*'([^']*)'$"
)
_EMAIL_EXISTS_RE = re.compile(
    r"^user\.emails\.exists\(\s*email\s*,\s*"
    r"email\.address\.startsWith\('([^']*)'\)\s*\)$"
)


@dataclass(slots=True)
class StructuredRule:
    """A dynamic-group membership rule in the shape HR edits.

    * ``field`` — one of :data:`KNOWN_FIELDS`, or ``None`` when the rule
      only pins explicit people (no attribute clause).
    * ``match_values`` — attribute values OR'd together on ``field``.
    * ``include_emails`` — people OR'd in by email prefix regardless of
      attributes (the ``emails.exists(...startsWith...)`` clauses).
    * ``negate`` — the whole attribute/email disjunction is wrapped in
      ``!( ... )`` (e.g. "everyone whose Country != CN").
    * ``parseable`` — ``False`` means the source CEL didn't fit the
      structured grammar; ``raw_query`` is then the source of truth and
      the UI shows it read-only.
    * ``raw_query`` — the original/queried CEL (always retained).
    """

    field: str | None = None
    match_values: list[str] = dc_field(default_factory=list)
    include_emails: list[str] = dc_field(default_factory=list)
    negate: bool = False
    parseable: bool = True
    raw_query: str = ""

    def to_dict(self) -> dict:
        return {
            "field": self.field,
            "match_values": list(self.match_values),
            "include_emails": list(self.include_emails),
            "negate": self.negate,
            "parseable": self.parseable,
            "raw_query": self.raw_query,
        }


class RuleValidationError(ValueError):
    """Raised when a structured rule can't be safely compiled to CEL."""


def _split_top_level_or(query: str) -> list[str]:
    """Split on ``||`` — safe here because none of our supported term
    shapes contain ``||`` (the ``emails.exists`` lambda uses ``,`` and
    ``.``). Whitespace/newlines around terms are stripped.
    """
    return [t.strip() for t in query.split("||") if t.strip()]


def parse_membership_query(query: str | None) -> StructuredRule:
    """Parse a Google CEL membership query into a :class:`StructuredRule`.

    Never raises — an unrecognised shape yields ``parseable=False`` with
    ``raw_query`` preserved, so import of the existing tenant can't fail
    on a weird group.
    """
    raw = (query or "").strip()
    rule = StructuredRule(raw_query=raw, parseable=False)
    if not raw:
        # Empty query: structurally trivial but nothing to edit — keep
        # it non-parseable so the UI treats it as Google-managed.
        return rule

    body = raw
    negate = False
    # Whole-expression negation: ``!( ... )``.
    if body.startswith("!(") and body.endswith(")"):
        negate = True
        body = body[2:-1].strip()

    # Any AND, or attribute/expr we don't model → bail to read-only.
    if "&&" in body:
        return rule

    terms = _split_top_level_or(body)
    if not terms:
        return rule

    fields: set[str] = set()
    values: list[str] = []
    emails: list[str] = []
    for term in terms:
        m = _SCHEMA_EQ_RE.match(term)
        if m:
            schema, fld, val = m.group(1), m.group(2), m.group(3)
            fields.add(f"{schema}.{fld}")
            values.append(val)
            continue
        m = _EMAIL_EXISTS_RE.match(term)
        if m:
            emails.append(m.group(1))
            continue
        # Unrecognised term (user.suspended, org_unit_id, email.type,
        # a different operator, …) → not structured-editable.
        return rule

    if len(fields) > 1:
        # Mixed fields in one OR — outside the single-field model.
        return rule

    rule.field = next(iter(fields)) if fields else None
    rule.match_values = values
    rule.include_emails = emails
    rule.negate = negate
    rule.parseable = True
    return rule


def compile_rule(
    *,
    field: str | None,
    match_values: list[str] | None,
    include_emails: list[str] | None = None,
    negate: bool = False,
) -> str:
    """Compile a structured rule back into a Google CEL membership query.

    Raises :class:`RuleValidationError` on anything Google would reject
    or that would silently change semantics (unknown field, empty rule,
    single-quote in a value/email — CEL string literals use single
    quotes and we don't emit escapes).
    """
    values = [v.strip() for v in (match_values or []) if v.strip()]
    emails = [e.strip() for e in (include_emails or []) if e.strip()]

    if field is not None and field not in KNOWN_FIELDS:
        raise RuleValidationError(f"unknown field {field!r}")
    if field is not None and not values:
        raise RuleValidationError(
            "a field-based rule needs at least one match value"
        )
    if field is None and not emails:
        raise RuleValidationError("empty rule (no field, no included people)")

    for v in values:
        if "'" in v:
            raise RuleValidationError(
                f"value {v!r} contains a single quote (unsupported in CEL)"
            )
    for e in emails:
        if "'" in e:
            raise RuleValidationError(
                f"email {e!r} contains a single quote (unsupported in CEL)"
            )

    terms: list[str] = []
    if field is not None:
        terms.extend(
            f"user.custom_schemas.{field}=='{v}'" for v in values
        )
    terms.extend(
        f"user.emails.exists(email, email.address.startsWith('{e}'))"
        for e in emails
    )
    joined = " || ".join(terms)
    return f"!({joined})" if negate else joined


# ── Local evaluator (expected-vs-actual reconcile) ─────────────────────
#
# We reproduce Google's membership computation for a structured rule
# against a snapshot of user attributes so HR can preview "who this rule
# matches" and diff it against Google's actual member list. Only the
# structured shape is evaluable; non-parseable rules return None.


@dataclass(slots=True)
class UserSnapshot:
    """Minimal per-user attribute snapshot for rule evaluation."""

    email: str
    # {"Department_Information.L1_Dept": "Engineering Group", ...}
    fields: dict[str, str]
    suspended: bool = False


def evaluate_rule(
    rule: StructuredRule, users: list[UserSnapshot],
) -> set[str] | None:
    """Return the set of emails a structured rule matches, or ``None`` if
    the rule isn't structured-evaluable. Dynamic groups only include
    non-suspended users, matching Google's behaviour.
    """
    if not rule.parseable:
        return None
    value_set = set(rule.match_values)
    prefixes = tuple(rule.include_emails)
    matched: set[str] = set()
    for u in users:
        if u.suspended:
            continue
        attr_hit = False
        if rule.field is not None:
            fv = u.fields.get(rule.field)
            attr_hit = fv is not None and fv in value_set
        email_hit = bool(prefixes) and u.email.lower().startswith(
            tuple(p.lower() for p in prefixes)
        )
        base = attr_hit or email_hit
        if rule.negate:
            # ``!( attr || email )`` — note email-include people are also
            # excluded under negation, matching Google's literal CEL.
            if not base:
                matched.add(u.email)
        elif base:
            matched.add(u.email)
    return matched


def dead_values(
    rule: StructuredRule, live_value_counts: dict[str, dict[str, int]],
) -> list[str]:
    """Return the match values referenced by ``rule`` that currently
    match **zero** live (active) users — the "died because HR renamed the
    department" case that silently empties a group.

    ``live_value_counts`` maps ``"<Schema>.<Field>" -> {value: count}``
    from a scan of active users. Only field-based, non-negated rules
    produce dead-value warnings (a negated rule matching zero of a value
    is fine; a pure-people rule has no attribute values).
    """
    if not rule.parseable or rule.field is None or rule.negate:
        return []
    counts = live_value_counts.get(rule.field, {})
    return [v for v in rule.match_values if counts.get(v, 0) == 0]