"""
Stale PR Manager
"""

import base64
import gzip
import json
import os
import sys
from datetime import UTC, datetime, timedelta
from enum import Enum
from typing import Any

import requests

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
GITHUB_TOKEN = os.getenv("GH_TOKEN") or os.getenv("GITHUB_TOKEN")
if not GITHUB_TOKEN:
    print("ERROR: GITHUB_TOKEN is not set")
    sys.exit(1)

OWNER = "theorchard"
REPO = "terraform-infra"
API = f"https://api.github.com/repos/{OWNER}/{REPO}"
HEADERS = {
    "Authorization": f"token {GITHUB_TOKEN}",
    "Accept": "application/vnd.github.v3+json",
    "X-GitHub-Api-Version": "2022-11-28",
}

# Timeout (connect, read) in seconds for every outbound HTTP request. Without a
# timeout a single hung connection would block the entire daily run forever.
HTTP_TIMEOUT = (10, 30)

# Safety cap on paginated fetches (100 items/page → up to 5000 items). Prevents an
# unbounded loop if an endpoint never returns a short page.
MAX_PAGES = 50

SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
SLACK_API = "https://slack.com/api"

if not SLACK_BOT_TOKEN:
    print("WARNING: SLACK_BOT_TOKEN is not set; Slack DMs will be skipped")

ORG_MEMBERS_B64GZ = os.getenv("ORG_MEMBERS_B64GZ", "")

_members_cache: dict[str, str] | None = None


def _int_env(name: str, default: int) -> int:
    """Parse an int from an env var, falling back to default on missing/invalid."""
    raw = os.getenv(name)
    if raw is None or raw.strip() == "":
        return default
    try:
        return int(raw)
    except ValueError:
        print(f"WARNING: {name}={raw!r} is not an integer; using default {default}")
        return default


STALE_DAYS = _int_env("STALE_DAYS", 7)
COMMENT_LEASE_DAYS = 3  # days to wait before posting again on the same PR
# NOTE: the hyphen here is U+2011 (non-breaking hyphen), not an ASCII "-". This
# exact glyph is what the bot has posted in production, so it must stay byte-for-byte
# identical or last_bot_comment() will stop recognizing existing comments.
BOT_COMMENT_MARKER = "[stale‑check]"  # noqa: RUF001

DRY_RUN = os.getenv("DRY_RUN", "").lower() in ("1", "true", "yes")
TEST_PR = _int_env("TEST_PR_NUMBER", 0) or None

if DRY_RUN:
    print("*** DRY-RUN mode: no comments or Slack DMs will be sent ***")
if TEST_PR:
    print(f"*** TEST mode: only processing PR #{TEST_PR} ***")


class Labels(Enum):
    # Human-applied opt-out labels: if present, the author has intentionally parked
    # the PR and the bot stays quiet.
    DONT_MERGE = "Dont Merge"
    DO_NOT_APPLY = "do not apply"


# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
def api_get(url: str, params: dict[str, Any] | None = None) -> Any:
    """Send a GET request to the GitHub API and return the parsed JSON response."""
    resp = requests.get(url, headers=HEADERS, params=params, timeout=HTTP_TIMEOUT)
    resp.raise_for_status()
    return resp.json()


def api_post(url: str, data: dict[str, Any]) -> Any:
    """Send a POST request to the GitHub API and return the parsed JSON response."""
    resp = requests.post(url, headers=HEADERS, json=data, timeout=HTTP_TIMEOUT)
    resp.raise_for_status()
    return resp.json()


def api_get_paginated(url: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
    """Fetch all pages from a GitHub API endpoint and return the combined result list.

    Forces per_page=100 and increments the page number until a partial page is
    returned, which signals the last page. A hard page cap guards against an
    unbounded loop if an endpoint never returns a short page.
    """
    results: list[dict[str, Any]] = []
    page = 1
    base_params = dict(params or {})
    base_params["per_page"] = 100
    while page <= MAX_PAGES:
        base_params["page"] = page
        resp = requests.get(url, headers=HEADERS, params=base_params, timeout=HTTP_TIMEOUT)
        resp.raise_for_status()
        batch: list[dict[str, Any]] = resp.json()
        results.extend(batch)
        if len(batch) < 100:
            break
        page += 1
    else:
        print(f"WARNING: pagination hit MAX_PAGES={MAX_PAGES} for {url}; results may be truncated")
    return results


def last_bot_comment(pr_number: int) -> datetime | None:
    """Return the timestamp of the most recent bot comment on a PR, or None if not found.

    Identifies bot comments by the presence of BOT_COMMENT_MARKER in the body.
    All pages of comments are fetched to avoid missing older bot comments on busy PRs.
    """
    url = f"{API}/issues/{pr_number}/comments"
    comments: list[dict[str, Any]] = api_get_paginated(url)
    # Find comments containing the marker
    for c in reversed(comments):
        if BOT_COMMENT_MARKER in c.get("body", ""):
            return datetime.strptime(c["created_at"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=UTC)
    return None


def has_label(pr: dict[str, Any], label_name: str) -> bool:
    """Return True if the PR has a label matching label_name (case-insensitive)."""
    return any(lbl["name"].lower() == label_name.lower() for lbl in pr["labels"])


def post_comment(pr_number: int, body: str) -> None:
    """Post a comment on the given PR."""
    if DRY_RUN:
        print(f"[DRY-RUN] Would comment on PR #{pr_number}:\n{body}\n")
        return
    api_post(f"{API}/issues/{pr_number}/comments", {"body": body})
    print(f"COMMENTED on PR #{pr_number}")


def _load_members() -> dict[str, str]:
    """Decode ORG_MEMBERS_B64GZ env var and return {github_username: email}.

    The env var contains base64-encoded, gzipped JSON produced by
    ``base64gzip(jsonencode(local.members))`` in Terraform.
    The JSON structure is {"username": {"email": "...", ...}, ...}.
    """
    if not ORG_MEMBERS_B64GZ:
        print("WARNING: ORG_MEMBERS_B64GZ is not set; email lookup will be unavailable")
        return {}
    try:
        raw = gzip.decompress(base64.b64decode(ORG_MEMBERS_B64GZ))
        data: dict[str, dict[str, str]] = json.loads(raw)
        return {k: v["email"] for k, v in data.items() if "email" in v}
    except Exception as exc:  # noqa: BLE001 — bad secret must not crash the run; degrade to no emails
        print(f"WARNING: Could not decode ORG_MEMBERS_B64GZ: {exc}")
        return {}


def get_user_email(username: str) -> str | None:
    """Look up the email for a GitHub user from the org members secret.

    Results are cached for the lifetime of the process to avoid re-decoding
    on every call within a single run.

    Returns the email string, or None if the user is not found.
    """
    global _members_cache
    if _members_cache is None:
        _members_cache = _load_members()
    return _members_cache.get(username)


def send_slack_dm(email: str, message: str) -> bool:
    """Send a plain-text Slack DM to the user identified by email.

    Makes three sequential Slack API calls:
      1. users.lookupByEmail  → resolve email to Slack user ID
      2. conversations.open   → open (or retrieve existing) DM channel
      3. chat.postMessage     → post the message

    Returns True if the message was delivered, False on any failure.
    Failures are logged as warnings and never raise exceptions.

    Required Slack bot token scopes: users:read.email, im:write, chat:write
    """
    if DRY_RUN:
        print(f"[DRY-RUN] Would send Slack DM to {email!r}:\n{message}\n")
        return True
    if not SLACK_BOT_TOKEN:
        return False

    slack_headers = {
        "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
        "Content-Type": "application/json",
    }

    try:
        resp = requests.get(
            f"{SLACK_API}/users.lookupByEmail",
            headers=slack_headers,
            params={"email": email},
            timeout=HTTP_TIMEOUT,
        )
        resp.raise_for_status()
        body = resp.json()
        if not body.get("ok"):
            print(
                f"WARNING: Slack users.lookupByEmail failed for {email!r}: "
                f"{body.get('error', 'unknown error')}"
            )
            return False
        slack_user_id: str = body["user"]["id"]

        resp = requests.post(
            f"{SLACK_API}/conversations.open",
            headers=slack_headers,
            json={"users": slack_user_id},
            timeout=HTTP_TIMEOUT,
        )
        resp.raise_for_status()
        body = resp.json()
        if not body.get("ok"):
            print(
                f"WARNING: Slack conversations.open failed for user {slack_user_id!r}: "
                f"{body.get('error', 'unknown error')}"
            )
            return False
        channel_id: str = body["channel"]["id"]

        resp = requests.post(
            f"{SLACK_API}/chat.postMessage",
            headers=slack_headers,
            json={"channel": channel_id, "text": message},
            timeout=HTTP_TIMEOUT,
        )
        resp.raise_for_status()
        body = resp.json()
        if not body.get("ok"):
            print(
                f"WARNING: Slack chat.postMessage failed for channel {channel_id!r}: "
                f"{body.get('error', 'unknown error')}"
            )
            return False

        print(f"SLACK DM sent to {email!r}")
        return True

    except Exception as exc:  # noqa: BLE001 — Slack is best-effort, never block the GitHub comment
        print(f"WARNING: Slack DM failed for {email!r}: {exc}")
        return False


def notify(pr_number: int, author: str, body: str, slack_body: str | None = None) -> None:
    """Post a GitHub PR comment and, if configured, send a Slack DM to the author.

    GitHub comment uses the existing body format including BOT_COMMENT_MARKER.
    If slack_body is provided it is sent as-is; otherwise a plain-text version
    of body is derived by stripping the marker and emoji.
    Slack failures are best-effort and never block the GitHub comment.
    """
    post_comment(pr_number, body)

    if not SLACK_BOT_TOKEN:
        return

    email = get_user_email(author)
    if not email:
        print(f"WARNING: No email found for {author!r}; skipping Slack DM for PR #{pr_number}")
        return

    if slack_body is not None:
        send_slack_dm(email, slack_body)
    else:
        plain = body.replace(BOT_COMMENT_MARKER, "").replace("✅", "").replace("⚠️", "").strip()
        send_slack_dm(email, plain)


# ----------------------------------------------------------------------
# Main
# ----------------------------------------------------------------------
def main() -> None:
    """Scan all open PRs and send each stale one a single polite nudge.

    For every open PR that was created more than STALE_DAYS ago, post one
    comment (and best-effort Slack DM) asking the author to merge or close it,
    unless:

    - it was already nudged within COMMENT_LEASE_DAYS (spam guard), or
    - it carries a human opt-out label (`Dont Merge` / `do not apply`).

    The bot does not inspect approval, Atlantis apply history, or merge state —
    it simply reminds the author, who decides whether to merge or close. Each PR
    is processed independently; a failure on one PR is logged and does not abort
    processing of the remaining PRs.
    """
    now = datetime.now(UTC)
    stale_cutoff = now - timedelta(days=STALE_DAYS)

    # 1. List PRs to process
    if TEST_PR:
        prs = [api_get(f"{API}/pulls/{TEST_PR}")]
        print(f"TEST mode: processing PR #{TEST_PR} regardless of staleness")
    else:
        prs = api_get_paginated(f"{API}/pulls", params={"state": "open"})
        print(f"Found {len(prs)} open PR(s)")

    failures = 0
    for pr in prs:
        try:
            process_pr(pr, now, stale_cutoff)
        except Exception as exc:  # noqa: BLE001 — one bad PR must not abort the run
            failures += 1
            if isinstance(pr, dict):
                pr_number = pr.get("number", "?")
                pr_url = pr.get("html_url", "unknown")
            else:
                pr_number, pr_url = "?", "unknown"
            print(f"ERROR: failed to process PR #{pr_number} ({pr_url}): {exc!r}")

    if failures:
        print(f"Completed with {failures} PR(s) failed; see errors above.")


def process_pr(pr: dict[str, Any], now: datetime, stale_cutoff: datetime) -> None:
    """Notify the author if this PR is stale and not opted out.

    Raises on unexpected API errors so the caller can isolate the failure to
    this PR and continue processing the rest of the open PRs.
    """
    pr_number: int = pr["number"]
    author: str = pr["user"]["login"]
    pr_url: str = pr["html_url"]
    created_at: datetime = datetime.strptime(pr["created_at"], "%Y-%m-%dT%H:%M:%SZ").replace(
        tzinfo=UTC
    )
    age_days: int = max((now - created_at).days, 0)

    # Human opt-out: the author deliberately parked this PR, so stay quiet.
    if has_label(pr, Labels.DONT_MERGE.value) or has_label(pr, Labels.DO_NOT_APPLY.value):
        return

    if not TEST_PR:
        # Staleness keys off created_at, not updated_at: GitHub bumps updated_at on
        # any activity including the bot's own stale-check comment, which would
        # reset the clock every run and the PR would never read as stale again.
        if created_at > stale_cutoff:
            # not stale
            return

        # ---- spam-prevention ----
        last_comment_ts: datetime | None = last_bot_comment(pr_number)
        if last_comment_ts and (now - last_comment_ts) < timedelta(days=COMMENT_LEASE_DAYS):
            # We posted a comment less than COMMENT_LEASE_DAYS ago
            return

    body = (
        f"{BOT_COMMENT_MARKER}\n\n"
        f"Hi @{author}, pull request #{pr_number} has been open for {age_days} days. "
        "Could you check whether these changes are still needed and either merge or "
        "close it?\n\n"
        "Long-open infrastructure PRs can drift from the live state or lose their "
        "Atlantis lock, so it's best not to leave them hanging. Thanks!"
    )
    slack_body = (
        f"Hi @{author}, your pull request has been open for {age_days} days. "
        "Could you check whether these changes are still needed and either merge or "
        "close it? Long-open infrastructure PRs can drift from the live state or lose "
        f"their Atlantis lock, so it's best not to leave them hanging. Thanks! {pr_url}"
    )
    notify(pr_number, author, body, slack_body=slack_body)