"""SSH / SFTP production deploy (D8).

Production is the one path in the product that a human must trigger, and the one
that can destroy a running site. Three rules follow from that and shape
everything here.

**Nothing is uploaded into the live directory.** A release lands in
``<remote_path>/releases/<timestamp>-<sha7>/`` and is switched in by moving a
``current`` symlink. That makes the switch atomic (``ln -sfn`` + ``mv`` on the
same filesystem) and makes rollback a pointer move rather than a re-upload.

**The host key is pinned on first success (TOFU).** The first ``test_connection``
that succeeds records the fingerprint; every later connection must match it. A
changed key is a hard failure, not a warning — it is either a rebuilt server (in
which case the operator re-tests deliberately) or an interception.

**Credentials never touch the target's disk.** The password or key is held in
memory for the connection only; the deploy scripts that run on the server are
built from a fixed template and never interpolate a secret.

The transport is injectable so the deploy logic — release numbering, pruning,
the symlink dance, rollback — is testable without a live sshd.
"""

from __future__ import annotations

import contextlib
import logging
import os
import shlex
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any, Protocol, runtime_checkable

from app.application.ports.deploy_target import DeployCheck, DeployTargetSpec
from app.core.redact import redact_text
from app.domain.platform.rules import assert_remote_host_allowed

logger = logging.getLogger("agy.deploy.ssh")

CONNECT_TIMEOUT_SECONDS = 15.0
COMMAND_TIMEOUT_SECONDS = 600.0
PROBE_TIMEOUT_SECONDS = 20.0

#: How many releases to keep on the target. Five is enough to roll back past a
#: bad deploy and a bad deploy of its fix, without filling the operator's disk.
KEEP_RELEASES = 5

#: Name of the pointer the web server is configured to serve from.
CURRENT_LINK = "current"


@dataclass(slots=True)
class CommandOutcome:
    exit_code: int
    output: str = ""
    timed_out: bool = False

    @property
    def ok(self) -> bool:
        return self.exit_code == 0 and not self.timed_out


@runtime_checkable
class SshTransport(Protocol):
    """The remote operations a deploy needs. Implemented over asyncssh."""

    async def connect(
        self,
        *,
        host: str,
        port: int,
        username: str,
        password: str | None,
        key: str | None,
        known_hosts: str | None,
    ) -> None: ...

    def host_key_fingerprint(self) -> str: ...

    async def run(
        self, command: str, *, timeout: float = COMMAND_TIMEOUT_SECONDS
    ) -> CommandOutcome: ...

    async def upload(self, *, local_path: str, remote_path: str) -> None: ...

    async def close(self) -> None: ...


TransportFactory = Callable[[], SshTransport]


class AsyncSshTransport:
    """The real transport: one asyncssh connection per deploy."""

    def __init__(self) -> None:
        # asyncssh is imported lazily inside connect(); the attribute is typed
        # loosely so this module imports without the dependency at hand.
        self._conn: Any = None
        self._fingerprint = ""

    async def connect(
        self,
        *,
        host: str,
        port: int,
        username: str,
        password: str | None,
        key: str | None,
        known_hosts: str | None,
    ) -> None:
        import asyncssh

        # `known_hosts=None` disables verification for this connection; the
        # fingerprint is captured and compared by the caller, which is where the
        # TOFU policy lives. Passing an explicit `known_hosts` string enforces
        # the pinned key inside asyncssh as well.
        kwargs: dict = {
            "host": host,
            "port": port,
            "username": username,
            "connect_timeout": CONNECT_TIMEOUT_SECONDS,
            "known_hosts": known_hosts if known_hosts else None,
        }
        if key:
            kwargs["client_keys"] = [asyncssh.import_private_key(key)]
            kwargs["password"] = None
        else:
            kwargs["password"] = password or ""
        self._conn = await asyncssh.connect(**kwargs)
        self._fingerprint = self._read_fingerprint()

    def _read_fingerprint(self) -> str:
        """Best-effort host key digest of the connection we just made."""
        try:
            key = self._conn.get_server_host_key()
            if key is None:
                return ""
            digest = key.get_fingerprint("sha256")
            return f"SHA256:{digest}" if not str(digest).startswith("SHA256:") else str(digest)
        except Exception:  # noqa: BLE001 - a missing fingerprint must not block a deploy
            return ""

    def host_key_fingerprint(self) -> str:
        return self._fingerprint

    async def run(
        self, command: str, *, timeout: float = COMMAND_TIMEOUT_SECONDS
    ) -> CommandOutcome:
        import asyncssh

        assert self._conn is not None, "connect() first"
        try:
            result = await self._conn.run(command, check=False, timeout=timeout)
        except (asyncssh.Error, TimeoutError) as exc:
            return CommandOutcome(exit_code=1, output=redact_text(str(exc))[:2000], timed_out=True)
        output = f"{result.stdout or ''}{result.stderr or ''}"
        return CommandOutcome(exit_code=int(result.exit_status or 0), output=output)

    async def upload(self, *, local_path: str, remote_path: str) -> None:
        assert self._conn is not None, "connect() first"
        async with self._conn.start_sftp_client() as sftp:
            await sftp.put(local_path, remote_path)

    async def close(self) -> None:
        if self._conn is not None:
            self._conn.close()
            with contextlib.suppress(Exception):
                await self._conn.wait_closed()
            self._conn = None


class SshDeployer:
    """``DeployTargetTester`` and the production deploy itself, over SSH."""

    def __init__(
        self,
        *,
        transport_factory: TransportFactory = AsyncSshTransport,
        artifacts_root: str | None = None,
        allow_private_hosts: bool = False,
        clock: Callable[[], float] = time.time,
    ) -> None:
        from app.core.config import settings

        self._factory = transport_factory
        self.artifacts_root = artifacts_root or settings.artifacts_root
        self.allow_private_hosts = allow_private_hosts
        self._clock = clock

    # -- testing -------------------------------------------------------------

    async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck:
        """Connect, pin the host key, and confirm the path is writable."""
        try:
            assert_remote_host_allowed(
                f"ssh://{target.host}:{target.port}", allow_private=self.allow_private_hosts
            )
        except Exception as exc:  # noqa: BLE001 - translated to a plain message
            return DeployCheck(ok=False, message=str(exc))

        transport = self._factory()
        try:
            await transport.connect(
                host=target.host,
                port=target.port,
                username=target.username,
                password=target.secret if target.auth_type == "password" else None,
                key=target.secret if target.auth_type == "key" else None,
                known_hosts=None,
            )
        except Exception as exc:  # noqa: BLE001 - never surface the raw error
            return DeployCheck(ok=False, message=_ssh_error_message(exc))
        finally:
            with contextlib.suppress(Exception):
                await transport.close()

        fingerprint = transport.host_key_fingerprint()
        if target.known_fingerprint and fingerprint and fingerprint != target.known_fingerprint:
            # A changed host key is a hard failure: it is either a rebuilt server
            # or an interception, and only a human can tell the two apart.
            return DeployCheck(
                ok=False,
                message="کلید میزبان تغییر کرده است؛ برای ادامه باید دوباره تأیید کنید.",
                fingerprint=fingerprint,
            )

        probe = await _probe_writable(transport, target.remote_path)
        if not probe.ok:
            return DeployCheck(
                ok=False,
                message=f"مسیر «{target.remote_path}» قابل نوشتن نیست.",
                fingerprint=fingerprint,
            )
        return DeployCheck(
            ok=True, message="اتصال و مسیر مقصد تأیید شد.", fingerprint=fingerprint
        )

    # -- deploy --------------------------------------------------------------

    async def deploy(
        self,
        *,
        target: DeployTargetSpec,
        artifact_path: str,
        sha: str,
        post_deploy_command: str | None = None,
        on_output: Callable[[str], Awaitable[None]] | None = None,
    ) -> tuple[bool, str, str]:
        """Upload, extract, run the post-deploy command, switch ``current``.

        Returns ``(ok, release_dir, message)``. On failure the ``current``
        symlink is untouched, so a failed deploy leaves the previous release
        serving — the same property the demo pipeline has.
        """
        if not os.path.isfile(artifact_path):
            return False, "", "فایل بستهٔ انتشار پیدا نشد."

        transport = self._factory()
        try:
            await transport.connect(
                host=target.host,
                port=target.port,
                username=target.username,
                password=target.secret if target.auth_type == "password" else None,
                key=target.secret if target.auth_type == "key" else None,
                known_hosts=None,
            )
        except Exception as exc:  # noqa: BLE001
            return False, "", _ssh_error_message(exc)

        try:
            fingerprint = transport.host_key_fingerprint()
            if target.known_fingerprint and fingerprint and fingerprint != target.known_fingerprint:
                return False, "", "کلید میزبان تغییر کرده است."

            release = _release_name(self._clock(), sha)
            release_dir = f"{target.remote_path.rstrip('/')}/releases/{release}"

            setup = await transport.run(
                f"mkdir -p {shlex.quote(release_dir)} && "
                f"mkdir -p {shlex.quote(target.remote_path.rstrip('/') + '/releases')}",
                timeout=PROBE_TIMEOUT_SECONDS,
            )
            if not setup.ok:
                return False, "", "ساخت پوشهٔ انتشار در سرور مقصد ناموفق بود."

            archive = f"{release_dir}/release.tar.gz"
            await transport.upload(local_path=artifact_path, remote_path=archive)
            await _stream(on_output, f"[deploy] بسته در {archive} بارگذاری شد.\n")

            extract = await transport.run(
                f"cd {shlex.quote(release_dir)} && tar -xzf release.tar.gz && rm -f release.tar.gz",
                timeout=COMMAND_TIMEOUT_SECONDS,
            )
            if on_output is not None:
                await on_output(redact_text(extract.output))
            if not extract.ok:
                return False, release_dir, "استخراج بسته در سرور مقصد ناموفق بود."

            if post_deploy_command:
                post = await transport.run(
                    f"cd {shlex.quote(release_dir)} && {post_deploy_command}",
                    timeout=COMMAND_TIMEOUT_SECONDS,
                )
                if on_output is not None:
                    await on_output(redact_text(post.output))
                if not post.ok:
                    # The previous release keeps serving: `current` never moved.
                    return False, release_dir, "اجرای دستور پس از استقرار ناموفق بود."

            switch = await transport.run(
                _switch_script(target.remote_path, release),
                timeout=PROBE_TIMEOUT_SECONDS,
            )
            if not switch.ok:
                return False, release_dir, "تغییر نسخهٔ فعال ناموفق بود."

            prune = await transport.run(
                _prune_script(target.remote_path, KEEP_RELEASES),
                timeout=PROBE_TIMEOUT_SECONDS,
            )
            if not prune.ok:
                # Pruning is housekeeping; a full disk is not a failed deploy.
                logger.warning("ssh deploy: pruning old releases failed: %s", prune.output[:200])

            return True, release_dir, "استقرار در production انجام شد."
        except Exception as exc:  # noqa: BLE001 - never leak the raw error
            logger.exception("ssh deploy failed")
            return False, "", _ssh_error_message(exc)
        finally:
            with contextlib.suppress(Exception):
                await transport.close()

    async def rollback(
        self,
        *,
        target: DeployTargetSpec,
        to_release: str | None = None,
        post_deploy_command: str | None = None,
        on_output: Callable[[str], Awaitable[None]] | None = None,
    ) -> tuple[bool, str, str]:
        """Re-point ``current`` at the previous release and re-run the hook."""
        transport = self._factory()
        try:
            await transport.connect(
                host=target.host,
                port=target.port,
                username=target.username,
                password=target.secret if target.auth_type == "password" else None,
                key=target.secret if target.auth_type == "key" else None,
                known_hosts=None,
            )
        except Exception as exc:  # noqa: BLE001
            return False, "", _ssh_error_message(exc)

        try:
            releases = await transport.run(
                _list_releases_script(target.remote_path), timeout=PROBE_TIMEOUT_SECONDS
            )
            if not releases.ok:
                return False, "", "خواندن فهرست نسخه‌ها ناموفق بود."
            names = [line.strip() for line in releases.output.splitlines() if line.strip()]
            if len(names) < 2:
                return False, "", "نسخهٔ قبلی‌ای برای بازگردانی وجود ندارد."

            # `ls -1t` is newest first, so the previous release is the second.
            current = names[0]
            target_name = to_release or names[1]
            if target_name == current:
                target_name = names[1]

            switch = await transport.run(
                _switch_script(target.remote_path, target_name), timeout=PROBE_TIMEOUT_SECONDS
            )
            if not switch.ok:
                return False, "", "بازگردانی ناموفق بود."

            if post_deploy_command:
                post = await transport.run(
                    f"cd {shlex.quote(target.remote_path.rstrip('/') + '/releases/' + target_name)}"
                    f" && {post_deploy_command}",
                    timeout=COMMAND_TIMEOUT_SECONDS,
                )
                if on_output is not None:
                    await on_output(redact_text(post.output))
                if not post.ok:
                    return False, target_name, "دستور پس از بازگردانی ناموفق بود."

            return True, target_name, f"بازگردانی به نسخهٔ {target_name} انجام شد."
        except Exception as exc:  # noqa: BLE001
            return False, "", _ssh_error_message(exc)
        finally:
            with contextlib.suppress(Exception):
                await transport.close()


# ---------------------------------------------------------------------------
# remote scripts — built from constants, never from user input
# ---------------------------------------------------------------------------


def _release_name(now: float, sha: str) -> str:
    stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime(now))
    return f"{stamp}-{(sha or 'unknown')[:7]}"


def _switch_script(remote_path: str, release: str) -> str:
    """Move the ``current`` symlink atomically.

    A relative symlink into ``releases/`` plus ``mv -T`` on the same filesystem
    is the only switch that is atomic *and* survives the directory being moved.
    """
    base = shlex.quote(remote_path.rstrip("/"))
    rel = shlex.quote(f"releases/{release}")
    link = shlex.quote(f"{remote_path.rstrip('/')}/{CURRENT_LINK}")
    tmp = shlex.quote(f"{remote_path.rstrip('/')}/.{CURRENT_LINK}.new")
    return f"ln -sfn {rel} {tmp} && mv -T {tmp} {link} && cd {base} && ls -l {link}"


def _prune_script(remote_path: str, keep: int) -> str:
    """Delete all but the newest ``keep`` release directories."""
    releases = shlex.quote(f"{remote_path.rstrip('/')}/releases")
    return (
        f"cd {releases} && ls -1t | tail -n +{keep + 1} | "
        f"while read -r old; do rm -rf -- \"$old\"; done"
    )


def _list_releases_script(remote_path: str) -> str:
    releases = shlex.quote(f"{remote_path.rstrip('/')}/releases")
    return f"cd {releases} && ls -1t"


async def _probe_writable(transport: SshTransport, remote_path: str) -> CommandOutcome:
    base = shlex.quote(remote_path.rstrip("/") or "/")
    return await transport.run(
        f"mkdir -p {base} && test -w {base} && "
        f"probe={base}/.smart-org-probe && touch $probe && rm -f $probe",
        timeout=PROBE_TIMEOUT_SECONDS,
    )


async def _stream(callback: Callable[[str], Awaitable[None]] | None, text: str) -> None:
    if callback is not None:
        await callback(text)


def _ssh_error_message(exc: BaseException) -> str:
    """A plain Persian message. Never the raw exception, which can carry the host."""
    text = str(exc).lower()
    if "permission denied" in text or "auth" in text:
        return "احراز هویت روی سرور مقصد ناموفق بود."
    if "timed out" in text or "timeout" in text:
        return "اتصال به سرور مقصد در زمان مجاز برقرار نشد."
    if "host key" in text:
        return "کلید میزبان تأیید نشد."
    if "no such file" in text or "not found" in text:
        return "مسیر یا فایل مقصد پیدا نشد."
    if "connection refused" in text or "unreachable" in text:
        return "اتصال به سرور مقصد برقرار نشد."
    return "استقرار در سرور مقصد ناموفق بود."


__all__ = [
    "COMMAND_TIMEOUT_SECONDS",
    "CONNECT_TIMEOUT_SECONDS",
    "CURRENT_LINK",
    "KEEP_RELEASES",
    "AsyncSshTransport",
    "CommandOutcome",
    "SshDeployer",
    "SshTransport",
]