from __future__ import annotations

from collections.abc import Iterable, Mapping
from typing import Any


BROAD_SCAN_OPERATORS = ("tablefullscan", "indexfullscan")


def explain_has_broad_scan(rows: Iterable[Mapping[str, Any]]) -> bool:
    """Fail closed on a full scan even when TiDB names only the table alias."""
    for row in rows:
        text = " ".join(str(value) for value in row.values()).lower()
        if any(operator in text for operator in BROAD_SCAN_OPERATORS):
            return True
    return False


def active_lease_sample_violations(
    sample: Mapping[str, Any],
    *,
    expired_processing: int = 0,
) -> tuple[str, ...]:
    """Return row-level WorkItem lease invariant failures for one DB snapshot."""
    active = int(sample.get("leased_work_items") or 0)
    violations: list[str] = []
    if expired_processing > 0:
        violations.append("expired_or_unleased_processing")
    if active <= 0:
        return tuple(violations)

    row_failures = (
        ("missing_effective_lane", "missing_effective_lane_count"),
        ("missing_lease_owner", "missing_lease_owner_count"),
        ("missing_lease_token", "missing_lease_token_count"),
        ("nonpositive_ownership_epoch", "nonpositive_ownership_epoch_count"),
    )
    for label, key in row_failures:
        if int(sample.get(key) or 0) > 0:
            violations.append(label)

    if int(sample.get("leased_lanes") or 0) != active:
        violations.append("duplicate_effective_lane")
    if int(sample.get("lease_tokens") or 0) != active:
        violations.append("duplicate_lease_token")
    return tuple(violations)


def valid_active_lease_width(sample: Mapping[str, Any], *, expired_processing: int = 0) -> int:
    active = int(sample.get("leased_work_items") or 0)
    if active <= 0 or active_lease_sample_violations(sample, expired_processing=expired_processing):
        return 0
    return active