Das-OS/srcpkgs/deployer/files/python/deployer/pipeline.py

"""Deploy steps: preview() builds a Plan (nothing changes on the server), execute() applies it."""
from __future__ import annotations

import os
import glob
import shlex
import subprocess
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

from . import kinds, remote, snapshots
from .config import App, Config
from .kinds import Options, Plan



def _history_path() -> Path:
    """history.log next to the sources (a checkout), else ~/.local/state/deployer/ (an installed package)."""
    local = Path(__file__).resolve().parent.parent
    if os.access(local, os.W_OK):
        return local / "history.log"
    return Path(os.environ.get("XDG_STATE_HOME") or Path.home() / ".local/state") / "deployer" / "history.log"


HISTORY = _history_path()


class DeployError(RuntimeError):
    pass


@dataclass
class Result:
    ok: bool
    snapshot: str | None
    message: str


_PROBE = r'''
if [ -d "$DEST" ]; then
    echo "EXISTS d"
    find "$DEST" -mindepth 1 -maxdepth 1 -printf 'E %y %f\n'
    find "$DEST" -mindepth 2 -maxdepth 2 -type l -printf 'L %P\n'
elif [ -e "$DEST" ]; then echo "EXISTS f"
else echo "EXISTS -"
fi
'''


def git_rev(src: Path) -> str:
    try:
        rev = subprocess.run(["git", "-C", str(src), "rev-parse", "--short", "HEAD"], capture_output=True,
                             text=True, timeout=10).stdout.strip()
        if not rev:
            return ""
        dirty = subprocess.run(["git", "-C", str(src), "status", "--porcelain"], capture_output=True, text=True,
                               timeout=10).stdout.strip()
        return rev + ("-dirty" if dirty else "")
    except (OSError, subprocess.SubprocessError):
        return ""


def find_artifact(app: App) -> Path:
    files = [Path(p) for p in glob.glob(str(app.src_path / app.artifact)) if Path(p).is_file()]
    if not files:
        raise DeployError(f"no file matches {app.artifact} in {app.src_path}")
    return max(files, key=lambda p: p.stat().st_mtime)


def probe(cfg: Config, dest: str) -> tuple[str, dict[str, str], list[str]]:
    """(d|f|-, top-level entries, symlinks two levels deep)."""
    out = remote.run(cfg.server, _PROBE, DEST=dest)
    exists, entries, links = "-", {}, []
    for line in out.splitlines():
        tag, _, rest = line.partition(" ")
        if tag == "EXISTS":
            exists = rest
        elif tag == "E":
            t, _, name = rest.partition(" ")
            entries[name] = "l" if t == "l" else ("d" if t == "d" else "f")
        elif tag == "L":
            links.append(rest)
    return exists, entries, links


def preview(cfg: Config, app: App, opts: Options, log=print) -> Plan:
    kind = kinds.get(app.kind)
    src = app.src_path
    if not src.exists():
        raise DeployError(f"local source not found: {src}")
    if opts.build and app.build:
        log(f"== build: {app.build}")
        remote.run_local(app.build, src if src.is_dir() else src.parent, log, app.build_env)

    dest = cfg.dest_path(app)
    artifact = None
    if app.kind == "artifact":
        artifact = find_artifact(app)
        if app.dest.endswith("/"):
            dest = f"{dest}/{artifact.name}"
        rsync_src = str(artifact)
    elif src.is_file():
        if app.dest.endswith("/"):
            dest = f"{dest}/{src.name}"
        rsync_src = str(src)
    else:
        rsync_src = f"{src}/"
    is_file = rsync_src != f"{src}/"

    log(f"== probe {cfg.server.target}:{dest}")
    exists, entries, links = probe(cfg, dest)
    if exists == ("d" if is_file else "f"):
        raise DeployError(f"{dest} is a {'directory' if exists == 'd' else 'file'} on the server")
    first = exists == "-"
    if first and app.kind == "python-update":
        raise DeployError(f"{dest} does not exist: deploy it with kind python-service first")

    protected = set() if is_file else {*kind.protected, *app.configs, *links,
                                       *(n for n, t in entries.items() if t == "l")}
    filters = [] if is_file else remote.build_filters(protected=protected,
                                                      exclude=[*cfg.server.dev_exclude, *app.exclude],
                                                      use_gitignore=app.use_gitignore)
    delete = (opts.delete if opts.delete is not None else app.delete) and not is_file
    extra = [f"--chmod=F{app.mode}"] if app.mode else []
    plan = Plan(cfg=cfg, app=app, opts=opts, dest=dest, first_install=first, rsync_src=rsync_src,
                filters=filters, delete=delete, extra=extra, remote_entries=entries, artifact_file=artifact,
                git_rev=git_rev(src if src.is_dir() else src.parent))

    rsync_dest = dest if is_file else f"{dest}/"
    plan.changes = remote.rsync(cfg.server, rsync_src, rsync_dest, filters=filters, delete=delete,
                                 extra=extra, dry_run=True)

    if first:
        plan.notes.append(f"first install: {dest} will be created")
    if artifact:
        plan.notes.append(f"artifact: {artifact.relative_to(src)} ({artifact.stat().st_size // 1024} KiB, "
                          f"{datetime.fromtimestamp(artifact.stat().st_mtime):%Y-%m-%d %H:%M})")
    if not is_file and opts.copy_configs:
        for name, target in kind.config_items(plan):
            state = entries.get(name)
            if state is None:
                local = kind.local_config(plan, name)
                plan.notes.append(f"config {name}: link -> {target}"
                                  + (f" (upload {local.name} if missing)" if local.exists() else " (no local copy)"))
            elif state != "l":
                plan.notes.append(f"config {name}: real {'dir' if state == 'd' else 'file'} on the server, kept")
    deleted = sum(c.action == "delete" for c in plan.changes)
    if deleted:
        plan.notes.append(f"{deleted} file(s) on the server will be DELETED")
    if plan.git_rev.endswith("-dirty"):
        plan.notes.append("working tree has uncommitted changes")
    return plan


def _record(plan: Plan, action: str, result: str, snap: str | None, log) -> None:
    ts = datetime.now().isoformat(timespec="seconds")
    line = f"{ts} {plan.app.name} {action} {plan.app.kind} {result} rev={plan.git_rev or '-'} snap={snap or '-'}"
    try:
        remote.run(plan.server, 'mkdir -p "$STATE" && printf "%s\\n" "$LINE" >> "$STATE/deploy.log"',
                   STATE=plan.server.state_dir, LINE=line)
    except remote.RemoteError as e:
        log(f"WARN: deploy.log not written: {e}")
    HISTORY.parent.mkdir(parents=True, exist_ok=True)
    with open(HISTORY, "a") as f:
        f.write(line + "\n")


def execute(plan: Plan, log=print) -> Result:
    kind = kinds.get(plan.app.kind)
    is_file = plan.rsync_src != f"{plan.app.src_path}/"
    snap = None
    try:
        if not plan.first_install:
            added = [c.path for c in plan.changes if c.action == "new" and not c.path.endswith("/")]
            snap = snapshots.create(plan.server, plan.app.name, plan.dest, [] if is_file else added, log)
            log(f"== snapshot {snap}")
        log("== sync")
        rsync_dest = plan.dest if is_file else f"{plan.dest}/"
        remote.rsync(plan.server, plan.rsync_src, rsync_dest, filters=plan.filters, delete=plan.delete,
                     extra=plan.extra, log=log)
        if not is_file and plan.opts.copy_configs:
            kinds.link_configs(plan, log)
        log("== install")
        kind.install(plan, log)
        ok = True
        if plan.opts.restart:
            log("== health")
            ok = kind.health(plan, log)
    except (remote.RemoteError, DeployError) as e:
        log(f"ERROR: {e}")
        _record(plan, "deploy", "FAILED", snap, log)
        return Result(False, snap, str(e))
    _record(plan, "deploy", "ok" if ok else "UNHEALTHY", snap, log)
    msg = "deployed" if ok else "deployed, but the health check failed"
    log(f"== {msg}" + (f" (rollback: snapshot {snap})" if snap and not ok else ""))
    return Result(ok, snap, msg)


def base_plan(cfg: Config, app: App, opts: Options | None = None) -> Plan:
    """Plan without a preview: for rollback, status and logs."""
    dest = cfg.dest_path(app)
    if app.kind == "artifact" and app.dest.endswith("/"):
        dest = f"{dest}/{find_artifact(app).name}"
    elif app.kind == "files" and app.src_path.is_file() and app.dest.endswith("/"):
        dest = f"{dest}/{app.src_path.name}"
    return Plan(cfg=cfg, app=app, opts=opts or Options(), dest=dest, first_install=False, rsync_src="",
                filters=[], delete=False, git_rev=git_rev(app.src_path if app.src_path.is_dir() else app.src_path.parent))


def rollback(cfg: Config, app: App, ts: str | None = None, log=print) -> Result:
    kind = kinds.get(app.kind)
    plan = base_plan(cfg, app)
    snaps = snapshots.listing(cfg.server, app.name)
    if not snaps:
        return Result(False, None, "no snapshots")
    ts = ts or snaps[0]
    if ts not in snaps:
        return Result(False, None, f"no snapshot {ts}")
    is_file = kind.dest_is_file or app.src_path.is_file()
    try:
        log(f"== restore {ts}")
        snapshots.restore(cfg.server, app.name, plan.dest, ts, is_file, log)
        kind.after_restore(plan, log)
        ok = kind.health(plan, log) if plan.opts.restart else True
    except (remote.RemoteError, DeployError) as e:
        log(f"ERROR: {e}")
        _record(plan, "rollback", "FAILED", ts, log)
        return Result(False, ts, str(e))
    _record(plan, "rollback", "ok" if ok else "UNHEALTHY", ts, log)
    return Result(ok, ts, f"rolled back to {ts}" + ("" if ok else ", health check failed"))


def code_state(local: str, deployed: str) -> str:
    """Local git rev vs the deployed one: same | changed | dirty (same commit, uncommitted changes) | ? (unknown)."""
    if not local or not deployed or deployed == "-":
        return "?"
    base = local.removesuffix("-dirty")
    if base != deployed.removesuffix("-dirty"):
        return "changed"
    return "same" if local == deployed == base else "dirty"


def deployed_rev(fields: list[str]) -> str:
    """Rev from a deploy.log line; "" when the server code is not known to match it (failed deploy, rollback)."""
    if len(fields) >= 6 and fields[2] == "deploy" and fields[4] in ("ok", "UNHEALTHY") \
            and fields[5].startswith("rev="):
        return fields[5][4:]
    return ""


def status_all(cfg: Config) -> dict[str, tuple[str, str, str]]:
    """name -> (state, last deploy line, code state vs local) in one ssh call."""
    parts = []
    for a in cfg.apps:
        k = kinds.get(a.kind)
        dest = cfg.dest_path(a)
        parts.append(f"DEST={shlex.quote(dest)} UNIT={shlex.quote(a.unit)} C={shlex.quote(a.health_container)}\n"
                     f"printf 'S %s ' {shlex.quote(a.name)}; ( {k.status_script()} ) | head -1")
    parts.append(f'[ -f {shlex.quote(cfg.server.state_dir)}/deploy.log ] && '
                 f'tail -n 1000 {shlex.quote(cfg.server.state_dir)}/deploy.log | sed "s/^/D /"')
    out = remote.run(cfg.server, "\n".join(parts) + "\ntrue\n", check=False)
    states, last, deployed = {}, {}, {}
    for line in out.splitlines():
        if line.startswith("S "):
            _, name, *rest = line.split(" ", 2)
            states[name] = rest[0].strip() if rest else "?"
        elif line.startswith("D "):
            f = line[2:].split()
            if len(f) >= 5:
                last[f[1]] = f"{f[0][:16].replace('T', ' ')} {f[2]} {f[4]}"
                deployed[f[1]] = deployed_rev(f)
    return {a.name: (states.get(a.name, "?"), last.get(a.name, ""),
                     code_state(git_rev(a.src_path if a.src_path.is_dir() else a.src_path.parent),
                                deployed.get(a.name, "")))
            for a in cfg.apps}


def logs_command(cfg: Config, app: App) -> str:
    return kinds.get(app.kind).logs_command(cfg, app, cfg.dest_path(app))