"""Bound safe preflight retries without replaying an ambiguous edit creation."""
from __future__ import annotations

import json
import re
import time

from google.auth.exceptions import RefreshError, TransportError
import requests
from urllib3 import exceptions as urllib_errors

RETRY_STATUSES = {429, 500, 502, 503, 504}
TRANSPORT_ERRORS = (requests.RequestException, TransportError)


def connection_not_sent(error: Exception) -> bool:
    """Retry only typed connection failure or a transient HTTP CONNECT refusal.

    Requests explicitly declares ConnectTimeout safe to retry. Other failures must
    carry urllib3's structured reason. ProxyError alone is ambiguous: it can wrap
    a dropped origin response. A plain OSError with http.client's exact tunnel
    status framing identifies CONNECT 502/503/504 before any origin TLS/POST.
    """
    if isinstance(error, requests.exceptions.ConnectTimeout):
        return error.response is None
    if not isinstance(error, requests.exceptions.ConnectionError) or error.response is not None:
        return False
    if len(error.args) != 1:
        return False
    reason = error.args[0]
    if isinstance(reason, urllib_errors.MaxRetryError):
        reason = reason.reason
    if isinstance(reason, urllib_errors.ConnectTimeoutError):
        return True
    if not isinstance(reason, urllib_errors.ProxyError):
        return False
    original = reason.original_error
    if isinstance(original, urllib_errors.ConnectTimeoutError):
        return True
    return (type(original) is OSError and original.errno is None
            and re.fullmatch(r"Tunnel connection failed: (?:502|503|504)(?: [^\r\n]*)?", str(original)) is not None)


def transport_code(error: Exception) -> str:
    pending, seen = [error], set()
    while pending and len(seen) < 12:
        item = pending.pop(0)
        if id(item) in seen:
            continue
        seen.add(id(item))
        for kind, code in ((requests.exceptions.ProxyError, "proxy_connection_failed"),
                           (requests.exceptions.SSLError, "tls_failed"),
                           (requests.exceptions.Timeout, "timed_out"),
                           (requests.exceptions.ConnectionError, "connection_failed")):
            if isinstance(item, kind):
                return code
        pending.extend(value for value in (item.__cause__, item.__context__, *item.args)
                       if isinstance(value, Exception))
    return "transport_failed"


def fail(phase: str, code: str, attempt: int, *, status: int | None = None,
         edit_id: str | None = None) -> None:
    evidence = {"schema": "gowalk-cicd/play-preflight-failure.v1", "phase": phase,
                "code": code, "attempts": attempt, "route": "assigned_proxy"}
    if status is not None:
        evidence["status"] = status
    if edit_id is not None:
        evidence["cleanup_required"] = True
        evidence["edit_id"] = edit_id
    encoded = json.dumps(evidence, sort_keys=True)
    annotation = encoded.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A")
    print("::error title=play_preflight_failed::" + annotation)
    raise SystemExit("Google Play preflight failed: " + encoded) from None


def call(phase: str, operation, *, retry: bool = False, edit_id: str | None = None,
         accepted: tuple[int, ...] | None = None, retry_connect: bool = False):
    limit = 3 if retry or retry_connect else 1
    for attempt in range(1, limit + 1):
        try:
            result = operation()
        except TRANSPORT_ERRORS as error:
            if attempt == limit or (not retry and not connection_not_sent(error)):
                fail(phase, transport_code(error), attempt, edit_id=edit_id)
        except RefreshError:
            fail(phase, "oauth_refresh_failed", attempt)
        else:
            if not retry or getattr(result, "status_code", None) not in RETRY_STATUSES:
                if accepted is not None and result.status_code not in accepted:
                    # A transient upstream status is not a refusal, even where the
                    # phase may not replay it: `edit_create` never repeats an
                    # ambiguous insert, so a 503 there ends the build — and calling
                    # that "provider_refused" sent the next session hunting for a
                    # configuration Google never objected to (Foundy, 2026-09-09,
                    # run 34345638488). The remedy for one is to dispatch again;
                    # for the other it is to change something first.
                    code = ("provider_unavailable" if result.status_code in RETRY_STATUSES
                            else "provider_refused")
                    fail(phase, code, attempt, status=result.status_code, edit_id=edit_id)
                return result
            if attempt == limit:
                fail(phase, "provider_unavailable", attempt, status=result.status_code, edit_id=edit_id)
        print(f"Google Play preflight retry: phase={phase} attempt={attempt + 1}/{limit}")
        time.sleep(attempt)