"""Direct HTTP recipe, not an official SDK. Python >= 3.10, standard library only."""
import hashlib
import json
import math
import os
from pathlib import Path
import re
import sys
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from http.client import HTTPException
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode, urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener
from uuid import uuid4


class RecipeError(Exception):
    pass


class NetworkError(RecipeError):
    pass


class ApiError(RecipeError):
    def __init__(self, status, code, method, path, retry_after=None):
        self.status, self.code = status, code
        delay = f" Retry after at least {math.ceil(retry_after)} seconds." if retry_after is not None else ""
        super().__init__(f"HTTP {status} {code} ({method} {path}). Check permissions, operation history and the API error guide." + delay)


def is_uuid(value):
    return isinstance(value, str) and re.fullmatch(r"[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", value) is not None


def code_of(reply):
    body = reply.get("body") or {}
    error = body.get("error") if isinstance(body, dict) else None
    code = error.get("code", "") if isinstance(error, dict) else ""
    return code if isinstance(code, str) and re.fullmatch(r"[a-z_]{1,80}", code) else "http_error"


def summary(body):
    return {key: body[key] for key in ("id", "status", "counts") if key in body}


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None  # Never forward a bearer key to a redirect target.


def http_transport(*, url, method, headers, body=None):
    request = Request(url, data=body.encode("utf-8") if body is not None else None, headers=headers, method=method)
    try:
        try:
            response = build_opener(NoRedirect()).open(request, timeout=30)
        except HTTPError as error:
            response = error  # HTTP errors still carry useful status and JSON.
        with response:
            content = response.read()
            try:
                data = json.loads(content)
            except (ValueError, UnicodeError):
                data = None
            return {"status": response.code, "body": data,
                    "headers": {key.lower(): value for key, value in response.headers.items()}}
    except (URLError, OSError, TimeoutError, EOFError, HTTPException):
        raise NetworkError("No readable HTTP response. The outcome may be uncertain.") from None


def body_for(resource, payload):
    allowed = {"to", "text", "sender_name", "send_at"} if resource == "messages" else (
        {"recipients", "text", "sender_name", "send_at"} if resource == "message-batches" else set())
    if not allowed or not isinstance(payload, dict) or not set(payload).issubset(allowed):
        raise RecipeError("Use messages or message-batches with only their documented request fields.")
    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)


class SmsClient:
    def __init__(self, *, api_key, organization, journal_directory,
                 base_url="http://127.0.0.1:8000/api/sms/v2/", transport=http_transport, wait=time.sleep):
        base = urlsplit(base_url)
        if (base.username or base.password or base.query or base.fragment or base.path != "/api/sms/v2/"
                or not (base.scheme == "https" or (base.scheme == "http" and base.hostname in {"127.0.0.1", "localhost", "::1"}))):
            raise RecipeError("Use an HTTPS API base ending /api/sms/v2/ (HTTP is allowed only on loopback).")
        if not api_key or not api_key.startswith("bf_sms_") or not organization or not journal_directory:
            raise RecipeError("Set BEFIANA_SMS_API_KEY, BEFIANA_SMS_ORGANIZATION and BEFIANA_SMS_JOURNAL.")
        self.api_key, self.organization, self.base_url = api_key, organization, base_url
        self.journal_directory = Path(journal_directory).resolve()
        self.transport, self.wait = transport, wait

    def request(self, method, path, body=None, key=None):
        headers = {"Authorization": f"Bearer {self.api_key}", "Accept": "application/json"}
        if body is not None:
            headers["Content-Type"] = "application/json"
        if key:
            headers["Idempotency-Key"] = key
        return self.transport(url=self.base_url + path, method=method, headers=headers, body=body)

    def save(self, path, record):
        temporary = path.with_name(path.name + f".{uuid4()}.tmp")
        fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as stream:
            json.dump(record, stream, ensure_ascii=False)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
        if os.name != "nt":
            directory_fd = os.open(self.journal_directory, os.O_RDONLY)
            try:
                os.fsync(directory_fd)
            finally:
                os.close(directory_fd)

    def submit(self, operation, resource, payload, *, max_attempts=3):
        if not isinstance(operation, str) or not operation.strip() or len(operation) > 200 or type(max_attempts) is not int or not 1 <= max_attempts <= 3:
            raise RecipeError("Invalid operation or attempt limit (1 to 3).")
        body = body_for(resource, payload)
        self.journal_directory.mkdir(parents=True, exist_ok=True, mode=0o700)
        path = self.journal_directory / (hashlib.sha256(operation.encode("utf-8")).hexdigest() + ".json")
        lock_path = path.with_name(path.name + ".lock")
        try:
            lock_fd = os.open(lock_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        except FileExistsError:
            raise RecipeError("Operation locked. Manual review of the outcome and private journal is required, even after previous processes stop.") from None
        release_lock = True

        def persist(record):
            nonlocal release_lock
            # A failed write may leave stale pending JSON after the server accepted the SMS.
            release_lock = False
            try:
                self.save(path, record)
            except Exception:
                raise RecipeError("Journal write failed. Operation remains locked. Review the outcome and private journal before any manual recovery.") from None
            release_lock = True

        try:
            try:
                record = json.loads(path.read_text(encoding="utf-8"))
            except FileNotFoundError:
                record = {"version": 1, "operation": operation, "organization": self.organization,
                          "base_url": self.base_url, "resource": resource, "body_json": body,
                          "key": str(uuid4()), "state": "pending", "attempts": 0}
                persist(record)  # Durable operation + key + exact body BEFORE any network call.
            except (ValueError, OSError):
                raise RecipeError("Unreadable operation journal. Restore it; never silently create a new key.") from None
            if not isinstance(record, dict):
                raise RecipeError("Unreadable operation journal. Restore it; never silently create a new key.")
            if (record.get("version") != 1 or not is_uuid(record.get("key")) or record.get("operation") != operation
                    or record.get("organization") != self.organization or record.get("base_url") != self.base_url
                    or record.get("resource") != resource or record.get("body_json") != body):
                raise RecipeError("Operation conflicts with its saved scope or payload. Do not change or delete its key.")
            if record["state"] == "follow_up":
                return self.follow_up(record)
            if record["state"] == "complete":
                return record["result"]
            if record["state"] != "pending":
                raise ApiError(record.get("http_status"), record.get("error_code"), "POST", resource + "/")
            for attempt in range(max_attempts):
                record["attempts"] += 1
                persist(record)
                try:
                    reply = self.request("POST", resource + "/", record["body_json"], record["key"])
                except NetworkError:
                    if attempt + 1 < max_attempts:
                        self.wait(2 ** attempt)
                        continue
                    raise NetworkError("No final response. Resume this SAME operation, journal and payload later.") from None
                data, code = reply.get("body") or {}, code_of(reply)
                if isinstance(data, dict) and is_uuid(data.get("message_id")):
                    record.update(state="follow_up", message_id=data["message_id"], http_status=reply["status"], error_code=code)
                    persist(record)
                    return self.follow_up(record)  # GET only, including on every later invocation.
                if code == "provider_result_unknown" or reply["status"] in (502, 504):
                    record.update(state="review_required", http_status=reply["status"], error_code=code)
                    persist(record)
                    raise ApiError(reply["status"], code, "POST", resource + "/")
                if reply["status"] in (201, 202) and isinstance(data, dict) and is_uuid(data.get("id")):
                    record.update(state="complete", result={"state": "complete", **summary(data)})
                    persist(record)
                    return record["result"]
                retry = reply["status"] in (429, 500, 503) or (reply["status"] == 409 and code == "idempotency_in_progress")
                if not retry:
                    record.update(state="review_required", http_status=reply["status"], error_code=code)
                    persist(record)
                    raise ApiError(reply["status"], code, "POST", resource + "/")
                header = reply.get("headers", {}).get("retry-after")
                try:
                    seconds = 2 ** attempt if header is None else (int(header) if header.isdigit() else
                        max(0, (parsedate_to_datetime(header) - datetime.now(timezone.utc)).total_seconds()))
                except (ValueError, TypeError, AttributeError):
                    seconds = 61
                if seconds > 60 or attempt + 1 == max_attempts:
                    raise ApiError(reply["status"], code, "POST", resource + "/", seconds)
                self.wait(seconds)
        finally:
            os.close(lock_fd)
            if release_lock:
                lock_path.unlink()

    def follow_up(self, record):
        return {"state": "follow_up", "message_id": record["message_id"], "error_code": record["error_code"],
                "message": summary(self.message_status(record["message_id"]))}

    def get(self, path):
        reply = self.request("GET", path)
        if reply["status"] != 200:
            raise ApiError(reply["status"], code_of(reply), "GET", path.split("?")[0])
        if not isinstance(reply.get("body"), dict):
            raise RecipeError("Invalid JSON status response. No SMS was resubmitted.")
        return reply["body"]

    def message_status(self, identifier):
        if not is_uuid(identifier):
            raise RecipeError("Expected a message UUID.")
        return self.get(f"messages/{quote(identifier, safe='')}/")

    def batch_messages(self, identifier):
        if not is_uuid(identifier):
            raise RecipeError("Expected a batch UUID.")
        results, seen, cursor = [], set(), None
        for _ in range(1000):
            query = "" if cursor is None else "?" + urlencode({"cursor": cursor})
            data = self.get(f"message-batches/{quote(identifier, safe='')}/messages/{query}")
            cursor = data.get("next_cursor")
            if not isinstance(data.get("results"), list) or "next_cursor" not in data or not (cursor is None or isinstance(cursor, str) and cursor):
                raise RecipeError("Invalid batch page. Results are incomplete.")
            results.extend(data["results"])
            if cursor is None:
                return results
            if cursor in seen:
                raise RecipeError("Repeated cursor. Results are incomplete.")
            seen.add(cursor)
        raise RecipeError("Page safety limit reached. Results are incomplete.")


def main():
    args = sys.argv[1:]
    client = SmsClient(api_key=os.environ.get("BEFIANA_SMS_API_KEY"), organization=os.environ.get("BEFIANA_SMS_ORGANIZATION"),
                       journal_directory=os.environ.get("BEFIANA_SMS_JOURNAL"),
                       base_url=os.environ.get("BEFIANA_SMS_BASE_URL", "http://127.0.0.1:8000/api/sms/v2/"))
    if len(args) == 4 and args[0] == "submit":
        result = client.submit(args[1], args[2], json.loads(Path(args[3]).read_text(encoding="utf-8")))
    elif len(args) == 2 and args[0] == "status":
        result = summary(client.message_status(args[1]))
    elif len(args) == 2 and args[0] == "batch-messages":
        result = [summary(row) for row in client.batch_messages(args[1])]
    else:
        raise RecipeError("Usage: client.py submit OPERATION messages|message-batches PAYLOAD.json | status MESSAGE_UUID | batch-messages BATCH_UUID")
    print(json.dumps(result, indent=2))  # No token, recipient, text or raw error body.


if __name__ == "__main__":
    try:
        main()
    except RecipeError as error:
        print(str(error), file=sys.stderr)
        sys.exit(1)
    except Exception:
        print("Local validation/journal error. Check inputs and the private journal; no automatic reset is performed.", file=sys.stderr)
        sys.exit(1)