#!/usr/bin/env python3
"""supaspin - create, list, show and delete throwaway Supabase instances on microk8s.

Each numbered instance gets its own namespace, capped disk images, capped memory,
an HTTPS hostname and fixed Postgres/Studio ports. Standard library plus python3-yaml.
Run as root: supaspin            (menu)
             supaspin create|list|show N|env N|tunnel N|delete N|setup|mail
             supaspin push-env N TARGET [--port P] [--key FILE] [--dry-run]
               TARGET = a .env file on this server (/path/.env) or on another computer
               (user@host:/path/.env, Windows: user@host:C:/path/.env). Puts the instance's
               KEY=VALUE pairs into it: existing keys get the new value, missing keys are added.
"""

import argparse
import base64
import glob
import hashlib
import hmac
import json
import os
import pwd
import re
import secrets
import shutil
import string
import subprocess
import sys
import tempfile
import termios
import time
import tty

import yaml

VERSION = "1.1.2"
CONFIG_PATH = os.environ.get("SUPASPIN_CONFIG", "/etc/supaspin/config.yaml")
STATE_DIR = os.environ.get("SUPASPIN_STATE", "/var/lib/supaspin")
CHART_REPO = "https://supabase-community.github.io/supabase-kubernetes"
CHART_NAME = "supabase"
CHART_VERSION = "0.8.0"

STORAGE_CLASS = "supaspin-local-sc"
ISSUER = "supaspin-letsencrypt"
PRIORITY_CLASS = "supaspin-low"
CERTS_NS = "supaspin-certs"

# Image cap per volume, in MiB. The chart's separate imgproxy volume is not used: imgproxy has to
# read the uploaded files, so it mounts the storage volume instead (see instance_values).
VOLUMES = {"db": 2048, "storage": 1024, "functions": 256,
           "snippets": 256, "deno": 256, "pgsodium": 256}
# The user Postgres runs as inside supabase/postgres 17.6.1.136 (`id postgres` in the image).
# Production's older image used 2000; do not copy that number.
DB_OWNED = ("db", "pgsodium")
DB_UID, DB_GID = 100, 101

# Memory per service: (request MiB, limit MiB). Half of maralago.ru production, except analytics,
# realtime and vector: at half of production they were killed for running out of memory in a loop
# (measured on a fresh instance: analytics peaks at 624 MiB, realtime at 280 MiB; vector reads the log
# files of every pod on the node, not only its own instance's, and was killed at 256 MiB on 2026-09-19).
SERVICES = ["db", "analytics", "kong", "auth", "functions", "imgproxy", "meta",
            "realtime", "rest", "storage", "studio", "vector"]
MEMORY = {"db": (512, 2048), "analytics": (128, 768), "kong": (128, 512), "realtime": (128, 384),
          "vector": (128, 512)}
DEFAULT_MEMORY = (128, 256)

DEFAULT_MAIL = {"host": "smtp.example.com", "port": 587, "user": "", "password": "",
                "admin_email": "admin@example.com", "sender_name": "Supabase"}

BOLD, DIM, RED, REV, OFF = "\x1b[1m", "\x1b[2m", "\x1b[31m", "\x1b[7m", "\x1b[0m"


# ------------------------------------------------------------ pure helpers --

def pg_port(n):
    return 5440 + n


def studio_port(n):
    return 3000 + n


def namespace(n):
    return f"supaspin-{n}"


def release(n):
    return f"supa{n}"


def fullname(n):
    """Name the chart gives to its objects: <release>-supabase."""
    return f"{release(n)}-supabase"


def pvc_name(n, vol):
    return f"{fullname(n)}-{vol}"


def tls_secret(n):
    return f"supa{n}-tls"


def lowest_free_number(slots, taken):
    """Lowest slot number that is not in `taken`; None if every slot is used."""
    for n in sorted(int(k) for k in slots):
        if n not in taken:
            return n
    return None


def b64url(data):
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def sign_jwt(secret, role, now=None, years=10):
    """HS256 token with the claims Supabase expects for the anon/service keys."""
    now = int(now if now is not None else time.time())
    header = b64url(json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode())
    payload = b64url(json.dumps(
        {"role": role, "iss": "supabase", "iat": now, "exp": now + years * 365 * 86400},
        separators=(",", ":")).encode())
    sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(), hashlib.sha256).digest()
    return f"{header}.{payload}.{b64url(sig)}"


def rand(n, alphabet=string.ascii_letters + string.digits):
    return "".join(secrets.choice(alphabet) for _ in range(n))


def generate_secrets():
    jwt_secret = rand(48)
    return {
        "JWT_SECRET": jwt_secret,
        "ANON_KEY": sign_jwt(jwt_secret, "anon"),
        "SERVICE_KEY": sign_jwt(jwt_secret, "service_role"),
        "DB_PASSWORD": rand(32),
        "DASHBOARD_USERNAME": "supabase",
        "DASHBOARD_PASSWORD": rand(24),
        "LOGFLARE_PUBLIC": rand(40),
        "LOGFLARE_PRIVATE": rand(40),
        "S3_KEY_ID": secrets.token_hex(16),
        "S3_ACCESS_KEY": secrets.token_hex(32),
        "REALTIME_SECRET_KEY_BASE": base64.b64encode(secrets.token_bytes(48)).decode(),
        "REALTIME_DB_ENC_KEY": secrets.token_hex(8),  # must be exactly 16 characters
        "META_CRYPTO_KEY": secrets.token_hex(16),
    }


def find_template():
    for d in (os.environ.get("SUPASPIN_TEMPLATES"),
              os.path.join(os.path.dirname(os.path.realpath(__file__)), "templates"),
              "/usr/share/supaspin/templates"):
        if d and os.path.exists(os.path.join(d, "values.yaml.tpl")):
            return os.path.join(d, "values.yaml.tpl")
    raise SystemExit("values.yaml.tpl not found (reinstall supaspin)")


def instance_values(cfg, n, sec, mail, template_text):
    """Fill the template, then add memory caps, priority class and volume sizes."""
    host = cfg["slots"][n]
    subs = dict(sec)
    subs.update({
        "PUBLIC_URL": f"https://{host}", "HOST": host,
        "TLS_SECRET": tls_secret(n), "INGRESS_CLASS": cfg["ingress_class"],
        "CLUSTER_ISSUER": ISSUER, "PROJECT_NAME": f"supaspin-{n}",
        "SMTP_HOST": mail["host"], "SMTP_PORT": mail["port"], "SMTP_USER": mail["user"],
        "SMTP_PASS": mail["password"], "SMTP_ADMIN_EMAIL": mail["admin_email"],
        "SMTP_SENDER_NAME": mail["sender_name"],
    })
    # json.dumps gives a quoted YAML-safe string; the port is quoted as well (env values are strings).
    quoted = {k: json.dumps(str(v)) for k, v in subs.items()}
    text = re.sub(r"\$\{(\w+)\}", lambda m: quoted[m.group(1)], template_text)
    values = yaml.safe_load(text)
    dep = values.setdefault("deployment", {})
    for svc in SERVICES:
        req, lim = MEMORY.get(svc, DEFAULT_MEMORY)
        d = dep.setdefault(svc, {})
        d["resources"] = {"requests": {"memory": f"{req}Mi"}, "limits": {"memory": f"{lim}Mi"}}
        d["priorityClassName"] = PRIORITY_CLASS
    persistence = values.setdefault("persistence", {})
    for vol, mib in VOLUMES.items():
        persistence[vol] = {"enabled": True, "storageClassName": STORAGE_CLASS,
                            "size": f"{mib}Mi"}
    # With the file backend storage-api gives imgproxy a local path, so imgproxy must mount the
    # very volume the uploads are written to; the chart's own imgproxy volume is empty and useless.
    persistence["imgproxy"] = {"enabled": True, "existingClaim": pvc_name(n, "storage")}
    return values


def total_memory_mib():
    """(requests, limits) in MiB, summed over all services."""
    req = sum(MEMORY.get(s, DEFAULT_MEMORY)[0] for s in SERVICES)
    lim = sum(MEMORY.get(s, DEFAULT_MEMORY)[1] for s in SERVICES)
    return req, lim


def render_env(cfg, n, sec, mail):
    host = cfg["slots"][n]
    url = f"https://{host}"
    lan = cfg["lan_ip"]
    pw = sec["DB_PASSWORD"]
    lines = [
        f"# supaspin instance {n} - {url}",
        "# Reach Postgres and Studio from outside the LAN through an SSH tunnel:",
        f"#   {tunnel_command(cfg, n)}",
        "",
        f"SUPABASE_URL={url}", f"SUPABASE_PUBLIC_URL={url}",
        f"PUBLIC_SUPABASE_URL={url}", f"API_EXTERNAL_URL={url}",
        "",
        f"PUBLIC_SUPABASE_ANON_KEY={sec['ANON_KEY']}", f"ANON_KEY={sec['ANON_KEY']}",
        f"JWT_ANON_KEY={sec['ANON_KEY']}",
        f"SERVICE_ROLE_KEY={sec['SERVICE_KEY']}", f"JWT_SERVICE_KEY={sec['SERVICE_KEY']}",
        f"JWT_SECRET={sec['JWT_SECRET']}", "JWT_EXPIRY=3600",
        "",
        f"POSTGRES_PASSWORD={pw}", "POSTGRES_DB=postgres", "POSTGRES_HOST=localhost",
        f"POSTGRES_PORT={pg_port(n)}",
        f"# On the LAN use {lan}:{pg_port(n)} instead of localhost.",
        f"DATABASE_URL=postgresql://postgres:{pw}@localhost:{pg_port(n)}/postgres",
        "",
        f"DASHBOARD_USERNAME={sec['DASHBOARD_USERNAME']}",
        f"DASHBOARD_PASSWORD={sec['DASHBOARD_PASSWORD']}",
        "",
        f"GOTRUE_SMTP_HOST={mail['host']}", f"GOTRUE_SMTP_PORT={mail['port']}",
        f"GOTRUE_SMTP_USER={mail['user']}", f"GOTRUE_SMTP_PASS={mail['password']}",
        f"GOTRUE_SMTP_ADMIN_EMAIL={mail['admin_email']}",
        f"GOTRUE_SMTP_SENDER_NAME={mail['sender_name']}",
        "",
    ]
    return "\n".join(lines)


SECRET_KEYS = ("KEY", "SECRET", "PASSWORD", "DATABASE_URL", "PASS")


def mask_env(text):
    out = []
    for line in text.splitlines():
        if "=" in line and not line.startswith("#"):
            k, _, v = line.partition("=")
            if any(s in k for s in SECRET_KEYS) and v:
                line = f"{k}=********"
        out.append(line)
    return "\n".join(out)


def tunnel_command(cfg, n):
    lan = cfg["lan_ip"]
    return (f"ssh -N -p {cfg['ssh_port']} -L {pg_port(n)}:{lan}:{pg_port(n)} "
            f"-L {studio_port(n)}:{lan}:{studio_port(n)} {cfg['ssh_user']}@{cfg['ssh_host']}")


def tunnel_socket(n):
    """The 'control file' of a background tunnel: ssh is later told to quit through it."""
    return f"/tmp/supaspin-{n}.sock"


def tunnel_background_command(cfg, n):
    lan = cfg["lan_ip"]
    return (f"ssh -f -N -M -S {tunnel_socket(n)} -o ExitOnForwardFailure=yes -p {cfg['ssh_port']} "
            f"-L {pg_port(n)}:{lan}:{pg_port(n)} -L {studio_port(n)}:{lan}:{studio_port(n)} "
            f"{cfg['ssh_user']}@{cfg['ssh_host']}")


def tunnel_control_command(cfg, n, action):
    """`check` asks the background tunnel whether it runs, `exit` closes it."""
    return f"ssh -S {tunnel_socket(n)} -O {action} {cfg['ssh_user']}@{cfg['ssh_host']}"


def tunnel_guide(cfg, n, detailed=False):
    """Plain-language tunnel instructions for instance n, as a list of lines."""
    if not cfg.get("ssh_host"):
        return ["Tunnel not available: run 'sudo supaspin setup' and fill in 'Public SSH host for tunnels'."]
    lines = [
        "Tunnel = a private connection from YOUR computer to this instance, so you can reach its",
        f"database (localhost:{pg_port(n)}) and Studio admin screen (http://localhost:{studio_port(n)}) from your computer.",
        "Run the commands below in a terminal on YOUR computer, not on this server.",
        "",
        "IMPORTANT - which computer you use matters:",
        "  Mac or Linux: all the commands below work.",
        "  Windows: ONLY the first command below (keep the tunnel open in the window) works. The background",
        "  and closing commands do NOT work on Windows, because the ssh program that comes with Windows",
        '  does not support the "control file" they need. On Windows keep the tunnel open in its own',
        "  PowerShell window and close it with Ctrl+C, or by closing that window.",
        "",
        "If you want to open the tunnel and keep it open in the terminal window you are using, run:",
        f"  {tunnel_command(cfg, n)}",
        "The window then looks stuck and prints nothing - that is normal, it means the tunnel is open.",
        "To close it, press Ctrl+C in that window.",
        "",
        "If you want the tunnel to keep running in the background, so your terminal stays free (Mac or Linux only), run:",
        f"  {tunnel_background_command(cfg, n)}",
        "If it asks for your password, type it. Then you get your terminal back and the tunnel stays open.",
        "",
        "If you no longer need the background tunnel and want to close it (Mac or Linux only), run:",
        f"  {tunnel_control_command(cfg, n, 'exit')}",
        'It answers "Exit request sent." and the tunnel is closed.',
    ]
    if not detailed:
        return lines + ["", f"More help (is the tunnel running? what if closing fails?): sudo supaspin tunnel {n}"]
    return lines + [
        "",
        "If you want to check whether the background tunnel is running (Mac or Linux only), run:",
        f"  {tunnel_control_command(cfg, n, 'check')}",
        'It answers "Master running" when the tunnel is open.',
        "",
        'If closing says "No such file or directory" but the tunnel still works, or starting a tunnel says',
        '"address already in use" and the check above does not say "Master running", a leftover tunnel is',
        "in the way. To stop it (Mac or Linux only), run:",
        f'  pkill -f "[s]upaspin-{n}.sock"',
        "(The [s] is on purpose: it stops the command from also matching, and stopping, itself.)",
        "Then start the tunnel again.",
    ]


def tunnel_help(cfg, n, pw="<password>"):
    return "\n".join(tunnel_guide(cfg, n, detailed=True) + [
        "",
        "When the tunnel is open, put this in your app's .env:",
        f"  DATABASE_URL=postgresql://postgres:{pw}@localhost:{pg_port(n)}/postgres",
        "  (the password is POSTGRES_PASSWORD in the instance's .env: sudo supaspin env " + str(n) + ")",
        f"Studio (admin UI): http://localhost:{studio_port(n)}",
        f"On the server's own network you do not need a tunnel: use {cfg['lan_ip']}:{pg_port(n)} "
        f"and http://{cfg['lan_ip']}:{studio_port(n)}",
    ])


def fstab_line(img, mount, root):
    return f"{img} {mount} ext4 loop,nofail,x-systemd.requires-mounts-for={root} 0 0"


def fstab_without(text, mount):
    """fstab text with every line whose mount point is `mount` removed."""
    keep = []
    for line in text.splitlines():
        f = line.split()
        if len(f) >= 2 and not line.lstrip().startswith("#") and f[1] == mount:
            continue
        keep.append(line)
    return "\n".join(keep) + "\n"


def fstab_mounts_under(text, prefix):
    """Mount points in fstab that lie below `prefix` (a folder), in file order."""
    out = []
    for line in text.splitlines():
        f = line.split()
        if len(f) >= 2 and not line.lstrip().startswith("#") and f[1].startswith(prefix.rstrip("/") + "/"):
            out.append(f[1])
    return out


def fstab_has(text, mount):
    return any(len(l.split()) >= 2 and not l.lstrip().startswith("#") and l.split()[1] == mount
               for l in text.splitlines())


class _Literal(str):
    pass


def _literal_repr(dumper, data):
    return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|")


yaml.SafeDumper.add_representer(_Literal, _literal_repr)


def strip_kong_services(text, names):
    """Cut whole `- name: X` service blocks out of kong.yml, leaving every other byte as it was.

    Kong's start script substitutes placeholders inside the quoted strings of this file, so it
    must not be parsed and re-written: a changed quoting style breaks the substitution.
    """
    out, skipping = [], False
    for line in text.splitlines(keepends=True):
        m = re.match(r"^  - name: (\S+)\s*$", line)
        if m:
            skipping = m.group(1) in names
        elif skipping and line.strip() and not line.startswith("    "):
            skipping = False
        if not skipping:
            out.append(line)
    return "".join(out)


def post_render(text, n, lan_ip):
    """Edit the rendered chart: fixed host ports on db and studio, no public Studio/pg-meta route."""
    docs = [d for d in yaml.safe_load_all(text) if d]
    for d in docs:
        kind, name = d.get("kind"), d.get("metadata", {}).get("name", "")
        if kind == "StatefulSet" and name == f"{fullname(n)}-db":
            _host_port(d, pg_port(n), lan_ip)
        elif kind == "Deployment" and name == f"{fullname(n)}-studio":
            _host_port(d, studio_port(n), lan_ip)
        elif kind == "ConfigMap" and "kong.yml" in (d.get("data") or {}):
            d["data"]["kong.yml"] = strip_kong_services(d["data"]["kong.yml"], ("meta", "dashboard"))
    for d in docs:  # keep multi-line scripts readable in the release manifest
        for k, v in (d.get("data") or {}).items():
            if isinstance(v, str) and "\n" in v:
                d["data"][k] = _Literal(v)
    return yaml.safe_dump_all(docs, sort_keys=False, default_flow_style=False)


def _host_port(doc, port, ip):
    c = doc["spec"]["template"]["spec"]["containers"][0]
    p = c["ports"][0]
    p["hostPort"] = port
    p["hostIP"] = ip


def pv_manifest(cfg, n, vol):
    return {
        "apiVersion": "v1", "kind": "PersistentVolume",
        "metadata": {"name": f"supaspin-{n}-{vol}",
                     "labels": {"supaspin/instance": str(n)}},
        "spec": {
            "capacity": {"storage": f"{VOLUMES[vol]}Mi"},
            "accessModes": ["ReadWriteOnce"],
            "persistentVolumeReclaimPolicy": "Retain",
            "storageClassName": STORAGE_CLASS,
            "claimRef": {"namespace": namespace(n), "name": pvc_name(n, vol)},
            "local": {"path": f"{cfg['storage_root']}/{n}/{vol}/data"},
            "nodeAffinity": {"required": {"nodeSelectorTerms": [{"matchExpressions": [
                {"key": "kubernetes.io/hostname", "operator": "In",
                 "values": [cfg["node_name"]]}]}]}},
        },
    }


# ------------------------------------------------------------------ system --

def run(args, input_=None, check=True):
    p = subprocess.run(args, input=input_, capture_output=True, text=True)
    if check and p.returncode != 0:
        raise RuntimeError(f"{' '.join(args)}\n{(p.stderr or p.stdout).strip()}")
    return p


def tool(name):
    """kubectl / helm: the microk8s snap commands when present, else plain ones."""
    snap = {"kubectl": "/snap/bin/microk8s.kubectl", "helm": "/snap/bin/microk8s.helm3"}[name]
    return snap if os.path.exists(snap) else (shutil.which(name) or name)


def kubectl(*args, input_=None, check=True):
    return run([tool("kubectl"), *args], input_=input_, check=check)


def helm(*args, check=True):
    return run([tool("helm"), *args], check=check)


def kjson(*args):
    p = kubectl(*args, "-o", "json", check=False)
    return json.loads(p.stdout) if p.returncode == 0 and p.stdout.strip() else None


def apply(manifests):
    docs = manifests if isinstance(manifests, list) else [manifests]
    kubectl("apply", "-f", "-", input_=yaml.safe_dump_all(docs))


def put_secret(manifest):
    """Create a Secret, replacing any old one. Not `kubectl apply`: apply stores a full copy of the
    secret (private key included) in a last-applied annotation."""
    kubectl("delete", "secret", manifest["metadata"]["name"], "-n", manifest["metadata"]["namespace"],
            "--ignore-not-found")
    kubectl("create", "-f", "-", input_=yaml.safe_dump(manifest))


def load_config():
    try:
        with open(CONFIG_PATH) as f:
            cfg = yaml.safe_load(f) or {}
    except FileNotFoundError:
        return None
    cfg["slots"] = {int(k): v for k, v in cfg.get("slots", {}).items()}
    return cfg


def save_config(cfg):
    os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
    data = dict(cfg)
    data["slots"] = {int(k): v for k, v in cfg["slots"].items()}
    tmp = CONFIG_PATH + ".tmp"
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        yaml.safe_dump(data, f, sort_keys=False)
    os.replace(tmp, CONFIG_PATH)


def mounts():
    with open("/proc/mounts") as f:
        return [l.split()[1] for l in f]


def is_mounted(path):
    return os.path.realpath(path) in mounts()


def inst_dir(cfg, n):
    return f"{cfg['storage_root']}/{n}"


def vol_paths(cfg, n, vol):
    d = f"{inst_dir(cfg, n)}/{vol}"
    return f"{d}/{vol}.img", f"{d}/data"


def existing_numbers(cfg):
    """Instances that have any trace: folder, namespace or fstab line."""
    found = set()
    root = cfg["storage_root"]
    if os.path.isdir(root):
        found |= {int(e) for e in os.listdir(root) if e.isdigit()}
    ns = kjson("get", "ns")
    for item in (ns or {}).get("items", []):
        m = re.fullmatch(r"supaspin-(\d+)", item["metadata"]["name"])
        if m:
            found.add(int(m.group(1)))
    try:
        with open("/etc/fstab") as f:
            for m in re.finditer(re.escape(root) + r"/(\d+)/", f.read()):
                found.add(int(m.group(1)))
    except FileNotFoundError:
        pass
    return found


# --------------------------------------------------------------- volumes ---

def provision_volume(cfg, n, vol, log):
    img, mnt = vol_paths(cfg, n, vol)
    root = cfg["storage_root"]
    os.makedirs(mnt, exist_ok=True)
    if not is_mounted(mnt):
        # After a reboot Kubernetes can start a pod before the image is mounted. Nothing may be
        # written into the empty folder meanwhile (Postgres would set up a fresh database there),
        # so it is made read-only until the image is mounted over it. Best effort.
        run(["chattr", "+i", mnt], check=False)
    fresh = not os.path.exists(img)
    if fresh:
        mib = VOLUMES[vol]
        if run(["fallocate", "-l", f"{mib}M", img], check=False).returncode != 0:
            run(["truncate", "-s", f"{mib}M", img])
        run(["/sbin/mkfs.ext4", "-F", "-m", "0", "-L", f"supa{n}-{vol}", img])
    with open("/etc/fstab") as f:
        fstab = f.read()
    if not fstab_has(fstab, mnt):
        with open("/etc/fstab", "a") as f:
            if not fstab.endswith("\n"):
                f.write("\n")
            f.write(fstab_line(img, mnt, root) + "\n")
    run(["systemctl", "daemon-reload"])
    if not is_mounted(mnt):
        run(["mount", mnt])
    # ext4 puts an empty lost+found in every new filesystem. The chart copies its default Postgres
    # config into the pgsodium volume only when that volume is empty, so the folder must go.
    lost = os.path.join(mnt, "lost+found")
    if os.path.isdir(lost) and not os.listdir(lost):
        os.rmdir(lost)
    if vol in DB_OWNED:
        os.chown(mnt, DB_UID, DB_GID)
        os.chmod(mnt, 0o770)
    else:
        os.chmod(mnt, 0o777)
    log(f"  volume {vol}: {VOLUMES[vol]} MiB mounted at {mnt}")


def instance_mounts(cfg, n):
    """Every mount point of the instance found on the system (mounted now or listed in fstab),
    not just the volumes this version creates: an instance made by an older version may have others."""
    d = inst_dir(cfg, n)
    found = {m for m in mounts() if m.startswith(d + "/")}
    try:
        with open("/etc/fstab") as f:
            found |= set(fstab_mounts_under(f.read(), d))
    except FileNotFoundError:
        pass
    return sorted(found)


def remove_mount(mnt, log):
    if is_mounted(mnt):
        r = run(["umount", mnt], check=False)
        if r.returncode != 0:
            raise RuntimeError(f"cannot unmount {mnt}: {r.stderr.strip()}")
    with open("/etc/fstab") as f:
        fstab = f.read()
    if fstab_has(fstab, mnt):
        with open("/etc/fstab", "w") as f:
            f.write(fstab_without(fstab, mnt))
    run(["chattr", "-i", mnt], check=False)
    log(f"  unmounted {mnt}, fstab line removed")


# ------------------------------------------------------------- operations --

def ensure_cluster_objects(cfg, log):
    docs = [
        {"apiVersion": "storage.k8s.io/v1", "kind": "StorageClass",
         "metadata": {"name": STORAGE_CLASS}, "provisioner": "kubernetes.io/no-provisioner",
         "reclaimPolicy": "Retain", "volumeBindingMode": "WaitForFirstConsumer"},
        {"apiVersion": "cert-manager.io/v1", "kind": "ClusterIssuer",
         "metadata": {"name": ISSUER},
         "spec": {"acme": {"server": "https://acme-v02.api.letsencrypt.org/directory",
                           "email": cfg["acme_email"],
                           "privateKeySecretRef": {"name": ISSUER + "-account-key"},
                           "solvers": [{"http01": {"ingress": {"class": cfg["ingress_class"]}}}]}}},
        {"apiVersion": "scheduling.k8s.io/v1", "kind": "PriorityClass",
         "metadata": {"name": PRIORITY_CLASS}, "value": -100,
         "preemptionPolicy": "Never", "globalDefault": False,
         "description": "supaspin test instances: evicted first, never evict others"},
        {"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": CERTS_NS}},
    ]
    for d in docs:
        # Create only what is missing; never overwrite something already there.
        r = kubectl("get", d["kind"].lower(), d["metadata"]["name"], check=False)
        if r.returncode != 0:
            apply(d)
            log(f"  created {d['kind']} {d['metadata']['name']}")
    chart_dir = f"{STATE_DIR}/charts"
    os.makedirs(chart_dir, exist_ok=True)
    tgz = f"{chart_dir}/{CHART_NAME}-{cfg['chart_version']}.tgz"
    if not os.path.exists(tgz):
        log(f"  downloading chart {cfg['chart_version']}")
        for attempt in (1, 2, 3):  # the chart host times out now and then
            try:
                helm("pull", CHART_NAME, "--repo", CHART_REPO, "--version", cfg["chart_version"],
                     "--destination", chart_dir)
                break
            except RuntimeError:
                if attempt == 3:
                    raise
                log(f"  download failed, retrying ({attempt}/3)")
                time.sleep(5)


def chart_path(cfg):
    return f"{STATE_DIR}/charts/{CHART_NAME}-{cfg['chart_version']}.tgz"


def carry_secret(src, ns, name):
    """Copy of a TLS secret for another namespace: data and type, plus the cert-manager.io
    annotations. Without those annotations cert-manager cannot tell which issuer signed the
    certificate, decides it is wrong ("IncorrectIssuer") and orders a new one from Let's Encrypt."""
    keep = {k: v for k, v in (src.get("metadata", {}).get("annotations") or {}).items()
            if k.startswith("cert-manager.io/")}
    meta = {"name": name, "namespace": ns}
    if keep:
        meta["annotations"] = keep
    return {"apiVersion": "v1", "kind": "Secret",
            "type": src.get("type", "kubernetes.io/tls"), "metadata": meta, "data": src["data"]}


def restore_cert(n, log):
    r = kubectl("get", "secret", tls_secret(n), "-n", CERTS_NS, "-o", "json", check=False)
    if r.returncode != 0:
        return
    put_secret(carry_secret(json.loads(r.stdout), namespace(n), tls_secret(n)))
    kubectl("delete", "secret", tls_secret(n), "-n", CERTS_NS, "--ignore-not-found")
    log("  restored saved TLS certificate")


def save_cert(n, log):
    r = kubectl("get", "secret", tls_secret(n), "-n", namespace(n), "-o", "json", check=False)
    if r.returncode != 0:
        return
    s = json.loads(r.stdout)
    if not s.get("data", {}).get("tls.crt"):
        return
    put_secret(carry_secret(s, CERTS_NS, tls_secret(n)))
    log("  kept TLS certificate in namespace " + CERTS_NS)


def pod_summary(n):
    """(ready_count, total, problems) over pods that are meant to keep running."""
    # Only the release's own pods: cert-manager parks a short-lived helper pod in the namespace
    # while it proves domain ownership to Let's Encrypt.
    data = kjson("get", "pods", "-n", namespace(n), "-l", f"app.kubernetes.io/instance={release(n)}") \
        or {"items": []}
    ready = total = 0
    problems = []
    for p in data["items"]:
        if p["status"].get("phase") == "Succeeded":
            continue
        total += 1
        cs = p["status"].get("containerStatuses", [])
        if p["status"].get("phase") == "Running" and cs and all(c.get("ready") for c in cs):
            ready += 1
        else:
            waiting = next((c["state"]["waiting"].get("reason") for c in cs
                            if "waiting" in c.get("state", {})), p["status"].get("phase"))
            problems.append(f"{p['metadata']['name'].replace(fullname(n) + '-', '')}={waiting}")
    return ready, total, problems


def wait_ready(n, log, timeout=1200):
    end = time.time() + timeout
    expected = len(SERVICES)
    while time.time() < end:
        ready, total, problems = pod_summary(n)
        log(f"  pods ready {ready}/{expected}" + (f"  waiting: {', '.join(problems)}" if problems else ""),
            live=True)
        if ready >= expected and total == ready:
            log("")
            return
        time.sleep(5)
    raise RuntimeError("pods did not become ready in time")


def create_instance(cfg, mail, log, number=None):
    taken = existing_numbers(cfg)
    n = number if number is not None else lowest_free_number(cfg["slots"], taken)
    if n is None:
        raise RuntimeError("no free hostname slot: delete an instance or add slots in Settings")
    if n in taken or n not in cfg["slots"]:
        raise RuntimeError(f"instance {n} is not available")
    host = cfg["slots"][n]
    log(f"Creating instance {n} ({host})")
    ensure_cluster_objects(cfg, log)
    try:
        os.makedirs(inst_dir(cfg, n), exist_ok=True)
        for vol in VOLUMES:
            provision_volume(cfg, n, vol, log)
        kubectl("create", "namespace", namespace(n))
        apply([pv_manifest(cfg, n, v) for v in VOLUMES])
        restore_cert(n, log)
        sec = generate_secrets()
        with open(find_template()) as f:
            values = instance_values(cfg, n, sec, mail, f.read())
        vfile = f"{STATE_DIR}/values-{n}.yaml"
        os.makedirs(STATE_DIR, exist_ok=True)
        fd = os.open(vfile, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as f:
            yaml.safe_dump(values, f, sort_keys=False)
        log("  installing chart")
        try:
            helm("install", release(n), chart_path(cfg), "-n", namespace(n), "-f", vfile,
                 "--post-renderer", os.path.realpath(sys.argv[0]),
                 "--post-renderer-args", "post-render", "--post-renderer-args", str(n))
        finally:
            os.remove(vfile)  # holds every secret
        wait_ready(n, log)
        env_path = f"{inst_dir(cfg, n)}/.env"
        fd = os.open(env_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as f:
            f.write(render_env(cfg, n, sec, mail))
        log(f"Instance {n} is ready: https://{host}\n.env written to {env_path}")
        return n
    except Exception:
        log(f"{RED}Create failed.{OFF}")
        if sys.stdin.isatty() and ask(f"Delete the partial instance {n} now? (Y/n)", "y").lower().startswith("y"):
            delete_instance(cfg, n, log)
        else:
            log(f"Run 'supaspin delete {n}' to clean up the partial instance.")
        raise


def delete_instance(cfg, n, log):
    log(f"Deleting instance {n}")
    save_cert(n, log)
    helm("uninstall", release(n), "-n", namespace(n), check=False)
    kubectl("delete", "namespace", namespace(n), "--ignore-not-found", "--wait=true")
    kubectl("delete", "pv", "-l", f"supaspin/instance={n}", "--ignore-not-found")
    for mnt in instance_mounts(cfg, n):
        remove_mount(mnt, log)
    run(["systemctl", "daemon-reload"])
    d = inst_dir(cfg, n)
    for mnt in glob.glob(f"{d}/*/data"):  # also the folders of a create that stopped before mounting
        if not is_mounted(mnt):
            run(["chattr", "-i", mnt], check=False)
    if os.path.isdir(d):
        busy = [m for m in mounts() if m == d or m.startswith(d + "/")]
        if busy:
            raise RuntimeError(f"refusing to delete {d}: still mounted: {', '.join(busy)}")
        shutil.rmtree(d)
    log(f"  removed {d}")
    log(f"Instance {n} deleted.")


def read_env(cfg, n):
    path = f"{inst_dir(cfg, n)}/.env"
    with open(path) as f:
        return f.read()


def memory_use(n):
    """Memory the instance's pods use right now, read from the kernel's per-pod counters
    (the cluster has no metrics server, so `kubectl top` does not work)."""
    pods = kjson("get", "pods", "-n", namespace(n)) or {"items": []}
    total, found = 0, False
    for p in pods["items"]:
        uid = p["metadata"]["uid"]
        for pattern in (f"/sys/fs/cgroup/kubepods/*/pod{uid}/memory.current",
                        f"/sys/fs/cgroup/kubepods/pod{uid}/memory.current",
                        f"/sys/fs/cgroup/kubepods.slice/*/kubepods-*-pod{uid.replace('-', '_')}.slice/memory.current"):
            hits = glob.glob(pattern)
            if hits:
                with open(hits[0]) as f:
                    total += int(f.read()) // 2**20
                found = True
                break
    if not found:
        return "n/a"
    return f"{total}/{total_memory_mib()[1]} MiB"


def disk_use(cfg, n):
    parts = []
    d = inst_dir(cfg, n)
    names = sorted({os.path.basename(os.path.dirname(m)) for m in instance_mounts(cfg, n)} | set(VOLUMES))
    for vol in names:
        mnt = f"{d}/{vol}/data"
        if is_mounted(mnt):
            u = shutil.disk_usage(mnt)
            parts.append(f"{vol} {u.used // 2**20}/{u.total // 2**20}M")
        else:
            parts.append(f"{vol} -")
    return ", ".join(parts)


def list_rows(cfg):
    rows = []
    for n in sorted(existing_numbers(cfg)):
        ready, total, _ = pod_summary(n)
        env = f"{inst_dir(cfg, n)}/.env"
        rows.append({"n": n, "host": cfg["slots"].get(n, "?"), "pods": f"{ready}/{total}",
                     "memory": memory_use(n), "disk": disk_use(cfg, n),
                     "env": env if os.path.exists(env) else "(none)"})
    return rows


# ------------------------------------------------------ copy .env values --

_ASSIGN = re.compile(r"^([ \t]*(?:export[ \t]+)?)([A-Za-z_][A-Za-z0-9_.\-]*)([ \t]*=[ \t]*)(.*)$")
_PAIR = re.compile(r"^[ \t]*(?:export[ \t]+)?([A-Za-z_][A-Za-z0-9_.\-]*)[ \t]*=(.*)$")
_REMOTE = re.compile(r"^(?:([^@:/\s]+)@)?(\[[^\]\s]+\]|[^@:/\s]+):(.+)$")
_WINDOWS_PATH = re.compile(r"^[A-Za-z]:[\\/]")
SSH_ON = [
    "  - Mac: System Settings > General > Sharing > turn on 'Remote Login'.",
    "  - Linux: run  sudo apt install openssh-server  and then  sudo systemctl enable --now ssh",
    "  - Windows: Settings > Apps > Optional features > add 'OpenSSH Server', then in an Administrator",
    "    PowerShell run:  Start-Service sshd; Set-Service sshd -StartupType Automatic",
]


class TargetError(Exception):
    """The target file could not be reached or changed. `why` says what happened, `advice` how to fix it."""

    def __init__(self, why, advice=()):
        super().__init__(why)
        self.why = why
        self.advice = list(advice)


class EnvLine:
    """One logical line of a .env file: a comment, a blank, an assignment (with the following lines of
    a multi-line quoted value) or anything else, which is kept as it is."""
    __slots__ = ("kind", "text", "eol", "head", "key", "value", "more")

    def __init__(self, kind, text, eol, head="", key="", value=""):
        self.kind, self.text, self.eol = kind, text, eol
        self.head, self.key, self.value = head, key, value
        self.more = []  # (text, line ending) of the lines that continue a multi-line value

    def raw(self):
        return self.text + self.eol + "".join(t + e for t, e in self.more)


def _text(data):
    """A .env file's bytes as text (odd bytes are kept as they are), without the byte-order mark."""
    text = data.decode("utf-8", "surrogateescape")
    return text[1:] if text.startswith("\ufeff") else text


def _split_lines(text):
    """(content, line ending) per line; the ending is '' only for a last line that has none."""
    parts = text.split("\n")
    out = []
    for i, part in enumerate(parts):
        if i == len(parts) - 1:
            if part:
                out.append((part, ""))
        elif part.endswith("\r"):
            out.append((part[:-1], "\r\n"))
        else:
            out.append((part, "\n"))
    return out


def _closes(rest, quote):
    """True if `rest` (the text after an opening quote) contains the matching closing quote."""
    i = 0
    while i < len(rest):
        if rest[i] == "\\" and quote != "`":
            i += 2
            continue
        if rest[i] == quote:
            return True
        i += 1
    return False


def scan_env(text):
    """Split .env text into EnvLine objects. A line whose first character (after spaces) is # is a comment
    and is never taken for a key."""
    lines = _split_lines(text)
    out, i = [], 0
    while i < len(lines):
        content, eol = lines[i]
        i += 1
        stripped = content.strip(" \t")
        if not stripped:
            out.append(EnvLine("blank", content, eol))
            continue
        if stripped.startswith("#"):
            out.append(EnvLine("comment", content, eol))
            continue
        m = _ASSIGN.match(content)
        if not m:
            out.append(EnvLine("other", content, eol))
            continue
        line = EnvLine("assign", content, eol, head=m.group(1) + m.group(2) + m.group(3),
                       key=m.group(2), value=m.group(4))
        quote = line.value[:1]
        if quote in ("'", '"', "`") and not _closes(line.value[1:], quote):
            # a quoted value that goes on over the next lines, up to the one with the closing quote;
            # without a closing quote anywhere it is taken as a one-line value
            j = i
            while j < len(lines) and not _closes(lines[j][0], quote):
                j += 1
            if j < len(lines):
                line.more = lines[i:j + 1]
                i = j + 1
        out.append(line)
    return out


def env_pairs(text):
    """The active KEY=VALUE lines of an instance's .env, in file order: [(key, value), ...]."""
    return [(e.key, e.value) for e in scan_env(text) if e.kind == "assign"]


def merge_env(data, pairs):
    """The .env file `data` (bytes) with `pairs` put in: an existing key gets the new value, a missing key
    is added at the end. Every other line stays byte for byte as it was.
    Returns (new bytes, report); the report lists the keys replaced, added and already correct."""
    if b"\x00" in data:
        raise TargetError("the file is not plain text (it looks like it was saved as UTF-16).",
                          ["If you want to fix it: open the file in an editor, save it again as UTF-8 and "
                           "run this again.", "Nothing was changed."])
    entries = scan_env(_text(data))
    wanted = dict(pairs)
    endings = [x for e in entries for x in [e.eol] + [m[1] for m in e.more]]
    eol = "\r\n" if endings.count("\r\n") > endings.count("\n") else "\n"
    status, out = {}, []
    for e in entries:
        if e.kind == "assign" and e.key in wanted:
            same = e.value == wanted[e.key] and not e.more
            status[e.key] = "same" if same and status.get(e.key, "same") == "same" else "replaced"
            out.append(e.raw() if same else e.head + wanted[e.key] + (e.more[-1][1] if e.more else e.eol))
        else:
            out.append(e.raw())
    missing = [k for k in wanted if k not in status]
    if missing:
        if out and not out[-1].endswith("\n"):
            out[-1] += eol
        out += [f"{k}={wanted[k]}{eol}" for k in missing]
    text = ("\ufeff" if data.startswith(b"\xef\xbb\xbf") else "") + "".join(out)
    report = {
        "replaced": [k for k in wanted if status.get(k) == "replaced"],
        "same": [k for k in wanted if status.get(k) == "same"],
        "added": missing,
        "lines": sum(1 + len(e.more) for e in entries),
        "untouched": sum(1 + len(e.more) for e in entries if not (e.kind == "assign" and e.key in wanted)),
        "comments": sum(e.kind == "comment" for e in entries),
    }
    return text.encode("utf-8", "surrogateescape"), report


def _other_pairs(data, wanted):
    """Every active KEY=VALUE line that is not one of the instance's keys. Deliberately simple (one line at a
    time, no multi-line values) so that it does not share the merge's own understanding of the file."""
    out = []
    for line in _text(data).split("\n"):
        line = line.rstrip("\r")
        m = _PAIR.match(line)
        if m and not line.lstrip(" \t").startswith("#") and m.group(1) not in wanted:
            out.append((m.group(1), m.group(2)))
    return out


def _kept(entries, wanted):
    """The lines that are not the instance's, as text. The only change allowed to them is a line ending on the
    very last line: a file without a final newline gets one when keys are added after it."""
    kept = [e.raw() for e in entries if not (e.kind == "assign" and e.key in wanted)]
    if kept:
        kept[-1] = kept[-1].rstrip("\r\n")
    return kept


def check_merge(before, after, pairs):
    """Compare the target file before the change and as it is now. Returns what is wrong (empty = all good):
    nothing but the instance's keys may differ, and each of them must have exactly the instance's value."""
    wanted = dict(pairs)
    b, a = scan_env(_text(before)), scan_env(_text(after))
    problems = []
    if _kept(b, wanted) != _kept(a, wanted):
        problems.append("lines that are not the instance's were changed, moved or removed")
    if _other_pairs(before, wanted) != _other_pairs(after, wanted):
        problems.append("a key that is not the instance's lost or changed its value")
    for key, value in wanted.items():
        found = [e for e in a if e.kind == "assign" and e.key == key]
        if not found:
            problems.append(f"{key} is missing")
        elif any(e.value != value or e.more for e in found):
            problems.append(f"{key} does not have the instance's value")
    return problems


def parse_target(text, home=None):
    """Where the .env file is: '/path', '~/path' or 'user@host:path'.
    Returns {'kind': 'local', 'path'} or {'kind': 'ssh', 'user', 'host', 'path'}."""
    text = text.strip()
    if not text:
        raise ValueError("Nothing was typed.")
    if text == "~" or text.startswith(("/", "./", "../", "~/")):
        return {"kind": "local", "path": home + text[1:] if home and text.startswith("~") else text}
    m = _REMOTE.match(text)
    if not m:
        return {"kind": "local", "path": text}
    user, host, path = m.groups()
    if len(host) == 1 and path[0] == "\\":
        raise ValueError("That looks like a Windows path without the computer's name. "
                         "Write it like this: user@computer:C:/Users/you/app/.env")
    if _WINDOWS_PATH.match(path):
        path = "/" + path.replace("\\", "/")
    return {"kind": "ssh", "user": user, "host": host, "path": path}


def invoking_user():
    """The passwd entry of the person who ran sudo, or None when supaspin runs directly as root."""
    name = os.environ.get("SUDO_USER")
    if os.geteuid() != 0 or not name or name == "root":
        return None
    try:
        return pwd.getpwnam(name)
    except KeyError:
        return None


def private_tmpdir(owner):
    """A folder that only this run can read, in memory (/run) when possible: it holds the safety copy of the
    target file while the change is made."""
    base = "/run" if os.path.isdir("/run") and os.access("/run", os.W_OK) else None
    d = tempfile.mkdtemp(prefix="supaspin-", dir=base)
    if owner:
        os.chown(d, owner.pw_uid, owner.pw_gid)
    return d


def put_private(path, data, owner):
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "wb") as f:
        f.write(data)
    if owner:
        os.chown(path, owner.pw_uid, owner.pw_gid)


def ssh_error(err, spec, me):
    """Turn what scp/ssh printed into a TargetError with advice in plain words."""
    host, port, path = spec["host"], spec["port"], spec["path"]
    user = spec.get("user") or me
    hand = f"  ssh -p {port} {user}@{host}"
    low = err.lower()
    first = next((l.strip() for l in err.splitlines() if l.strip()), "no message")
    if ("identity file" in low and "not accessible" in low) or "load key" in low or "incorrect passphrase" in low:
        return TargetError(f"the SSH key file cannot be used: {first}",
                           [f"If you want to fix it: check that the path is right, that {me} can read the file "
                            "and that only its owner may (chmod 600 <the key file>)."])
    if "could not resolve hostname" in low:
        advice = ["If you want to fix it: use the computer's IP address (for example 192.168.10.50) instead of its "
                  "name, or check the spelling.", "A name that works on your own computer may not be known to this server."]
        if len(host) == 1:
            advice.append(f"If you meant a Windows path such as {host}:/Users/..., write it as "
                          f"user@computer:{host}:/Users/...")
        return TargetError(f"the name '{host}' is not known to this server.", advice)
    if "remote host identification has changed" in low:
        forget = host if port == 22 else f"[{host}]:{port}"
        return TargetError(f"the computer at {host} identifies itself differently than last time.", [
            "This happens when that computer was reinstalled or when a different computer now has that address. "
            "It can also mean that somebody is pretending to be it.",
            f"If you are sure it is the right computer, run this on this server (as {me}, without sudo) to forget "
            "the old fingerprint, then try again:", f"  ssh-keygen -R {forget}"])
    if "host key verification failed" in low:
        return TargetError(f"this server does not trust {host} yet (its fingerprint is not on the list of "
                           "known computers).", [
            f"If you want to trust that computer, run this once on this server (as {me}, without sudo), check the "
            "fingerprint it shows and answer yes:", hand, "Then run this again."])
    if any(s in low for s in ("timed out", "no route to host", "network is unreachable")):
        return TargetError(f"{host} did not answer on port {port} (no connection could be made).", [
            "This usually means this server cannot reach that computer at all.",
            "If you want to fix it, check these in this order:",
            "  1. Is the computer switched on, awake and connected to the network?",
            "  2. Is the address right? Addresses can change - use the computer's current one.",
            "  3. Is SSH switched on there?", *SSH_ON,
            f"  4. Is a firewall or your router blocking port {port}?",
            "To try the connection yourself, run this on this server:", hand])
    if "connection refused" in low:
        return TargetError(f"{host} answered, but nothing is listening for SSH on port {port}.", [
            "If you want to fix it: switch SSH on there, or use the right port number.", *SSH_ON,
            "To try the connection yourself, run this on this server:", hand])
    if "permission denied (" in low or "too many authentication failures" in low or "authentication failed" in low:
        return TargetError(f"{host} refused the login for the user '{user}'.", [
            "Either the user name or the password is wrong, or that computer does not know this server's key.",
            "If you want to log in with a password: run this again in a terminal (not from a script) and type the "
            "password when SSH asks for it.",
            f"If you want this server to log in without a password: run this on this server (as {me}, without "
            "sudo). It asks for the password once and installs the key:", f"  ssh-copy-id -p {port} {user}@{host}",
            "ssh-copy-id does not work for a Windows computer: copy the content of ~/.ssh/id_ed25519.pub from this "
            "server into C:\\Users\\<user>\\.ssh\\authorized_keys there (for an administrator account: "
            "C:\\ProgramData\\ssh\\administrators_authorized_keys).",
            "If the user name is wrong, run this again with the right one: user@computer:/path/.env"])
    if "no such file" in low:
        return TargetError(f"the file {path} does not exist on {host}.",
                           ["If you want to fix it: check the spelling and the folder. Nothing was created."])
    if "permission denied" in low:
        return TargetError(f"the login worked, but user '{user}' is not allowed to open or change {path} on {host}.",
                           ["If you want to fix it: give that user write permission on the file (and read "
                            "permission on its folder), or use the user that owns the file."])
    if "subsystem request failed" in low:
        return TargetError(f"{host} accepted the login but does not allow file copying (SFTP).", [
            "If you want to fix it: on that computer make sure the SSH server settings contain a line starting "
            "with 'Subsystem sftp', then restart SSH (Linux: sudo systemctl restart ssh). Mac and Windows have it "
            "on by default."])
    return TargetError(f"the connection to {host} failed: {first}",
                       ["To see more, try the connection by hand on this server:", hand])


class LocalTarget:
    """A .env file on this server, changed in place so that its owner and permissions stay as they are."""

    def __init__(self, spec):
        self.path = self.label = spec["path"]

    def _error(self, e):
        if isinstance(e, FileNotFoundError):
            return TargetError(f"the file {self.path} does not exist on this server.",
                               ["If you want to fix it: check the spelling and the folder. Nothing was created."])
        if isinstance(e, PermissionError):
            return TargetError(f"this server is not allowed to open {self.path}.")
        return TargetError(f"{self.path} cannot be used: {e.strerror or e}.")

    def read(self):
        try:
            with open(self.path, "rb") as f:
                return f.read()
        except OSError as e:
            raise self._error(e)

    def write(self, data):
        try:
            with open(self.path, "r+b") as f:
                f.write(data)
                f.truncate()
                f.flush()
                os.fsync(f.fileno())
        except OSError as e:
            raise self._error(e)

    def close(self):
        pass


class SshTarget:
    """A .env file on another computer, copied with scp. scp needs nothing on the other side except an SSH
    server with file copying (SFTP), so this works for Mac, Linux and Windows computers. SSH runs as the person
    who typed sudo: their keys and their list of known computers apply. One connection, opened first, is shared
    by all copies, so a password is asked only once."""

    def __init__(self, spec, tmp, owner):
        self.spec, self.tmp, self.owner = spec, tmp, owner
        self.me = owner.pw_name if owner else "root"
        self.dest = f"{spec['user']}@{spec['host']}" if spec.get("user") else spec["host"]
        self.label = f"{self.dest}:{spec['path']}"
        self.sock = os.path.join(tmp, "c")
        self.opts = ["-o", "ConnectTimeout=10", "-o", "ServerAliveInterval=10", "-o", "ServerAliveCountMax=3",
                     "-o", f"Port={spec['port']}"]
        if not sys.stdin.isatty():
            self.opts += ["-o", "BatchMode=yes"]  # never wait for a password nobody can type
        if spec.get("key"):
            self.opts += ["-i", spec["key"], "-o", "IdentitiesOnly=yes"]  # -i (not -o) so ssh warns about a bad path
        self.connected = False

    def _as_owner(self, cmd):
        return ["runuser", "-u", self.owner.pw_name, "--"] + cmd if self.owner else cmd

    def _connect(self):
        """Open the one connection that all copies share. This is where SSH asks for a password or a key
        passphrase, once per run. (scp cannot open a shared connection by itself, so ssh does it.)"""
        if self.connected:
            return
        with tempfile.TemporaryFile("w+") as err:
            p = subprocess.run(self._as_owner(["ssh", "-f", "-N", "-M", "-S", self.sock, *self.opts, self.dest]),
                               stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=err)
            err.seek(0)
            text = err.read()
        if p.returncode != 0:
            raise ssh_error(text, self.spec, self.me)
        self.connected = True

    def _scp(self, src, dst):
        self._connect()
        p = subprocess.run(self._as_owner(["scp", "-o", f"ControlPath={self.sock}", *self.opts, src, dst]),
                           stdin=subprocess.DEVNULL, capture_output=True, text=True)
        if p.returncode != 0:
            raise ssh_error(p.stderr or p.stdout, self.spec, self.me)

    def read(self):
        dst = os.path.join(self.tmp, "download")
        self._scp(f"{self.dest}:{self.spec['path']}", dst)
        with open(dst, "rb") as f:
            data = f.read()
        os.remove(dst)
        return data

    def write(self, data):
        src = os.path.join(self.tmp, "upload")
        put_private(src, data, self.owner)
        try:
            self._scp(src, f"{self.dest}:{self.spec['path']}")
        finally:
            os.remove(src)

    def close(self):
        if os.path.exists(self.sock):
            subprocess.run(self._as_owner(["ssh", "-o", f"ControlPath={self.sock}", "-O", "exit", self.dest]),
                           stdin=subprocess.DEVNULL, capture_output=True)


def open_target(spec, tmp, owner):
    return LocalTarget(spec) if spec["kind"] == "local" else SshTarget(spec, tmp, owner)


def undo_write(target, original):
    """After a failed write: 'unchanged' (the file never changed), 'restored' (put back and confirmed) or
    'failed' (the file may be damaged and could not be put back)."""
    try:
        if target.read() == original:
            return "unchanged"
        target.write(original)
        return "restored" if target.read() == original else "failed"
    except TargetError:
        return "failed"


def _print_changes(log, rep, dry):
    names = ("Would replace", "Would add", "Already correct") if dry else ("Replaced", "Added", "Already correct")
    for name, keys in zip(names, (rep["replaced"], rep["added"], rep["same"])):
        log(f"  {name} ({len(keys)}): {', '.join(keys) if keys else '(none)'}")
    log(f"  Untouched: {rep['untouched']} other lines (other keys, comments, blank lines) "
        f"{'stay' if dry else 'are'} exactly as they {'are' if dry else 'were'}.")


def push_env(cfg, n, spec, log, dry_run=False):
    """Put instance n's .env values into the target file and report what happened. Before anything is
    written, the target is copied to a private temporary folder; after writing, the file is read back and
    compared with that copy. If anything is off, the copy is written back. The copy is deleted at the end,
    except when it could not be written back (then it is the only good copy). Returns 0 or 1."""
    who = f"instance {n} ({cfg['slots'].get(n, '?')})"
    try:
        pairs = env_pairs(read_env(cfg, n))
    except FileNotFoundError:
        log(f"{RED}FAILED{OFF}  {who} has no .env file ({inst_dir(cfg, n)}/.env), so there is nothing to copy.")
        return 1
    owner = invoking_user()
    tmp = private_tmpdir(owner)
    target, written, keep = None, False, False
    try:
        try:
            target = open_target(spec, tmp, owner)
            log(f"Reading {target.label} ...")
            original = target.read()
            put_private(os.path.join(tmp, "original.env"), original, None)
            new, rep = merge_env(original, pairs)
            log(f"  ok - {rep['lines']} lines, {rep['comments']} commented-out lines are ignored")
            log(f"Working out the changes: {len(rep['replaced'])} to replace, {len(rep['added'])} to add, "
                f"{len(rep['same'])} already correct")
            if dry_run:
                log(f"{BOLD}DRY RUN{OFF}  {who} -> {target.label}   Nothing was written.")
                _print_changes(log, rep, dry=True)
                return 0
            if new == original:
                log(f"{BOLD}SUCCESS{OFF}  {who} -> {target.label}")
                log("  Nothing to change: every value in the file was already correct. The file was not touched.")
                return 0
            log("Writing the file ...")
            written = True  # from here on the file may be damaged, even if the write itself fails
            target.write(new)
            log("Reading it back and checking that nothing else changed ...")
            back = target.read()
            problems = check_merge(original, back, pairs)
            if not problems and back != new:
                problems = ["the file that arrived is not the one that was sent"]
            if problems:
                raise TargetError("the check after writing failed: " + "; ".join(problems) + ".")
        except (TargetError, KeyboardInterrupt) as e:
            err = e if isinstance(e, TargetError) else TargetError("it was interrupted (Ctrl+C).")
            state = undo_write(target, original) if written else "unchanged"
            keep = state == "failed"
            log(f"{RED}FAILED{OFF}  {who} was NOT written to {target.label}")
            log(f"  Why: {err.why}")
            if state == "unchanged":
                log("  Your file was not changed.")
            elif state == "restored":
                log("  Your file was restored to exactly how it was. Nothing was lost.")
            else:
                log(f"  {RED}WARNING: the file may be damaged and could not be put back.{OFF}")
                log(f"  An untouched copy of the original is kept on this server: {tmp}/original.env")
                log(f"  Copy it back to {target.label} by hand, then delete that folder: sudo rm -r {tmp}")
            for line in err.advice:
                log(f"  {line}" if line else "")
            return 1
        log(f"{BOLD}SUCCESS{OFF}  {who} -> {target.label}")
        _print_changes(log, rep, dry=False)
        return 0
    finally:
        if target is not None:
            target.close()
        if not keep:
            shutil.rmtree(tmp, ignore_errors=True)


def ask_target(user):
    """Ask where the .env file is and how to log in there. Returns a target spec, or None."""
    print("Which .env file should receive the values? Type where it is:")
    print("  on this server:               /home/igor/app/.env")
    print("  on a Mac or Linux computer:   igor@192.168.10.50:/Users/igor/app/.env")
    print("  on a Windows computer:        igor@192.168.10.50:C:/Users/igor/app/.env")
    print("The file must already exist. Only the instance's own keys are changed; everything else stays.\n")
    while True:
        text = ask("File (Enter to cancel)")
        if not text:
            return None
        try:
            spec = parse_target(text, user.pw_dir if user else None)
            break
        except ValueError as e:
            print(f"{RED}{e}{OFF}")
    if spec["kind"] == "ssh":
        if not spec["user"]:
            spec["user"] = ask("SSH user on that computer", user.pw_name if user else "root")
        try:
            spec["port"] = int(ask("SSH port", "22"))
        except ValueError:
            print(f"{RED}The port must be a number.{OFF}")
            return None
        spec["key"] = ask("SSH key file (press Enter to use your normal keys, or to type a password when asked)") or None
    return spec


def push_env_command(cfg, args):
    ap = argparse.ArgumentParser(prog="supaspin push-env", description="Put an instance's .env values into "
                                 "another .env file: existing keys get the new value, missing keys are added.")
    ap.add_argument("number", type=int, help="instance number (see: supaspin list)")
    ap.add_argument("target", nargs="?", help="/path/.env on this server, or user@host:/path/.env on another "
                    "computer (Windows: user@host:C:/path/.env). Put it in quotes. Asked for if left out.")
    ap.add_argument("--port", type=int, default=22, help="SSH port of the other computer (default 22)")
    ap.add_argument("--key", help="SSH key file to log in with (default: your normal keys, or a password)")
    ap.add_argument("--dry-run", action="store_true", help="show what would change, write nothing")
    a = ap.parse_args(args)
    if a.number not in existing_numbers(cfg):
        print(f"{RED}Instance {a.number} does not exist.{OFF} See: sudo supaspin list", file=sys.stderr)
        return 1
    user = invoking_user()
    if a.target:
        try:
            spec = parse_target(a.target, user.pw_dir if user else None)
        except ValueError as e:
            print(f"{RED}{e}{OFF}", file=sys.stderr)
            return 1
        if spec["kind"] == "ssh":
            spec.update(port=a.port, key=a.key)
    elif sys.stdin.isatty():
        spec = ask_target(user)
        if spec is None:
            return 1
    else:
        print(f"{RED}Say where the file is: supaspin push-env {a.number} user@host:/path/.env{OFF}", file=sys.stderr)
        return 1
    return push_env(cfg, a.number, spec, log_line, a.dry_run)


# --------------------------------------------------------------------- ui --

def log_line(msg, live=False):
    if live and sys.stdout.isatty():
        sys.stdout.write("\r\x1b[K" + msg)
        sys.stdout.flush()
    else:
        print(msg, flush=True)


def ask(prompt, default=None):
    suffix = f" [{default}]" if default not in (None, "") else ""
    try:
        val = input(f"{prompt}{suffix}: ").strip()
    except EOFError:
        val = ""
    return val or (default if default is not None else "")


def confirm_number(n, what):
    return ask(f"Type the instance number ({n}) to confirm {what}") == str(n)


def menu(title, items):
    """Arrow-key menu; returns index or None (q / Esc). Falls back to numbers without a terminal."""
    if not sys.stdin.isatty():
        print(title)
        for i, it in enumerate(items, 1):
            print(f"  {i}. {it}")
        v = ask("Choose")
        return int(v) - 1 if v.isdigit() and 0 < int(v) <= len(items) else None
    fd = sys.stdin.fileno()
    saved = termios.tcgetattr(fd)
    sel = 0
    try:
        tty.setraw(fd)
        while True:
            sys.stdout.write("\x1b[H\x1b[2J" + BOLD + title + OFF + "\r\n\r\n")
            for i, it in enumerate(items):
                sys.stdout.write((f"  {REV} {it} {OFF}" if i == sel else f"   {it}") + "\r\n")
            sys.stdout.write(DIM + "\r\n up/down move, Enter choose, q back" + OFF + "\r\n")
            sys.stdout.flush()
            ch = os.read(fd, 8)
            if ch in (b"\x1b[A", b"k"):
                sel = (sel - 1) % len(items)
            elif ch in (b"\x1b[B", b"j"):
                sel = (sel + 1) % len(items)
            elif ch in (b"\r", b"\n"):
                return sel
            elif ch in (b"q", b"\x1b", b"\x03"):
                return None
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, saved)


def pause():
    if sys.stdin.isatty():
        input("\nPress Enter to go back.")


def pick_instance(cfg, title):
    nums = sorted(existing_numbers(cfg))
    if not nums:
        print("No instances.")
        return None
    i = menu(title, [f"{n}  {cfg['slots'].get(n, '?')}" for n in nums])
    return None if i is None else nums[i]


def mail_dialog(cfg):
    m = dict(cfg.get("mail") or DEFAULT_MAIL)
    for key, label in (("host", "SMTP host"), ("port", "SMTP port"), ("user", "SMTP user"),
                       ("password", "SMTP password"), ("admin_email", "Admin e-mail"),
                       ("sender_name", "Sender name")):
        m[key] = ask(label, m.get(key, ""))
    m["port"] = int(m["port"])
    return m


def choose_mail(cfg):
    saved = cfg.get("mail") or DEFAULT_MAIL
    print(f"Saved mail settings: {saved['host']}:{saved['port']} as {saved['user'] or '(no user)'}")
    if ask("Use the saved mail settings? (Y/n)", "y").lower().startswith("n"):
        return mail_dialog(cfg)
    return saved


def detect_defaults():
    d = {"node_name": "", "lan_ip": "", "ssh_user": os.environ.get("SUDO_USER", "root")}
    try:
        d["node_name"] = kubectl("get", "nodes", "-o", "jsonpath={.items[0].metadata.name}").stdout.strip()
    except Exception:
        pass
    try:
        m = re.search(r"src (\S+)", run(["ip", "-4", "route", "get", "1.1.1.1"]).stdout)
        d["lan_ip"] = m.group(1) if m else ""
    except Exception:
        pass
    return d


def setup_dialog(cfg, log):
    d = detect_defaults()
    cfg = cfg or {}
    cfg["storage_root"] = ask("Storage root (images live here)", cfg.get("storage_root", "/home/igor/k8s-storage/supabase"))
    cfg["node_name"] = ask("Kubernetes node name", cfg.get("node_name", d["node_name"]))
    cfg["lan_ip"] = ask("Server LAN IP (Postgres/Studio ports bind here)", cfg.get("lan_ip", d["lan_ip"]))
    cfg["acme_email"] = ask("Let's Encrypt e-mail", cfg.get("acme_email", ""))
    cfg["ingress_class"] = ask("Ingress class", cfg.get("ingress_class", "public"))
    cur = ",".join(cfg["slots"][k] for k in sorted(cfg.get("slots", {}))) if cfg.get("slots") else ""
    hosts = [h.strip() for h in ask("Hostname slots, comma separated (first = instance 1)", cur).split(",") if h.strip()]
    cfg["slots"] = {i + 1: h for i, h in enumerate(hosts)}
    cfg["ssh_host"] = ask("Public SSH host for tunnels", cfg.get("ssh_host", ""))
    cfg["ssh_port"] = int(ask("Public SSH port", cfg.get("ssh_port", 22)))
    cfg["ssh_user"] = ask("SSH user", cfg.get("ssh_user", d["ssh_user"]))
    cfg["chart_version"] = cfg.get("chart_version", CHART_VERSION)
    print("Default mail settings:")
    cfg["mail"] = mail_dialog(cfg)
    save_config(cfg)
    ensure_cluster_objects(cfg, log)
    log("Settings saved to " + CONFIG_PATH)
    return cfg


def show_menu(cfg, n):
    while True:
        i = menu(f"Instance {n} - {cfg['slots'].get(n, '?')}",
                 [".env", ".env (secrets shown)", "Tunnel command", "Pods and events", "Back"])
        if i is None or i == 4:
            return
        print()
        try:
            if i in (0, 1):
                text = read_env(cfg, n)
                print(f"{inst_dir(cfg, n)}/.env\n\n" + (text if i == 1 else mask_env(text)))
            elif i == 2:
                print(tunnel_help(cfg, n))
            else:
                print(kubectl("get", "pods", "-n", namespace(n), "-o", "wide").stdout)
                print(kubectl("get", "events", "-n", namespace(n), "--sort-by=.lastTimestamp").stdout[-3000:])
        except Exception as e:
            print(f"{RED}{e}{OFF}")
        pause()


def print_list(cfg):
    rows = list_rows(cfg)
    if not rows:
        print("No instances.")
        return
    for r in rows:
        print(f"{BOLD}{r['n']}{OFF}  {r['host']}  pods {r['pods']}  memory {r['memory']}")
        print(f"     disk {r['disk']}\n     env  {r['env']}")
        for line in tunnel_guide(cfg, r["n"]):
            print(f"     {line}" if line else "")
        print()


def main_menu(cfg, log):
    while True:
        i = menu(f"supaspin {VERSION}", ["Create instance", "List instances", "Show instance",
                                          "Delete instance", "Copy instance values into another .env file",
                                          "Mail settings", "Settings", "Quit"])
        if i is None or i == 7:
            return
        print()
        try:
            if i == 0:
                create_instance(cfg, choose_mail(cfg), log)
            elif i == 1:
                print_list(cfg)
            elif i == 2:
                n = pick_instance(cfg, "Show which instance?")
                if n:
                    show_menu(cfg, n)
                    continue
            elif i == 3:
                n = pick_instance(cfg, "Delete which instance?")
                if n and confirm_number(n, "deleting it"):
                    delete_instance(cfg, n, log)
            elif i == 4:
                n = pick_instance(cfg, "Copy which instance's values?")
                spec = ask_target(invoking_user()) if n else None
                if spec:
                    print()
                    push_env(cfg, n, spec, log)
            elif i == 5:
                cfg["mail"] = mail_dialog(cfg)
                save_config(cfg)
            elif i == 6:
                cfg = setup_dialog(cfg, log)
        except Exception as e:
            print(f"{RED}{e}{OFF}")
        pause()


def main(argv):
    if len(argv) >= 2 and argv[1] == "post-render":  # called by Helm; no root check, no output but YAML
        cfg = load_config()
        sys.stdout.write(post_render(sys.stdin.read(), int(argv[2]), cfg["lan_ip"]))
        return 0
    if len(argv) > 1 and argv[1] in ("help", "-h", "--help"):
        print(__doc__)
        return 0
    if os.geteuid() != 0:
        print("supaspin must run as root: sudo supaspin", file=sys.stderr)
        return 1
    cmd = argv[1] if len(argv) > 1 else "menu"
    cfg = load_config()
    if cmd in ("--version", "version"):
        print(f"supaspin {VERSION}")
        return 0
    try:
        if cfg is None or cmd == "setup":
            print("First-time setup." if cfg is None else "Settings.")
            cfg = setup_dialog(cfg, log_line)
            if cmd == "setup":
                return 0
        n = int(argv[2]) if len(argv) > 2 and argv[2].isdigit() else None
        if cmd == "menu":
            main_menu(cfg, log_line)
        elif cmd == "create":
            mail = cfg.get("mail") or DEFAULT_MAIL
            if "--ask-mail" in argv:
                mail = choose_mail(cfg)
            create_instance(cfg, mail, log_line)
        elif cmd == "list":
            print_list(cfg)
        elif cmd == "show" and n:
            print(mask_env(read_env(cfg, n)))
            print("\n" + tunnel_help(cfg, n))
        elif cmd == "env" and n:
            print(read_env(cfg, n), end="")
        elif cmd == "tunnel" and n:
            print(tunnel_help(cfg, n))
        elif cmd == "delete" and n:
            if "--yes" in argv or confirm_number(n, "deleting it"):
                delete_instance(cfg, n, log_line)
        elif cmd == "push-env":
            return push_env_command(cfg, argv[2:])
        elif cmd == "mail":
            cfg["mail"] = mail_dialog(cfg)
            save_config(cfg)
        else:
            print(__doc__)
            return 2
    except (RuntimeError, KeyboardInterrupt) as e:
        print(f"{RED}{e}{OFF}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))
