#!/usr/bin/env python3
"""Retired Mail artifacts stay deleted and the Mail vertical slice does not
leak into generic extension packages.

Ported from internal/extension/providers/mail/deletion_ratchet_test.go and
package_boundary_test.go so the checks run as harness guards without
expanding the source-assertion test-file ratchet. The scanning rules,
allowlist stale-entry discipline, tracked-file semantics, package-boundary
exemptions, and failure diagnostics must stay equivalent to the retired
Go scanners.
"""
import os
import re
import subprocess
import sys
from pathlib import Path

GUARD_ROOT = Path(
    os.environ.get("AGENTTEAM_RETIRED_MAIL_LEAK_GUARD_ROOT")
    or Path(__file__).resolve().parents[2]
)

MAIL_PROVIDER_PREFIX = "internal/extension/providers/mail/"
MAIL_STORE_PREFIX = MAIL_PROVIDER_PREFIX + "mailstore/"
MAIL_RUNTIME_PREFIX = MAIL_PROVIDER_PREFIX + "mailruntime/"

CONTENT_PATTERNS = [
    ("gmail", "gmail", True,
     re.compile(r"(\bgmail\b|provider_?gmail\b|gmail(client|provider|adapter|api|runtime|service|store|oauth)\b)", re.IGNORECASE)),
    ("microsoft graph", "microsoft", True,
     re.compile(r"(\bmicrosoft[_ -]*graph\b|providermicrosoftgraph\b|graph\.microsoft\.com)", re.IGNORECASE)),
    ("mail oauth", "oauth", True, re.compile(r"oauth", re.IGNORECASE)),
    ("mail graph provider", "graph", True,
     re.compile(r"(\bgraph\b|graph(client|provider|adapter|runtime|service)\b)", re.IGNORECASE)),
    ("legacy mail tables", "mail", False,
     re.compile(
         r"\bmail[_ -]*(oauth[_ -]*flows|provider[_ -]*setup[_ -]*drafts|"
         r"connection[_ -]*capabilities|resource[_ -]*refs|"
         r"draft[_ -]*revisions|send[_ -]*snapshots|"
         r"operation[_ -]*effects|attachment[_ -]*refs)",
         re.IGNORECASE)),
    ("get-operation-status", "operation", False,
     re.compile(r"get[-_ ]*operation[-_ ]*status", re.IGNORECASE)),
    ("prepare-send", "prepare", False,
     re.compile(r"prepare[-_ ]*send", re.IGNORECASE)),
]

ALLOWLIST = {
    # These tests make negative assertions about the retired contracts.
    "internal/extension/providers/mail/mailapi/routes_test.go": {
        "mail oauth",
    },
    "internal/extension/providers/mail/definition_test.go": {
        "gmail",
        "microsoft graph",
        "mail oauth",
        "get-operation-status",
    },
    "internal/extension/providers/mail/mailruntime/connection_material_test.go": {
        "prepare-send",
    },
    "internal/tidb/email_app_migration_contract_test.go": {
        "gmail",
        "microsoft graph",
        "mail oauth",
        "get-operation-status",
        "prepare-send",
    },
    # This is the required provider-neutral icon statement.
    "web/public/app-icons/README.md": {
        "gmail",
    },
}

GMAIL_ADDRESS_PATTERN = re.compile(
    r"[a-z0-9.!#$%&'*+/=?^_`{|}~-]+@gmail\.com", re.IGNORECASE
)

ARTIFACT_PATH_PATTERN = re.compile(
    r"(^|[/_.-])mail[-_.]*(oauth[-_.]*flows|"
    r"provider[-_.]*setup[-_.]*drafts|connection[-_.]*capabilities|"
    r"resource[-_.]*refs|draft[-_.]*revisions|send[-_.]*snapshots|"
    r"operation[-_.]*effects|attachment[-_.]*refs)",
    re.IGNORECASE,
)

ACTION_PATH_PATTERN = re.compile(
    r"(get[-_. ]*operation[-_. ]*status|prepare[-_. ]*send)", re.IGNORECASE
)

OWNERSHIP_PATTERN = re.compile(
    r"(\b(email|mail) app\b|\bapp_email\b|/apps/email\b|"
    r"\bmail_(oauth_flows|provider_setup_drafts|connection_capabilities|"
    r"resource_refs|draft_revisions|send_snapshots|operation_effects|"
    r"attachment_refs)\b|get[-_ ]*operation[-_ ]*status|"
    r"send[-_ ]*draft)",
    re.IGNORECASE,
)

CONTENT_EXTENSIONS = {".go", ".sql", ".md", ".mdx", ".rst", ".adoc"}
TEXT_EXTENSIONS = {
    ".json", ".yaml", ".yml", ".txt", ".txtar", ".golden", ".toml",
    ".xml", ".html", ".csv", ".eml",
}

MAIL_STORE_KEYWORDS = (
    "attachment", "attachments", "capability", "capabilities",
    "draft", "drafts", "effect", "effects", "mailbox", "mailboxes",
    "repository", "resource", "resources",
)
MAIL_RUNTIME_KEYWORDS = ("gmail", "graph", "provider", "refresh")

GENERIC_PACKAGE_ROOTS = (
    "internal/service/extensionflow",
    "internal/store/extension",
    "internal/store/app",
)
BOUNDARY_EXEMPT_PATHS = {
    "internal/store/app/registry/product_apps.go",
}
BOUNDARY_FORBIDDEN_TOKENS = (
    "internal/extension/mail",
    "mail_connection_capabilities",
    "mail.metadata.read",
    "mail.content.read",
    "mail.draft.write",
    "mail.send",
    "app_email",
    "EmailAppID",
    "EmailAppSlug",
)


def tracked_repository_paths(root: Path) -> list:
    output = subprocess.run(
        ["git", "-C", str(root), "ls-files", "-z", "--cached"],
        check=True,
        capture_output=True,
    ).stdout.decode()
    return [part for part in output.split("\x00") if part]


def scan_excluded(relative_path: str) -> bool:
    return (
        relative_path.startswith("x/")
        or relative_path.startswith("vendor/")
        or "/vendor/" in relative_path
    )


def path_words(path: str) -> set:
    return {word for word in re.split(r"[^a-z0-9]+", path) if word}


def path_has_component(path: str, component: str) -> bool:
    return component in path.split("/")


def path_finding(relative_path: str) -> str:
    lower_path = relative_path.lower()
    if scan_excluded(lower_path):
        return ""
    words = path_words(lower_path)
    if "gmail" in lower_path:
        return "contains forbidden gmail token"
    if (
        "microsoftgraph" in lower_path
        or "microsoft_graph" in lower_path
        or "microsoft-graph" in lower_path
    ):
        return "contains forbidden Microsoft Graph tokens"
    if ARTIFACT_PATH_PATTERN.search(lower_path):
        return "contains a legacy mail_* artifact name"
    if ACTION_PATH_PATTERN.search(lower_path):
        return "contains a retired action name"

    mail_owned = (
        lower_path.startswith(MAIL_PROVIDER_PREFIX)
        or "mail" in words
        or "email" in words
    )
    if not mail_owned:
        return ""
    base = lower_path.rsplit("/", 1)[-1]
    if lower_path.endswith(".log"):
        return "is a tracked Mail log artifact"
    if "oauth" in lower_path:
        return "contains forbidden oauth token"
    if lower_path.startswith(MAIL_PROVIDER_PREFIX) and "graph" in base:
        return "contains forbidden graph token"
    if lower_path.startswith(MAIL_STORE_PREFIX):
        for keyword in MAIL_STORE_KEYWORDS:
            if keyword in base:
                return "contains legacy Mail store " + keyword
    if lower_path.startswith(MAIL_RUNTIME_PREFIX):
        for keyword in MAIL_RUNTIME_KEYWORDS:
            if keyword in base:
                return "contains legacy Mail runtime " + keyword
    return ""


def text_extension(path: str) -> bool:
    return extension_of(path) in TEXT_EXTENSIONS


def extension_of(path: str) -> str:
    base = path.rsplit("/", 1)[-1]
    return base[base.rfind("."):] if "." in base else ""


def content_candidate(relative_path: str) -> bool:
    lower_path = relative_path.lower()
    if extension_of(lower_path) in CONTENT_EXTENSIONS:
        return True
    if (lower_path.startswith("docs/") or lower_path.startswith("specs/")) and \
            text_extension(lower_path):
        return True
    if not (
        path_has_component(lower_path, "testdata")
        or path_has_component(lower_path, "fixture")
        or path_has_component(lower_path, "fixtures")
        or "fixture" in lower_path.rsplit("/", 1)[-1]
    ):
        return False
    return text_extension(lower_path)


def content_is_mail_owned(relative_path: str, text: str) -> bool:
    lower_path = relative_path.lower()
    if lower_path.startswith(MAIL_PROVIDER_PREFIX):
        return True
    tokens = path_words(lower_path)
    if "mail" in tokens or "email" in tokens:
        return True
    return bool(OWNERSHIP_PATTERN.search(text))


def content_findings(relative_path: str, payload: str) -> list:
    if scan_excluded(relative_path):
        return []
    text = payload
    lower_text = text.lower()
    if "gmail" in lower_text:
        text = GMAIL_ADDRESS_PATTERN.sub("", text)
        lower_text = text.lower()
    mail_owned = content_is_mail_owned(relative_path, text)
    findings = []
    for name, needle, mail_only, pattern in CONTENT_PATTERNS:
        if mail_only and not mail_owned:
            continue
        if needle not in lower_text:
            continue
        if pattern.search(text):
            findings.append(name)
    return findings


def boundary_findings(root: Path) -> list:
    findings = []
    for generic_root in GENERIC_PACKAGE_ROOTS:
        absolute_root = root / generic_root
        if not absolute_root.is_dir():
            continue
        for path in sorted(absolute_root.rglob("*.go")):
            relative_path = path.relative_to(root).as_posix()
            if path.name.endswith("_test.go"):
                continue
            if relative_path in BOUNDARY_EXEMPT_PATHS:
                continue
            text = path.read_text(errors="ignore")
            for forbidden in BOUNDARY_FORBIDDEN_TOKENS:
                if forbidden in text:
                    findings.append(
                        f"{relative_path} contains Mail-owned token {forbidden!r}"
                    )
    return findings


def collect_findings(root: Path) -> list:
    findings = []
    used_allowances = {}
    for relative_path in tracked_repository_paths(root):
        if scan_excluded(relative_path):
            continue
        reason = path_finding(relative_path)
        if reason:
            findings.append(relative_path + ": filename " + reason)
        if not content_candidate(relative_path):
            continue
        path = root / relative_path
        # git ls-files includes tracked paths removed in the working tree until
        # the deletion is staged. A retired file must not make this guardrail
        # crash while the change is being reviewed.
        if not path.is_file():
            continue
        payload = path.read_text(errors="ignore")
        for pattern in content_findings(relative_path, payload):
            if pattern in ALLOWLIST.get(relative_path, {}):
                used_allowances.setdefault(relative_path, set()).add(pattern)
                continue
            findings.append(
                relative_path + ": content contains " + pattern
            )
    for path, patterns in ALLOWLIST.items():
        for pattern in patterns:
            if pattern not in used_allowances.get(path, set()):
                findings.append(
                    path + ": stale content allowlist entry for " + pattern
                )
    findings.extend(boundary_findings(root))
    return sorted(findings)


def main() -> int:
    findings = collect_findings(GUARD_ROOT)
    if findings:
        print(
            "retired Mail artifacts returned to the tracked tree "
            "or leaked into generic packages:"
        )
        for finding in findings:
            print("  " + finding)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())