#!/usr/bin/env python3
"""Read-only by default. Reuse/prepare the fixed Beijing host via the aliyun CLI."""

import argparse
import base64
import fcntl
import hashlib
import json
import math
import os
import shutil
import subprocess
import sys
import time
import uuid
from pathlib import Path

REGION = "cn-beijing"
PUBLIC_IP = "123.56.71.98"


class CloudError(RuntimeError):
    pass


class AliyunCLI:
    def __init__(self, profile=None):
        self.binary = shutil.which("aliyun")
        if not self.binary:
            raise CloudError(
                "aliyun CLI is not installed; configure CLI credentials locally, never paste secrets into chat"
            )
        self.profile = profile

    def __call__(self, product, action, **params):
        cmd = [self.binary, product, action, "--RegionId", REGION]
        if self.profile:
            cmd += ["--profile", self.profile]
        for key, value in params.items():
            cmd += [
                "--" + key,
                json.dumps(value, separators=(",", ":"))
                if isinstance(value, (list, dict, bool))
                else str(value),
            ]
        try:
            result = subprocess.run(
                cmd, capture_output=True, text=True, timeout=60, check=True
            )
            data = json.loads(result.stdout)
            if not isinstance(data, dict):
                raise CloudError(
                    f"{product} {action} returned an invalid response shape"
                )
            return data
        except (
            subprocess.CalledProcessError,
            subprocess.TimeoutExpired,
            json.JSONDecodeError,
        ) as exc:
            # CLI diagnostics can contain request details; don't log raw stdout/stderr or credentials.
            raise CloudError(
                f"{product} {action} failed; inspect the authenticated CLI locally (no mutation retry here)"
            ) from exc


def identity(cfg):
    if cfg.get("region") != REGION or cfg.get("public_ip") != PUBLIC_IP:
        raise CloudError("region/IP must remain cn-beijing / 123.56.71.98")
    if (
        not isinstance(cfg.get("project_tag"), str)
        or not cfg["project_tag"]
        or len(cfg["project_tag"]) > 64
    ):
        raise CloudError("a dedicated project_tag is required")


def parse_count(value):
    try:
        count = int(value)
    except (TypeError, ValueError) as exc:
        raise CloudError("invalid inventory count; no mutations allowed") from exc
    if isinstance(value, bool) or count < 0:
        raise CloudError("invalid inventory count; no mutations allowed")
    return count


def instances(api, **filters):
    found = []
    seen = set()
    for page in range(1, 21):
        data = api("ecs", "DescribeInstances", PageSize=100, PageNumber=page, **filters)
        rows = data.get("Instances", {}).get("Instance")
        if not isinstance(rows, list):
            raise CloudError("malformed instance inventory; no mutations allowed")
        for row in rows:
            key = row.get("InstanceId") if isinstance(row, dict) else None
            if not isinstance(key, str) or not key or key in seen:
                raise CloudError(
                    "invalid or changing instance inventory; no mutations allowed"
                )
            seen.add(key)
        found.extend(rows)
        total = data.get("TotalCount")
        if total is not None:
            expected = parse_count(total)
            if len(found) == expected:
                return found
            if len(found) > expected or not rows:
                raise CloudError(
                    "incomplete or changing instance inventory; no mutations allowed"
                )
        elif len(rows) < 100:
            return found
    raise CloudError(
        "instance inventory exceeded the safety bound; no mutations allowed"
    )


def eip(api):
    data = api(
        "vpc", "DescribeEipAddresses", EipAddress=PUBLIC_IP, PageSize=100, PageNumber=1
    )
    rows = data.get("EipAddresses", {}).get("EipAddress")
    if not isinstance(rows, list):
        raise CloudError("malformed EIP inventory")
    if parse_count(data.get("TotalCount", len(rows))) != len(rows):
        raise CloudError("incomplete EIP inventory")
    matches = [row for row in rows if row.get("IpAddress") == PUBLIC_IP]
    if len(matches) > 1:
        raise CloudError("ambiguous EIP ownership")
    if matches and not matches[0].get("AllocationId"):
        raise CloudError("EIP allocation ID missing; no creation allowed")
    return matches[0] if matches else None


def owned(row, cfg):
    tags = row.get("Tags", {}).get("Tag", [])
    return row.get("InstanceId") == cfg.get("adopt_instance_id") or any(
        t.get("TagKey") == "project" and t.get("TagValue") == cfg["project_tag"]
        for t in tags
    )


def plan(cfg, api):
    identity(cfg)
    address = eip(api)
    if address and address.get("InstanceId"):
        if address.get("InstanceType") != "EcsInstance":
            raise CloudError(
                "fixed EIP is attached to a non-ECS resource; refusing to rebind"
            )
        rows = instances(api, InstanceIds=[address["InstanceId"]])
    elif address:
        if address.get("Status") != "Available":
            raise CloudError(
                "EIP is not available; wait/reconcile instead of creating a host"
            )
        rows = instances(
            api, **{"Tag.1.Key": "project", "Tag.1.Value": cfg["project_tag"]}
        )
    else:
        rows = instances(api, PublicIpAddresses=[PUBLIC_IP])
        rows = [
            r
            for r in rows
            if PUBLIC_IP in r.get("PublicIpAddress", {}).get("IpAddress", [])
        ]
        if not rows:
            raise CloudError(
                "fixed IP is neither an owned EIP nor an existing ECS public IP; cannot recreate this address"
            )
    if len(rows) > 1:
        raise CloudError(
            "multiple candidate instances; explicit reconciliation required"
        )
    if address and address.get("InstanceId") and not rows:
        raise CloudError(
            "EIP attachment has no visible instance; do not create or detach"
        )
    row = rows[0] if rows else None
    if row and not owned(row, cfg):
        raise CloudError(
            "existing instance is not project-owned; set adopt_instance_id only after owner approval"
        )
    return {
        "region": REGION,
        "public_ip": PUBLIC_IP,
        "action": "reuse"
        if row and (not address or address.get("InstanceId"))
        else ("bind" if row else "create_and_bind"),
        "instance_id": row.get("InstanceId") if row else None,
        "instance_status": row.get("Status") if row else None,
        "instance_charge_type": row.get("InstanceChargeType") if row else None,
        "spot_strategy": row.get("SpotStrategy") if row else None,
        "allocation_id": address.get("AllocationId") if address else None,
        "ip_kind": "eip" if address else "instance_public_ip",
        "warning": "spot interruption can terminate calls; fixed instance public IP is not recoverable like an EIP",
    }


def create_params(cfg):
    required = (
        "image_id",
        "instance_type",
        "vswitch_id",
        "security_group_id",
        "key_pair_name",
    )
    if any(
        not isinstance(cfg.get(k), str)
        or not cfg[k].strip()
        or cfg[k].startswith(("CHANGE_ME", "--"))
        for k in required
    ):
        raise CloudError(
            "creation requires approved image/type/VSwitch/security group/SSH KeyPair"
        )
    spot_strategy = cfg.get("spot_strategy", "SpotWithPriceLimit")
    if spot_strategy not in ("SpotWithPriceLimit", "SpotAsPriceGo"):
        raise CloudError("spot_strategy must be SpotWithPriceLimit or SpotAsPriceGo")
    price = cfg.get("spot_price_limit")
    if spot_strategy == "SpotWithPriceLimit" and (
        isinstance(price, bool)
        or not isinstance(price, (int, float))
        or not math.isfinite(price)
        or price <= 0
    ):
        raise CloudError(
            "a positive, finite spot_price_limit is required for SpotWithPriceLimit"
        )
    disk = cfg.get("system_disk_gib", 40)
    if isinstance(disk, bool) or not isinstance(disk, int) or not 40 <= disk <= 200:
        raise CloudError(
            "system_disk_gib must be within the approved 40–200 GiB safety bound"
        )
    params = {
        "ImageId": cfg["image_id"],
        "InstanceType": cfg["instance_type"],
        "VSwitchId": cfg["vswitch_id"],
        "SecurityGroupId": cfg["security_group_id"],
        "KeyPairName": cfg["key_pair_name"],
        "Amount": 1,
        "InstanceName": cfg["project_tag"],
        "InstanceChargeType": "PostPaid",
        "InternetMaxBandwidthOut": 0,
        "SpotStrategy": spot_strategy,
        "SystemDisk.Category": cfg.get("system_disk_category", "cloud_essd"),
        "SystemDisk.Size": disk,
        "SystemDisk.PerformanceLevel": cfg.get("system_disk_performance_level", "PL1"),
        "Tag.1.Key": "project",
        "Tag.1.Value": cfg["project_tag"],
    }
    if spot_strategy == "SpotWithPriceLimit":
        params["SpotPriceLimit"] = price
    user_data_file = cfg.get("user_data_file")
    if user_data_file is not None:
        if not isinstance(user_data_file, str) or not user_data_file.strip():
            raise CloudError("user_data_file must be a non-empty local file path")
        path = Path(user_data_file)
        if path.is_symlink() or not path.is_file():
            raise CloudError("user_data_file must be an existing non-symlink file")
        try:
            content = path.read_bytes()
        except OSError as exc:
            raise CloudError("user_data_file cannot be read") from exc
        if len(content) > 16 * 1024:
            raise CloudError("user_data_file exceeds the ECS 16 KiB limit")
        params["UserData"] = base64.b64encode(content).decode("ascii")
    return params


def save_state(path, data):
    temporary = path.with_suffix(".tmp")
    fd = os.open(
        temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600
    )
    with os.fdopen(fd, "w") as file:
        json.dump(data, file)
        file.flush()
        os.fsync(file.fileno())
    temporary.replace(path)
    directory_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(directory_fd)
    finally:
        os.close(directory_fd)


def wait_running(api, instance_id, sleep=time.sleep):
    for _ in range(60):
        rows = instances(api, InstanceIds=[instance_id])
        if len(rows) == 1 and rows[0].get("Status") == "Running":
            return
        if rows and rows[0].get("Status") in ("Stopped", "Stopping"):
            raise CloudError(
                f"instance {instance_id} is stopped; refusing to start or replace it automatically"
            )
        sleep(5)
    raise CloudError(
        f"instance {instance_id} is not Running; preserve state and retry/reconcile, do not create manually"
    )


def apply(cfg, api, state_path, sleep=time.sleep):
    state_path = Path(state_path)
    state_path.parent.mkdir(parents=True, exist_ok=True)
    if state_path.is_symlink():
        raise CloudError("state file must not be a symlink")
    lock_fd = os.open(
        str(state_path) + ".lock", os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW, 0o600
    )
    with os.fdopen(lock_fd, "w") as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError as exc:
            raise CloudError(
                "another provisioning process owns this state file"
            ) from exc
        current = plan(cfg, api)  # Re-read immediately before mutations.
        instance_id = current["instance_id"]
        if current["action"] == "reuse":
            if current["instance_status"] != "Running":
                raise CloudError(
                    "existing host is not Running; no automatic start/replace"
                )
            return current
        if current["action"] == "create_and_bind":
            params = create_params(cfg)
            fingerprint = hashlib.sha256(
                json.dumps([REGION, PUBLIC_IP, params], sort_keys=True).encode()
            ).hexdigest()
            try:
                state = (
                    json.loads(state_path.read_text()) if state_path.exists() else {}
                )
            except (OSError, ValueError) as exc:
                raise CloudError(
                    "state file cannot be read; no creation allowed"
                ) from exc
            if not isinstance(state, dict):
                raise CloudError("invalid state; no creation allowed")
            if state and state.get("fingerprint") != fingerprint:
                raise CloudError(
                    "creation configuration changed; reconcile old resources/state before proceeding"
                )
            if not state:
                state = {"fingerprint": fingerprint, "client_token": str(uuid.uuid4())}
                save_state(state_path, state)  # Persist BEFORE the billable request.
            try:
                uuid.UUID(state.get("client_token", ""))
            except (ValueError, TypeError, AttributeError) as exc:
                raise CloudError(
                    "invalid persisted ClientToken; reconcile state before creation"
                ) from exc
            result = api(
                "ecs", "RunInstances", ClientToken=state["client_token"], **params
            )
            ids = result.get("InstanceIdSets", {}).get("InstanceIdSet", [])
            if len(ids) != 1:
                raise CloudError(
                    "unexpected creation result; preserve ClientToken and reconcile"
                )
            instance_id = ids[0]
            state["instance_id"] = instance_id
            save_state(state_path, state)
        wait_running(api, instance_id, sleep)
        latest = eip(api)
        if not latest or latest.get("AllocationId") != current["allocation_id"]:
            raise CloudError(
                "fixed EIP changed/disappeared; created host preserved, no substitute IP"
            )
        if latest.get("InstanceId") and latest["InstanceId"] != instance_id:
            raise CloudError(
                "EIP became attached to another instance; refusing to detach it"
            )
        if not latest.get("InstanceId"):
            if latest.get("Status") != "Available":
                raise CloudError(
                    "EIP is not safely bindable; preserve instance and retry later"
                )
            api(
                "vpc",
                "AssociateEipAddress",
                AllocationId=latest["AllocationId"],
                InstanceId=instance_id,
                InstanceType="EcsInstance",
            )
        for _ in range(30):
            bound = eip(api)
            if (
                bound
                and bound.get("InstanceId") == instance_id
                and bound.get("Status") == "InUse"
            ):
                return dict(
                    current,
                    action="ready",
                    instance_id=instance_id,
                    instance_status="Running",
                )
            sleep(2)
        raise CloudError(
            "EIP binding not confirmed; preserve host/state and reconcile, do not allocate another address"
        )


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", required=True)
    parser.add_argument("--state", default=".local/aliyun-host.json")
    parser.add_argument(
        "--apply",
        action="store_true",
        help="owner-approved billable creation/EIP association",
    )
    args = parser.parse_args()
    try:
        cfg = json.loads(Path(args.config).read_text())
        identity(cfg)
        api = AliyunCLI(cfg.get("profile"))
        result = apply(cfg, api, args.state) if args.apply else plan(cfg, api)
        print(json.dumps(result, ensure_ascii=False, indent=2))
    except (CloudError, OSError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())