#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///
"""Boot a comet and hand you your ship, logged in.

    uv run https://outbox.willhanlen.com/try-hawk.py

One file, stdlib only, no arguments. It fetches a runtime, mines an
identity from a pill that already holds hawk, signs you in, and sends
the browser to your ship. Run it again and it goes straight there.

State lives in one directory (see STATE_DIR). Delete that directory
and the next run starts a new comet.
"""

import http.server
import json
import os
import platform
import pty
import re
import select
import signal
import socket
import subprocess
import sys
import tarfile
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
from pathlib import Path

SERVER_PORT = 6060          # the watch page; fixed, so the installer is one url
SHIP_PORT = 6767            # the ship; fixed, so a bookmark keeps working
PILL = "https://outbox.willhanlen.com/hawk.pill"

VERE_TAG = "vere-v4.6"
VERE_ASSETS = {
    ("Darwin", "arm64"): "macos-aarch64.tgz",
    ("Darwin", "x86_64"): "macos-x86_64.tgz",
    ("Linux", "aarch64"): "linux-aarch64.tgz",
    ("Linux", "arm64"): "linux-aarch64.tgz",
    ("Linux", "x86_64"): "linux-x86_64.tgz",
}
VERE_URL = f"https://github.com/urbit/vere/releases/download/{VERE_TAG}/"

# the page and the ship must be reached under the SAME host name. Eyre's
# session cookie carries no Domain, and cookies ignore the port — so a cookie
# this server sets for 127.0.0.1 is sent to the ship, and one set under the
# name localhost is not.
HOST = "127.0.0.1"
PAGE_URL = f"http://{HOST}:{SERVER_PORT}/"
SHIP_URL = f"http://{HOST}:{SHIP_PORT}/"

ANSI = re.compile(rb"\x1b\[[0-9;?]*[a-zA-Z]|\x1b[()][B0]|\x1b[=>]|\r")
PROMPT_PREFIX = re.compile(r"^(?:~[a-z_-]+:dojo> ?)+")
# the runtime narrates its boot in lines that open with a lowercase vane name.
# anything else at the terminal is a noun it dumped — brackets, indentation,
# spinner residue — and none of it is about the boot.
NARRATION = re.compile(r"^[a-z0-9%]")
PREFIX = re.compile(r"^([a-z-]{2,12}):")
# a comet @p is eight blocks, and the two halves are joined by a DOUBLE
# hyphen. Matching single hyphens only takes the first half and looks right.
COMET_RE = re.compile(r"found comet ~([a-z]{6}(?:-{1,2}[a-z]{6})+)")
PROMPT_RE = re.compile(rb":dojo>")
# the code on a line of its own, behind however many prompts dill painted.
# anchored, because a bare four-block match also fits half a comet name and
# every kiln line.
CODE_RE = re.compile(rb"^(?:~[a-z_-]+:dojo> )*([a-z]{6}(?:-[a-z]{6}){3})[ \t]*$", re.M)
READY_GRACE = 8.0           # seconds to keep serving after the page is told


def state_dir() -> Path:
    if sys.platform == "darwin":
        return Path.home() / "Library/Application Support/hawk"
    return Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local/share")) / "hawk"


STATE_DIR = state_dir()
PIER = STATE_DIR / "pier"
VERE = STATE_DIR / "vere" / "urbit"
LOG = STATE_DIR / "boot.log"
SHIP_JSON = STATE_DIR / "ship.json"


def read_ship() -> dict:
    #
    try:
        return json.loads(SHIP_JSON.read_text())
    except (OSError, ValueError):
        return {}


def write_ship(**fields) -> None:
    #
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    SHIP_JSON.write_text(json.dumps({**read_ship(), **fields}, indent=2) + "\n")


def say(line: str) -> None:
    # our own notes go in the runtime's log, so the story the page tails reads
    # in one order.
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    with LOG.open("a") as f:
        f.write(line.rstrip() + "\n")


def listening(port: int) -> bool:
    # is anything holding the port? A tcp connect, which a busy ship answers
    # instantly — unlike an http request, which queues behind its events.
    with socket.socket() as s:
        s.settimeout(0.5)
        return s.connect_ex((HOST, port)) == 0


def answering(port: int, timeout: float = 20.0) -> bool:
    # has eyre started serving? Any status is an answer. The timeout is long on
    # purpose: a ship mid-boot binds the port well before it can reply, and a
    # short deadline reads that as "not up yet" forever.
    try:
        urllib.request.urlopen(f"http://{HOST}:{port}/", timeout=timeout).close()
        return True
    except urllib.error.HTTPError:
        return True
    except Exception:
        return False


def alive(pid: int | None) -> bool:
    #
    if not pid:
        return False
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False


def log_tail(lines: int = 14, raw: bool = False) -> list[str]:
    # the boot's own narration. The ship also dumps whole nouns at the terminal
    # — an azimuth error is a hundred lines of brackets — and on a page that is
    # four inches tall those bury the one line worth reading. `raw` keeps
    # everything, for when the thing to explain IS the dump.
    try:
        with LOG.open("rb") as f:
            # the ship appends here for as long as it runs, and the page reads
            # this every poll. The end is the only part anyone wants.
            f.seek(max(0, f.seek(0, os.SEEK_END) - 65536))
            text = f.read().decode(errors="replace")
    except OSError:
        return []
    out: list[str] = []
    for line in text.split("\n"):
        line = line.rstrip().replace("\r", "")
        if raw:
            if line.strip():
                out.append(line)
            continue
        line = PROMPT_PREFIX.sub("", line)
        if not line or not NARRATION.match(line):
            continue
        # a run of lines from one vane says one thing — every czar it resolved,
        # every desk it installed. Keep the latest and let it stand for the run.
        prefix = PREFIX.match(line)
        if prefix and out and PREFIX.match(out[-1]) and \
                PREFIX.match(out[-1]).group(1) == prefix.group(1):
            out[-1] = line
            continue
        out.append(line)
    return out[-lines:]


def mined_comet() -> str | None:
    #
    try:
        found = COMET_RE.search(LOG.read_text(errors="replace"))
    except OSError:
        return None
    return found.group(1) if found else None


def fetch_vere(note) -> Path:
    # download the runtime once. The release asset is a tarball holding exactly
    # one file, the binary, named for its version.
    if VERE.exists():
        return VERE
    key = (platform.system(), platform.machine())
    asset = VERE_ASSETS.get(key)
    if not asset:
        raise RuntimeError(f"no urbit runtime published for {key[0]} {key[1]}")
    note(f"downloading {VERE_TAG}")
    say(f"try-hawk: downloading {VERE_TAG} {asset}")
    VERE.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory() as tmp:
        tarball = Path(tmp) / asset
        with urllib.request.urlopen(VERE_URL + asset, timeout=300) as src:
            tarball.write_bytes(src.read())
        with tarfile.open(tarball) as tar:
            member = next((m for m in tar.getmembers() if m.isfile()), None)
            if not member:
                raise RuntimeError(f"{asset} holds no runtime binary")
            extracted = tar.extractfile(member)
            if extracted is None:
                raise RuntimeError(f"{asset} would not open")
            VERE.write_bytes(extracted.read())
    VERE.chmod(0o755)
    if sys.platform == "darwin":
        # a downloaded binary carries com.apple.quarantine and Gatekeeper kills
        # it without a word.
        subprocess.run(["xattr", "-d", "com.apple.quarantine", str(VERE)],
                       check=False, capture_output=True)
    return VERE


def boot_argv() -> list[str]:
    # a pier that does not exist is created as a comet: -c with no identity.
    if PIER.exists():
        return [str(VERE), str(PIER)]
    return [str(VERE), "-c", str(PIER), "-u", PILL]


def fetch_code(boot: "Boot") -> str:
    # the ship's login code, read the only way there is: from the dojo, on a
    # terminal we hold. The pier's lock means this cannot happen beside the
    # running ship, so it happens BEFORE it — this owns the pier for as long as
    # the first boot takes, then hands it back.
    first = not PIER.exists()
    boot.set(phase="mining" if first else "booting",
             note="mining an identity" if first else "starting your urbit")
    master, slave = pty.openpty()
    child = subprocess.Popen(boot_argv(), stdin=slave, stdout=slave, stderr=slave,
                             start_new_session=True, cwd=str(STATE_DIR))
    os.close(slave)
    try:
        return read_code(master, child, boot)
    finally:
        os.close(master)
        stop(child)


def read_code(master: int, child: subprocess.Popen, boot: "Boot") -> str:
    # wait for a dojo prompt, let the boot go quiet, type +code, read the line
    # it prints back.
    seen, asked, quiet, asked_at = b"", False, 0.0, 0.0
    handle = LOG.open("ab")
    while True:
        if child.poll() is not None:
            raise RuntimeError("the runtime exited:\n" + "\n".join(log_tail(20, raw=True)))
        chunk = read_some(master)
        now = time.monotonic()
        if chunk:
            handle.write(unbackspace(ANSI.sub(b"", chunk)))
            handle.flush()
            seen += chunk
            quiet = now
            if boot.phase == "mining" and mined_comet():
                boot.set(phase="booting", ship=mined_comet(), note="building your urbit")
        if not asked:
            # the prompt alone is not enough: dill repaints it after every boot
            # line, and keystrokes typed into that stream are lost.
            if quiet and now - quiet > 2 and PROMPT_RE.search(ANSI.sub(b"", seen)):
                asked, seen, asked_at = True, b"", now
                boot.set(note="reading your login code")
                os.write(master, b"+code\r")
            continue
        found = CODE_RE.search(ANSI.sub(b"", seen))
        if found:
            return found.group(1).decode()
        if now - asked_at > 45:
            asked_at = now
            os.write(master, b"+code\r")


def unbackspace(data: bytes) -> bytes:
    # apply the backspaces a terminal would. Dill paints its spinner by writing
    # a character and erasing it; written to a file instead, that is a line of
    # /-\| and hint text with no meaning left in it.
    out = bytearray()
    for byte in data:
        if byte == 8:
            if out:
                out.pop()
        else:
            out.append(byte)
    return bytes(out)


def read_some(master: int) -> bytes:
    # one read, through select, so a quiet ship does not spin us.
    ready, _, _ = select.select([master], [], [], 0.5)
    if not ready:
        return b""
    try:
        return os.read(master, 65536)
    except OSError:
        return b""


def stop(child: subprocess.Popen, timeout: int = 90) -> None:
    # bring the runtime down and wait for it, so the pier's lock is free for the
    # next start. SIGTERM makes vere snapshot and exit.
    if child.poll() is not None:
        return
    child.terminate()
    for _ in range(timeout):
        if child.poll() is not None:
            return
        time.sleep(1)
    child.send_signal(signal.SIGKILL)
    child.wait(timeout=30)


def start_ship() -> int:
    # run the ship detached, on the fixed port, with its output in the log.
    # -t, not -d: -d daemonizes, and the process that stays is then not the one
    # we started, so we could not tell a crash from a fork. start_new_session
    # detaches it anyway, so the ship outlives us.
    argv = [str(VERE), "-t", "--http-port", str(SHIP_PORT), str(PIER)]
    handle = LOG.open("ab")
    child = subprocess.Popen(argv, stdout=handle, stderr=subprocess.STDOUT,
                             stdin=subprocess.DEVNULL, start_new_session=True,
                             cwd=str(STATE_DIR))
    write_ship(pid=child.pid, port=SHIP_PORT)
    return child.pid


def sign_in(code: str) -> str | None:
    # trade the code for a session and keep eyre's own set-cookie line. The page
    # hands that line back to the browser from OUR origin; same host, and a
    # cookie ignores the port, so the ship reads it as its own.
    body = urllib.parse.urlencode({"password": code}).encode()
    request = urllib.request.Request(f"http://{HOST}:{SHIP_PORT}/~/login", data=body)
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return response.headers.get("set-cookie")
    except urllib.error.HTTPError:
        # a stale code: the ship is fine, it just will not take this one.
        return None


class Boot:
    # what the page is watching, in one object behind one lock.

    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.phase = "runtime"
        self.note = "starting"
        self.ship: str | None = read_ship().get("ship")
        self.cookie: str | None = None
        self.error: str | None = None
        self.ready_at: float | None = None
        self.told = False

    def set(self, **fields) -> None:
        with self.lock:
            for k, v in fields.items():
                setattr(self, k, v)

    def state(self) -> tuple[dict, str | None]:
        # the page's view, plus the cookie to set on THIS response. Handed over
        # once: a cookie repeated on every poll is noise.
        with self.lock:
            cookie = None
            if self.phase == "ready":
                cookie, self.cookie, self.told = self.cookie, None, True
                self.ready_at = self.ready_at or time.monotonic()
            return ({"phase": self.phase, "note": self.note, "ship": self.ship,
                     "error": self.error, "log": log_tail(), "url": SHIP_URL},
                    cookie)

    def done(self) -> bool:
        # the server's job ends once the page has seen "ready" and had time to
        # follow the redirect.
        with self.lock:
            return bool(self.told and self.ready_at
                        and time.monotonic() - self.ready_at > READY_GRACE)


def run_boot(boot: Boot) -> None:
    # fetch, boot, sign in. Every phase is something the page says out loud; the
    # log tail underneath carries the detail.
    try:
        fetch_vere(lambda m: boot.set(note=m))
        if not listening(SHIP_PORT):
            code = read_ship().get("code") or fetch_code(boot)
            write_ship(code=code, ship=mined_comet() or boot.ship)
            boot.set(phase="booting", ship=mined_comet() or boot.ship,
                     note="starting your urbit")
            pid = start_ship()
            while not answering(SHIP_PORT):
                if not alive(pid):
                    raise RuntimeError("the runtime exited:\n" + "\n".join(log_tail(20, raw=True)))
        else:
            say("try-hawk: your urbit is already up")
        boot.set(phase="signing", note="signing you in")
        code = read_ship().get("code")
        cookie = sign_in(code) if code else None
        say("try-hawk: signed in" if cookie else "try-hawk: no session, eyre will ask")
        boot.set(phase="ready", note="opening your urbit", cookie=cookie)
    except Exception as e:  # noqa: BLE001
        boot.set(phase="failed", error=str(e), note="boot failed")


PAGE = """<!doctype html>
<meta charset="utf-8">
<title>hawk</title>
<style>
  :root { color-scheme: dark; --fg: #e8e6e1; --dim: #8a8682; --bg: #121210;
          --line: #2a2926; --hot: #e8a33d; }
  * { box-sizing: border-box; }
  body { margin: 0; min-height: 100vh; display: flex; align-items: center;
         justify-content: center; background: var(--bg); color: var(--fg);
         font: 14px/1.6 ui-monospace, SFMono-Regular, Menlo, monospace; }
  main { width: 100%; max-width: 40rem; padding: 2rem 1rem; }
  h1 { font-size: 1rem; font-weight: 600; margin: 0 0 1.5rem; letter-spacing: .04em; }
  .phase { display: flex; align-items: baseline; gap: .6rem;
           padding-bottom: 1rem; border-bottom: 1px solid var(--line); }
  .dot { width: .5rem; height: .5rem; border-radius: 50%; background: var(--hot);
         flex: none; animation: pulse 1.2s ease-in-out infinite; }
  .done .dot { animation: none; background: #6fbf73; }
  .failed .dot { animation: none; background: #d0574a; }
  @keyframes pulse { 0%, 100% { opacity: 1 } 50% { opacity: .25 } }
  .note { flex: 1; }
  .ship { color: var(--dim); }
  pre { margin: 1rem 0 0; padding: 0; color: var(--dim); font-size: 12px;
        white-space: pre-wrap; word-break: break-all; }
  .err { color: #e08a7e; white-space: pre-wrap; margin-top: 1rem; }
</style>
<main>
  <h1>hawk</h1>
  <div class="phase" id="phase">
    <span class="dot"></span>
    <span class="note" id="note">starting</span>
    <span class="ship" id="ship"></span>
  </div>
  <pre id="log" aria-live="polite"></pre>
  <div class="err" id="err" hidden></div>
</main>
<script>
  const phase = document.getElementById('phase')
  const note = document.getElementById('note')
  const ship = document.getElementById('ship')
  const log = document.getElementById('log')
  const err = document.getElementById('err')
  let gone = false

  async function poll() {
    if (gone) return
    let s
    try { s = await (await fetch('/api/state')).json() } catch (e) { return }
    note.textContent = s.note
    ship.textContent = s.ship ? '~' + s.ship : ''
    log.textContent = (s.log || []).join('\\n')
    phase.className = 'phase' + (s.phase === 'ready' ? ' done'
                               : s.phase === 'failed' ? ' failed' : '')
    if (s.error) { err.hidden = false; err.textContent = s.error }
    if (s.phase === 'ready' && s.url) {
      gone = true
      setTimeout(() => location.replace(s.url), 400)
    }
  }
  poll()
  setInterval(poll, 800)
</script>
"""


class Handler(http.server.BaseHTTPRequestHandler):
    server_version = "try-hawk"
    protocol_version = "HTTP/1.1"
    boot: Boot

    def log_message(self, *a) -> None:
        #
        pass

    def reply(self, ctype: str, body: bytes, cookie: str | None = None) -> None:
        #
        self.send_response(200)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        if cookie:
            self.send_header("Set-Cookie", cookie)
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        #
        if self.path == "/":
            return self.reply("text/html; charset=utf-8", PAGE.encode())
        if self.path == "/api/state":
            body, cookie = self.boot.state()
            return self.reply("application/json", json.dumps(body).encode(), cookie)
        self.send_error(404)


def already_running() -> bool:
    # is the thing on our port another try-hawk? Ask it. Anything else there is
    # a collision worth naming, not a page to send someone to.
    try:
        with urllib.request.urlopen(f"{PAGE_URL}api/state", timeout=2) as response:
            return "phase" in json.loads(response.read())
    except Exception:
        return False


def serve(boot: Boot) -> None:
    # the page, on a fixed port. Bound before the boot starts, so the browser
    # has something to open the moment it is asked to.
    Handler.boot = boot
    try:
        httpd = http.server.ThreadingHTTPServer((HOST, SERVER_PORT), Handler)
    except OSError as e:
        if already_running():
            print(f"try-hawk is already running — {PAGE_URL}", flush=True)
            webbrowser.open(PAGE_URL)
            return
        raise SystemExit(f"try-hawk: cannot bind {HOST}:{SERVER_PORT} ({e}) — "
                         f"something else is on it")
    httpd.timeout = 1.0  # so a closed browser cannot hold the loop open
    print(f"try-hawk  {PAGE_URL}\n  state {STATE_DIR}\n  Ctrl-C to stop", flush=True)
    threading.Thread(target=run_boot, args=(boot,), daemon=True).start()
    threading.Timer(0.4, lambda: webbrowser.open(PAGE_URL)).start()
    try:
        while not boot.done():
            httpd.handle_request()
    except KeyboardInterrupt:
        print()
    finally:
        httpd.server_close()
    if boot.error:
        raise SystemExit(f"try-hawk: {boot.error}")
    print(f"~{boot.ship or 'your urbit'} is up at {SHIP_URL}", flush=True)


if __name__ == "__main__":
    serve(Boot())