#!/usr/bin/env python3
"""
Daneshyar - Fast Atomic Update
Run: python scripts/update.py

Uploads only changed files, prepares an isolated release, warms Laravel caches,
then switches the live symlink atomically. The first run converts APP_DIR to a
release-based layout; later runs are fully atomic.

Requires: pip install paramiko
"""

from __future__ import annotations

import base64
import getpass
import hashlib
import io
import json
import os
import posixpath
import shlex
import shutil
import subprocess
import sys
import tarfile
import tempfile
import time
from pathlib import Path

if hasattr(sys.stdout, "reconfigure"):
    try:
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

# ── CONFIG (must match deploy.py) ─────────────────────────────────────────────
SERVER_HOST = "91.107.131.171"
SERVER_PORT = 2435
SSH_USER    = "root"
SSH_PASS    = os.environ.get("DANESHYAR_SSH_PASSWORD", "V9r!k2Qp7#Lm3@tX8sN5$zHu") #dont delete password fucking hell
APP_DIR     = "/var/www/html"
PHP_VERSION = "8.3"

# Release storage is deliberately outside APP_DIR so a symlink swap is atomic.
# /var/www/daneshyar is the canonical live app pointer; /var/www/html remains
# a compatibility symlink because Apache is already configured to use it.
LIVE_LINK    = "/var/www/daneshyar"
RELEASES_DIR = "/var/www/daneshyar-releases"
SHARED_DIR   = "/var/www/daneshyar-shared"
SOURCES_DIR  = "/var/www/daneshyar-sources"
PUBLIC_APP_URL = os.environ.get("DANESHYAR_APP_URL", "https://mydaneshyar.ir")
KEEP_RELEASES = 1
# ──────────────────────────────────────────────────────────────────────────────

ROOT = Path(__file__).resolve().parents[1]

SKIP_DIR_NAMES = {
    ".git", ".idea", ".vscode", ".agents", ".codex", "node_modules",
    "vendor", "__pycache__", "tmp",
}
SKIP_PATH_PREFIXES = {
    "build",
    "storage",
    "bootstrap/cache",
    # This is the 2,000-file Bootstrap Icons source package. Vite already emits
    # the only required font/CSS assets into public/build, so deploying the
    # package source wastes most of the update scan and transfer time.
    "public/assets/icons/bootstrap-icons",
}
SKIP_FILES = {".env", ".DS_Store", "Thumbs.db", ".deploy-manifest.json"}
SKIP_EXTENSIONS = {".zip", ".pyc"}


class UpdateError(RuntimeError):
    """A deployment step failed and the active release must remain unchanged."""


def _q(value: str) -> str:
    return shlex.quote(value)


def _connect(password: str):
    try:
        import paramiko
    except ImportError:
        print("ERROR: paramiko not installed. Run: pip install paramiko")
        sys.exit(1)

    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    print(f"Connecting to {SSH_USER}@{SERVER_HOST}:{SERVER_PORT} ...")
    client.connect(
        SERVER_HOST,
        port=SERVER_PORT,
        username=SSH_USER,
        password=password,
        timeout=30,
        banner_timeout=30,
        auth_timeout=30,
    )
    transport = client.get_transport()
    if transport is not None:
        transport.set_keepalive(20)
        # A larger SSH window materially improves throughput on higher latency links.
        transport.default_window_size = 16 * 1024 * 1024
    print("Connected.\n")
    return client


def _run(
    client,
    cmd: str,
    *,
    check: bool = True,
    show_command: bool = True,
    stream: bool = True,
) -> tuple[int, str, str]:
    if show_command:
        print(f"  -> {cmd}")
    # PTY mode merges command progress streams and avoids stderr-buffer
    # deadlocks during Composer. Quiet capture commands return only small JSON.
    stdin, stdout, stderr = client.exec_command(cmd, get_pty=stream)
    stdin.close()

    if stream:
        out_lines: list[str] = []
        for line in iter(stdout.readline, ""):
            print(f"     {line}", end="")
            out_lines.append(line)
        output = "".join(out_lines)
    else:
        output = stdout.read().decode("utf-8", errors="replace")

    code = stdout.channel.recv_exit_status()
    error = stderr.read().decode("utf-8", errors="replace")
    if check and code != 0:
        details = error.strip() or output.strip() or "No command output"
        raise UpdateError(f"Command failed ({code}): {details}")
    return code, output, error


def _is_skipped(relative_path: str) -> bool:
    parts = relative_path.split("/")
    if any(part in SKIP_DIR_NAMES for part in parts[:-1]):
        return True
    for skipped in SKIP_PATH_PREFIXES:
        if relative_path == skipped or relative_path.startswith(skipped + "/"):
            return True
    name = parts[-1]
    return name in SKIP_FILES or Path(name).suffix.lower() in SKIP_EXTENSIONS


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def build_local_manifest() -> tuple[dict[str, dict[str, int | str]], dict[str, Path]]:
    print("Scanning local release files ...")
    manifest: dict[str, dict[str, int | str]] = {}
    files: dict[str, Path] = {}

    for current_dir, dir_names, file_names in os.walk(ROOT, topdown=True):
        current = Path(current_dir)

        # Prune expensive trees before os.walk enters them. In particular this
        # avoids visiting tens of thousands of node_modules and Git files.
        kept_dirs: list[str] = []
        for name in dir_names:
            candidate = current / name
            relative_dir = candidate.relative_to(ROOT).as_posix()
            if name in SKIP_DIR_NAMES or _is_skipped(relative_dir + "/placeholder"):
                continue
            if candidate.is_symlink():
                continue
            kept_dirs.append(name)
        dir_names[:] = kept_dirs

        for name in file_names:
            path = current / name
            if path.is_symlink():
                continue
            relative = path.relative_to(ROOT).as_posix()
            if _is_skipped(relative):
                continue
            stat = path.stat()
            manifest[relative] = {
                "sha256": _sha256(path),
                "size": stat.st_size,
            }
            files[relative] = path

    return manifest, files


def _write_remote_text(sftp, remote_path: str, content: str) -> None:
    with sftp.file(remote_path, "w") as handle:
        handle.write(content)
        handle.flush()


def ensure_release_layout(client, release_id: str) -> tuple[str, str]:
    """Return (current_release, new_release), bootstrapping symlinks once."""
    bootstrap = f"""set -eu
mkdir -p {_q(RELEASES_DIR)} {_q(SHARED_DIR)} {_q(SOURCES_DIR)}
if [ ! -L {_q(APP_DIR)} ]; then
  test -d {_q(APP_DIR)}
  bootstrap_release={_q(posixpath.join(RELEASES_DIR, 'bootstrap-' + release_id))}
  mv {_q(APP_DIR)} "$bootstrap_release"

  if [ -f "$bootstrap_release/.env" ] && [ ! -f {_q(posixpath.join(SHARED_DIR, '.env'))} ]; then
    mv "$bootstrap_release/.env" {_q(posixpath.join(SHARED_DIR, '.env'))}
  fi
  if [ -f {_q(posixpath.join(SHARED_DIR, '.env'))} ]; then
    rm -f "$bootstrap_release/.env"
    ln -s {_q(posixpath.join(SHARED_DIR, '.env'))} "$bootstrap_release/.env"
  fi

  if [ -d "$bootstrap_release/storage" ] && [ ! -e {_q(posixpath.join(SHARED_DIR, 'storage'))} ]; then
    mv "$bootstrap_release/storage" {_q(posixpath.join(SHARED_DIR, 'storage'))}
  fi
  rm -rf "$bootstrap_release/storage"
  ln -s {_q(posixpath.join(SHARED_DIR, 'storage'))} "$bootstrap_release/storage"
fi

current_release=$(readlink -f {_q(APP_DIR)})
if [ -e {_q(LIVE_LINK)} ] && [ ! -L {_q(LIVE_LINK)} ]; then
  if [ -d {_q(LIVE_LINK)} ] && [ -z "$(find {_q(LIVE_LINK)} -mindepth 1 -maxdepth 1 -print -quit)" ]; then
    rmdir {_q(LIVE_LINK)}
  else
    mv {_q(LIVE_LINK)} {_q(posixpath.join(SOURCES_DIR, 'legacy-daneshyar-' + release_id))}
  fi
fi
if [ ! -L {_q(LIVE_LINK)} ] || [ "$(readlink -f {_q(LIVE_LINK)})" != "$current_release" ]; then
  ln -s "$current_release" {_q(LIVE_LINK + '.next-' + release_id)}
  mv -Tf {_q(LIVE_LINK + '.next-' + release_id)} {_q(LIVE_LINK)}
fi
if [ ! -L {_q(APP_DIR)} ] || [ "$(readlink {_q(APP_DIR)})" != {_q(LIVE_LINK)} ]; then
  ln -s {_q(LIVE_LINK)} {_q(APP_DIR + '.next-' + release_id)}
  mv -Tf {_q(APP_DIR + '.next-' + release_id)} {_q(APP_DIR)}
fi
current_release=$(readlink -f {_q(LIVE_LINK)})
new_release={_q(posixpath.join(RELEASES_DIR, release_id))}
test ! -e "$new_release"
mkdir -p "$new_release"
cp -al "$current_release/." "$new_release/"
# .env must always be a symlink to the shared file — a hardlinked copy goes
# stale the moment the shared .env is replaced with a new inode.
rm -f "$new_release/.env"
ln -s {_q(posixpath.join(SHARED_DIR, '.env'))} "$new_release/.env"
printf '%s\n%s\n' "$current_release" "$new_release"
"""
    _, output, _ = _run(
        client,
        bootstrap,
        show_command=False,
        stream=False,
    )
    lines = [line.strip() for line in output.splitlines() if line.strip()]
    if len(lines) < 2:
        raise UpdateError("Could not determine current and new release paths.")
    return lines[-2], lines[-1]


def ensure_shared_environment(client) -> None:
    """Keep production URL sane before Laravel caches config."""
    script = f'''from pathlib import Path
env_path = Path({posixpath.join(SHARED_DIR, ".env")!r})
public_url = {PUBLIC_APP_URL!r}.rstrip("/")
lines = env_path.read_text(encoding="utf-8", errors="replace").splitlines()
seen = set()
changed = False
next_lines = []
for line in lines:
    if not line or line.lstrip().startswith("#") or "=" not in line:
        next_lines.append(line)
        continue
    key, value = line.split("=", 1)
    seen.add(key)
    raw = value.strip().strip('"').strip("'")
    if key == "APP_URL" and (not raw or "127.0.0.1" in raw or "localhost" in raw):
        next_lines.append(f"APP_URL={{public_url}}")
        changed = True
    elif key == "ASSET_URL" and ("127.0.0.1" in raw or "localhost" in raw):
        next_lines.append("ASSET_URL=")
        changed = True
    else:
        next_lines.append(line)
if "APP_URL" not in seen:
    next_lines.append(f"APP_URL={{public_url}}")
    changed = True
if changed:
    backup = env_path.with_name(".env.bak-url")
    backup.write_text(env_path.read_text(encoding="utf-8", errors="replace"), encoding="utf-8")
    env_path.write_text("\\n".join(next_lines) + "\\n", encoding="utf-8")
    print(f"Updated APP_URL/ASSET_URL in {{env_path}}")
else:
    print("APP_URL/ASSET_URL already production-safe")
'''
    remote_script = f"/tmp/daneshyar-env-{os.getpid()}.py"
    sftp = client.open_sftp()
    try:
        _write_remote_text(sftp, remote_script, script)
        _run(client, f"python3 {_q(remote_script)}", show_command=False)
    finally:
        try:
            sftp.remove(remote_script)
        except OSError:
            pass
        sftp.close()


REMOTE_PLANNER = r'''#!/usr/bin/env python3
import hashlib
import json
import os
import shutil
import sys

release, incoming_path, plan_path = sys.argv[1:4]
manifest_path = os.path.join(release, ".deploy-manifest.json")

with open(incoming_path, "r", encoding="utf-8") as handle:
    incoming = json.load(handle)

try:
    with open(manifest_path, "r", encoding="utf-8") as handle:
        previous = json.load(handle)
except (FileNotFoundError, json.JSONDecodeError):
    previous = {}

def digest(path):
    value = hashlib.sha256()
    with open(path, "rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            value.update(chunk)
    return value.hexdigest()

changed = []
for relative, metadata in incoming.items():
    old = previous.get(relative)
    existing = os.path.join(release, *relative.split("/"))
    if old and old.get("sha256") == metadata["sha256"]:
        try:
            if os.path.getsize(existing) == metadata["size"] and digest(existing) == metadata["sha256"]:
                continue
        except (FileNotFoundError, IsADirectoryError, PermissionError):
            pass
        changed.append(relative)
        continue
    if not previous:
        try:
            if os.path.getsize(existing) == metadata["size"] and digest(existing) == metadata["sha256"]:
                continue
        except (FileNotFoundError, IsADirectoryError, PermissionError):
            pass
    changed.append(relative)

removed = sorted(set(previous) - set(incoming))
for relative in removed:
    target = os.path.join(release, *relative.split("/"))
    if os.path.isdir(target) and not os.path.islink(target):
        shutil.rmtree(target)
    else:
        try:
            os.unlink(target)
        except FileNotFoundError:
            pass

for relative in changed:
    os.makedirs(os.path.dirname(os.path.join(release, *relative.split("/"))), exist_ok=True)

with open(plan_path, "w", encoding="utf-8") as handle:
    json.dump({"changed": changed, "removed": removed}, handle, ensure_ascii=False)
'''


def plan_delta(client, sftp, release: str, release_id: str, manifest: dict) -> tuple[list[str], list[str], str]:
    incoming_path = f"/tmp/daneshyar-manifest-{release_id}.json"
    planner_path = f"/tmp/daneshyar-planner-{release_id}.py"
    plan_path = f"/tmp/daneshyar-plan-{release_id}.json"

    _write_remote_text(sftp, incoming_path, json.dumps(manifest, separators=(",", ":")))
    _write_remote_text(sftp, planner_path, REMOTE_PLANNER)
    _run(
        client,
        f"python3 {_q(planner_path)} {_q(release)} {_q(incoming_path)} {_q(plan_path)}",
        show_command=False,
        stream=False,
    )

    with sftp.file(plan_path, "r") as handle:
        plan = json.loads(handle.read().decode("utf-8"))

    for temporary in (planner_path, plan_path):
        try:
            sftp.remove(temporary)
        except OSError:
            pass
    # Keep incoming_path until it is atomically installed as the release manifest.
    return plan["changed"], plan["removed"], incoming_path


def _atomic_sftp_replace(sftp, source: str, destination: str) -> None:
    try:
        sftp.posix_rename(source, destination)
    except OSError:
        try:
            sftp.remove(destination)
        except OSError:
            pass
        sftp.rename(source, destination)


def upload_delta(
    client,
    sftp,
    release: str,
    changed: list[str],
    files: dict[str, Path],
    manifest_temp: str,
    password: str,
) -> int:
    source_bytes = sum(files[relative].stat().st_size for relative in changed)
    archive_path = ""

    try:
        with tempfile.NamedTemporaryFile(prefix="daneshyar-delta-", suffix=".tar.gz", delete=False) as temporary:
            archive_path = temporary.name

        # One archive avoids an SFTP request/response cycle for every tiny file.
        # This route benchmarks at only ~18 KB/s, so maximum compression saves
        # much more time than its sub-second CPU cost for a typical delta.
        with tarfile.open(archive_path, "w:gz", compresslevel=9) as archive:
            for relative in changed:
                info = archive.gettarinfo(str(files[relative]), arcname=relative)
                info.mode = 0o644
                info.uid = 0
                info.gid = 0
                info.uname = "root"
                info.gname = "root"
                with files[relative].open("rb") as source:
                    archive.addfile(info, source)

            with sftp.file(manifest_temp, "rb") as remote_manifest:
                manifest_data = remote_manifest.read()
            manifest_info = tarfile.TarInfo(".deploy-manifest.json")
            manifest_info.size = len(manifest_data)
            manifest_info.mode = 0o644
            manifest_info.mtime = int(time.time())
            archive.addfile(manifest_info, io.BytesIO(manifest_data))

        archive_bytes = os.path.getsize(archive_path)
        print(
            f"  Streaming {len(changed)} changed files in one SSH transfer "
            f"({source_bytes / 1024:.1f} KB -> {archive_bytes / 1024:.1f} KB) ..."
        )

        started = time.monotonic()
        pscp = None
        if os.name == "nt":
            pscp = shutil.which("pscp.exe") or shutil.which("pscp")

        if pscp:
            # Native PuTTY SCP avoids Paramiko's expensive per-request SFTP
            # behavior. Pinning the SSH host key keeps this noninteractive.
            server_key = client.get_transport().get_remote_server_key()
            fingerprint = "SHA256:" + base64.b64encode(
                hashlib.sha256(server_key.asbytes()).digest()
            ).decode().rstrip("=")
            remote_archive = f"/tmp/daneshyar-delta-{os.getpid()}.tar.gz"
            target = f"{SSH_USER}@{SERVER_HOST}:{remote_archive}"
            print("  Transport: native PSCP (single archive)")
            result = subprocess.run(
                [
                    pscp, "-batch", "-P", str(SERVER_PORT), "-pw", password,
                    "-hostkey", fingerprint, archive_path, target,
                ],
                timeout=300,
            )
            if result.returncode != 0:
                raise UpdateError(f"PSCP failed with exit code {result.returncode}.")
            try:
                _run(
                    client,
                    f"tar -xzf {_q(remote_archive)} --unlink-first --no-same-owner -C {_q(release)}",
                    show_command=False,
                )
            finally:
                _run(client, f"rm -f {_q(remote_archive)}", check=False, show_command=False, stream=False)
        else:
            # Portable fallback: one SSH stream. Reading exactly archive_bytes
            # avoids relying on Paramiko's unreliable half-close behavior here.
            print("  Transport: raw SSH stream (single archive)")
            command = (
                f"head -c {archive_bytes} | "
                f"tar -xzf - --unlink-first --no-same-owner -C {_q(release)}"
            )
            stdin, stdout, stderr = client.exec_command(command)
            stdin.channel.settimeout(120)
            with open(archive_path, "rb") as source:
                while chunk := source.read(256 * 1024):
                    stdin.channel.sendall(chunk)
            stdin.channel.shutdown_write()
            deadline = time.monotonic() + 120
            while not stdout.channel.exit_status_ready():
                if time.monotonic() >= deadline:
                    stdout.channel.close()
                    raise UpdateError("SSH archive extraction did not finish within 120 seconds.")
                time.sleep(0.05)
            code = stdout.channel.recv_exit_status()
            error = stderr.read().decode("utf-8", errors="replace").strip()
            if code != 0:
                raise UpdateError(f"SSH archive extraction failed ({code}): {error or 'no output'}")

        elapsed = max(time.monotonic() - started, 0.001)
        print(f"  Transfer completed in {elapsed:.2f}s ({archive_bytes / 1024 / elapsed:.1f} KB/s).")
        try:
            sftp.remove(manifest_temp)
        except OSError:
            pass
        return archive_bytes
    finally:
        if archive_path:
            try:
                os.unlink(archive_path)
            except OSError:
                pass


def prepare_release(client, release: str, changed: list[str]) -> None:
    php = f"php{PHP_VERSION}"
    composer_changed = any(path in {"composer.json", "composer.lock"} for path in changed)

    print("\nPreparing isolated release ...")
    writable_dirs = [
        release + "/bootstrap/cache",
        SHARED_DIR + "/storage/framework/cache",
        SHARED_DIR + "/storage/framework/sessions",
        SHARED_DIR + "/storage/framework/views",
        SHARED_DIR + "/storage/logs",
        SHARED_DIR + "/storage/app",
        SHARED_DIR + "/storage/app/private",
        SHARED_DIR + "/storage/app/public",
    ]
    _run(client, "install -d -o www-data -g www-data -m 775 " + " ".join(_q(path) for path in writable_dirs))
    _run(client, f"rm -f {_q(release + '/bootstrap/cache')}/*.php", show_command=False)

    code, _, _ = _run(
        client,
        f"test -f {_q(release + '/vendor/autoload.php')}",
        check=False,
        show_command=False,
        stream=False,
    )
    if composer_changed or code != 0:
        print("  Composer files changed; rebuilding vendor for this release ...")
        _run(client, f"rm -rf {_q(release + '/vendor')}", show_command=False)
        install_command = (
            f"cd {_q(release)} && COMPOSER_ALLOW_SUPERUSER=1 {php} /usr/local/bin/composer "
            "install --no-dev --prefer-dist --optimize-autoloader --no-interaction"
        )
        install_code, install_output, install_error = _run(
            client,
            install_command,
            check=False,
        )
        if install_code != 0:
            lock_error = install_output + install_error
            if 'phpseclib/phpseclib' not in lock_error or 'not present in the lock file' not in lock_error:
                details = install_error.strip() or install_output.strip() or 'No command output'
                raise UpdateError(f"Command failed ({install_code}): {details}")
            print("  Lock file predates phpseclib; resolving only phpseclib and its dependencies ...")
            _run(
                client,
                f"cd {_q(release)} && COMPOSER_ALLOW_SUPERUSER=1 {php} /usr/local/bin/composer "
                "update phpseclib/phpseclib --with-dependencies --no-dev --prefer-dist "
                "--optimize-autoloader --no-interaction",
            )
    else:
        print("  Composer unchanged; reusing vendor instantly.")

    print("  Applying database migrations ...")
    _run(client, f"cd {_q(release)} && {php} artisan migrate --force", show_command=False)

    _run(client, f"cd {_q(release)} && {php} artisan storage:link --force", show_command=False)

    print("  Fixing ownership and writable paths ...")
    permission_cmd = f"""set -eu
chown -h www-data:www-data {_q(release + '/.env')} {_q(release + '/storage')} {_q(release + '/public/storage')} 2>/dev/null || true
chown -R www-data:www-data {_q(release)} {_q(posixpath.join(SHARED_DIR, 'storage'))}
find {_q(release)} -type d -exec chmod 755 {{}} +
find {_q(release)} -type f -exec chmod 644 {{}} +
chmod 755 {_q(release + '/artisan')}
find {_q(release + '/bootstrap/cache')} {_q(posixpath.join(SHARED_DIR, 'storage'))} -type d -exec chmod 775 {{}} +
find {_q(posixpath.join(SHARED_DIR, 'storage'))} -type f -exec chmod 664 {{}} +
"""
    _run(client, permission_cmd, show_command=False)

    print("  Warming Laravel caches ...")
    _run(client, f"cd {_q(release)} && runuser -u www-data -- {php} artisan config:cache && runuser -u www-data -- {php} artisan route:cache && runuser -u www-data -- {php} artisan view:cache", show_command=False)


def atomic_switch(client, release: str, release_id: str) -> None:
    next_link = f"{LIVE_LINK}.next-{release_id}"
    command = (
        f"ln -s {_q(release)} {_q(next_link)} && "
        f"mv -Tf {_q(next_link)} {_q(LIVE_LINK)} && "
        f"if [ ! -L {_q(APP_DIR)} ] || [ \"$(readlink {_q(APP_DIR)})\" != {_q(LIVE_LINK)} ]; then "
        f"ln -s {_q(LIVE_LINK)} {_q(APP_DIR + '.next-' + release_id)} && "
        f"mv -Tf {_q(APP_DIR + '.next-' + release_id)} {_q(APP_DIR)}; fi"
    )
    _run(client, command, show_command=False)


def rollback_switch(client, previous_release: str, release_id: str) -> None:
    rollback_link = f"{LIVE_LINK}.rollback-{release_id}"
    _run(
        client,
        f"ln -s {_q(previous_release)} {_q(rollback_link)} && mv -Tf {_q(rollback_link)} {_q(LIVE_LINK)} && systemctl reload apache2",
        check=False,
        show_command=False,
    )


def cleanup_old_releases(client, active_release: str) -> None:
    cleanup_script = f'''import os, shutil
root = {RELEASES_DIR!r}
active = os.path.realpath({active_release!r})
keep = {KEEP_RELEASES}
items = []
for name in os.listdir(root):
    path = os.path.join(root, name)
    if os.path.isdir(path) and not os.path.islink(path):
        items.append((os.path.getmtime(path), path))
for _, path in sorted(items, reverse=True)[keep:]:
    if os.path.realpath(path) != active:
        shutil.rmtree(path)
'''
    sftp = client.open_sftp()
    remote_script = f"/tmp/daneshyar-clean-{os.getpid()}.py"
    try:
        _write_remote_text(sftp, remote_script, cleanup_script)
        _run(client, f"python3 {_q(remote_script)}", check=False, show_command=False, stream=False)
    finally:
        try:
            sftp.remove(remote_script)
        except OSError:
            pass
        sftp.close()


def speed_test(client, password: str) -> None:
    """Measure the exact native SCP transport used for Windows updates."""
    size = 1024 * 1024
    remote_path = f"/tmp/daneshyar-speedtest-{os.getpid()}.bin"
    local_path = ""
    pscp = None
    if os.name == "nt":
        pscp = shutil.which("pscp.exe") or shutil.which("pscp")
    if not pscp:
        raise UpdateError("PSCP was not found. Install PuTTY or add pscp.exe to PATH.")

    with tempfile.NamedTemporaryFile(prefix="daneshyar-speedtest-", suffix=".bin", delete=False) as temporary:
        local_path = temporary.name
        temporary.write(os.urandom(size))

    server_key = client.get_transport().get_remote_server_key()
    fingerprint = "SHA256:" + base64.b64encode(
        hashlib.sha256(server_key.asbytes()).digest()
    ).decode().rstrip("=")
    target = f"{SSH_USER}@{SERVER_HOST}:{remote_path}"
    print("Testing a 1 MB native PSCP upload ...")
    _run(client, "rm -f /tmp/daneshyar-speedtest*.bin /tmp/daneshyar-pscp-speedtest.bin", check=False, show_command=False, stream=False)
    started = time.monotonic()
    try:
        result = subprocess.run(
            [
                pscp, "-batch", "-P", str(SERVER_PORT), "-pw", password,
                "-hostkey", fingerprint, local_path, target,
            ],
            timeout=120,
        )
        elapsed = time.monotonic() - started
        if result.returncode != 0:
            raise UpdateError(f"PSCP speed test failed with exit code {result.returncode}.")
        print(f"  1.0 MB in {elapsed:.2f}s = {size / 1024 / elapsed:.1f} KB/s")
    finally:
        _run(client, f"rm -f {_q(remote_path)}", check=False, show_command=False, stream=False)
        if local_path:
            for _ in range(5):
                try:
                    os.unlink(local_path)
                    break
                except PermissionError:
                    time.sleep(0.2)


def deploy(client, password: str) -> None:
    if not (ROOT / "public" / "build" / "manifest.json").exists():
        raise UpdateError("public/build/manifest.json is missing. Run: python scripts/build.py")

    started = time.monotonic()
    release_id = time.strftime("%Y%m%d-%H%M%S")
    manifest, files = build_local_manifest()
    total_size = sum(int(item["size"]) for item in manifest.values())
    print(f"  {len(files)} release files indexed ({total_size / 1024 / 1024:.1f} MB total).")

    previous_release = ""
    new_release = ""
    switched = False
    sftp = client.open_sftp()
    try:
        print("\nCreating copy-on-write release ...")
        previous_release, new_release = ensure_release_layout(client, release_id)
        ensure_shared_environment(client)
        changed, removed, manifest_temp = plan_delta(client, sftp, new_release, release_id, manifest)
        print(f"  Delta: {len(changed)} changed, {len(removed)} removed, {len(files) - len(changed)} reused.")
        transferred = upload_delta(client, sftp, new_release, changed, files, manifest_temp, password)
        sftp.close()

        prepare_release(client, new_release, changed)
        print("\nSwitching live release atomically ...")
        atomic_switch(client, new_release, release_id)
        switched = True
        _run(client, "systemctl reload apache2", show_command=False)
        # Long-running workers hold the previous release and cached config in memory.
        _run(client, "systemctl try-restart daneshyar-queue || true", show_command=False)
        _run(client, "systemctl try-restart laravel-reverb || true", show_command=False)
        cleanup_old_releases(client, new_release)

        elapsed = time.monotonic() - started
        print("\n" + "=" * 54)
        print("  ATOMIC UPDATE COMPLETE")
        print("=" * 54)
        print(f"  Changed files: {len(changed)}")
        print(f"  Uploaded:      {transferred / 1024:.1f} KB compressed")
        print(f"  Reused files:  {len(files) - len(changed)}")
        print(f"  Duration:      {elapsed:.1f}s")
        print(f"  Release:       {release_id}")
        print("=" * 54)
    except Exception:
        try:
            sftp.close()
        except Exception:
            pass
        if switched and previous_release:
            print("\nUpdate failed after activation; restoring previous release ...")
            rollback_switch(client, previous_release, release_id)
        elif new_release:
            print("\nUpdate failed before activation; live site was not changed.")
            _run(client, f"rm -rf {_q(new_release)}", check=False, show_command=False, stream=False)
        raise


def main() -> None:
    print("=" * 54)
    print("  Daneshyar - Fast Atomic Update")
    print("=" * 54)
    print(f"\n  Target: {SSH_USER}@{SERVER_HOST}:{SERVER_PORT}")
    print(f"  App:    {APP_DIR}\n")
    print(f"  Live:   {LIVE_LINK}")
    print(f"  URL:    {PUBLIC_APP_URL}\n")

    password = SSH_PASS or getpass.getpass(f"SSH password for {SSH_USER}@{SERVER_HOST}: ")
    client = _connect(password)
    try:
        if "--speed-test" in sys.argv:
            speed_test(client, password)
        else:
            deploy(client, password)
    except (UpdateError, OSError) as exc:
        print(f"\n[FAILED] {exc}")
        sys.exit(1)
    finally:
        client.close()


if __name__ == "__main__":
    main()