granthi-sync/client/granthi_sync_client.py

#!/usr/bin/env python3
"""granthi-sync client daemon.

Commands:
  link  [--server URL] [--token ZITADEL_TOKEN] [--device NAME]
        Zitadel OAuth device flow against https://id.shre.ai (or --token for
        the headless path), then POST /v1/link on the granthi-link service.
        Stores {server, gitea_base, login, token, token_name} in
        ~/.granthi-sync/config.json (0600).
  add   <folder> [--name N] [--private/--public]
        git init (branch main) if needed, create the cloud repo via
        /v1/repos, add remote 'granthi', initial commit + push. The token
        is supplied via a git credential helper (this script's hidden
        `git-credential` subcommand), never embedded in the remote URL.
  watch [--interval 30] [--once]
        Poll linked folders: autocommit local changes, fetch, ff-pull when
        the remote is strictly ahead, push when local is strictly ahead.
        DIVERGED branches are logged + recorded and SKIPPED -- never force,
        never merge (same policy as the estate's gitea_sync.py mesh).
        SIGTERM-clean.
  status
        Table of linked folders, last sync, divergence flags.

Stdlib + git CLI only.
"""

import argparse
import base64
import json
import os
import re
import signal
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone

VERSION = "1.0.0"

CONFIG_DIR = os.path.expanduser(os.environ.get("GRANTHI_SYNC_HOME", "~/.granthi-sync"))
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.json")

ZITADEL_BASE = "https://id.shre.ai"
DEVICE_CLIENT_ID = "386909715541590022"
# ^ Zitadel native app "granthi-sync-device" (appId 386909715541524486) in
#   project granthi-forge (386906525790109702); device-code + refresh grants.
DEFAULT_SERVER = "http://100.111.127.127:3042"
DEVICE_SCOPE = "openid profile email"


def log(msg):
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print(f"[{ts}] {msg}", flush=True)


# --------------------------------------------------------------------------
# HTTP helpers (patchable in tests)
# --------------------------------------------------------------------------

def http_json(method, url, headers=None, body=None, form=None, timeout=30):
    """Returns (status, parsed-json-or-{'raw': text})."""
    data = None
    hdrs = dict(headers or {})
    # Cloudflare in front of id.shre.ai 403s the default Python-urllib UA.
    hdrs.setdefault("User-Agent", f"granthi-sync/{VERSION}")
    if form is not None:
        data = urllib.parse.urlencode(form).encode()
        hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded")
    elif body is not None:
        data = json.dumps(body).encode()
        hdrs.setdefault("Content-Type", "application/json")
    req = urllib.request.Request(url, data=data, headers=hdrs, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw, status = resp.read(), resp.status
    except urllib.error.HTTPError as e:
        raw, status = e.read(), e.code
    except (urllib.error.URLError, OSError) as e:
        return 599, {"error": str(e)}
    try:
        return status, json.loads(raw) if raw else {}
    except ValueError:
        return status, {"raw": raw.decode(errors="replace")}


# --------------------------------------------------------------------------
# Config
# --------------------------------------------------------------------------

def load_config():
    try:
        with open(CONFIG_PATH) as f:
            return json.load(f)
    except FileNotFoundError:
        return {}


def save_config(cfg):
    os.makedirs(CONFIG_DIR, mode=0o700, exist_ok=True)
    tmp = CONFIG_PATH + ".tmp"
    with open(tmp, "w") as f:
        json.dump(cfg, f, indent=2)
    os.chmod(tmp, 0o600)
    os.replace(tmp, CONFIG_PATH)


# --------------------------------------------------------------------------
# git helpers
# --------------------------------------------------------------------------

def git(folder, *args, check=True):
    """Run git in folder. Returns (rc, stdout). Never uses --force."""
    proc = subprocess.run(["git", "-C", folder] + list(args),
                          capture_output=True, text=True)
    if check and proc.returncode != 0:
        raise RuntimeError(
            f"git {' '.join(args)} failed in {folder}: {proc.stderr.strip()}")
    return proc.returncode, proc.stdout.strip()


def ensure_repo(folder):
    if os.path.isdir(os.path.join(folder, ".git")):
        return
    rc, _ = git(folder, "init", "-b", "main", check=False)
    if rc != 0:  # git < 2.28 fallback
        git(folder, "init")
        git(folder, "symbolic-ref", "HEAD", "refs/heads/main")


def autocommit(folder):
    """Commit all local changes as 'sync: <ISO ts>'. Returns True if a
    commit was made."""
    _, status = git(folder, "status", "--porcelain")
    if not status:
        return False
    git(folder, "add", "-A")
    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
    git(folder, "commit", "-m", f"sync: {ts}")
    return True


def sync_state(folder, remote="granthi", branch="main"):
    """Classify local vs remote after a fetch.

    Returns one of: 'no-remote-branch', 'in-sync', 'local-ahead',
    'remote-ahead-ff', 'diverged'.
    """
    rc, _ = git(folder, "rev-parse", "--verify", "--quiet",
                f"refs/remotes/{remote}/{branch}", check=False)
    if rc != 0:
        return "no-remote-branch"
    _, local = git(folder, "rev-parse", branch)
    _, remote_sha = git(folder, "rev-parse", f"refs/remotes/{remote}/{branch}")
    if local == remote_sha:
        return "in-sync"
    rc, base = git(folder, "merge-base", branch,
                   f"refs/remotes/{remote}/{branch}", check=False)
    if rc != 0:
        return "diverged"  # unrelated histories: treat as divergence
    base = base.strip()
    if base == remote_sha:
        return "local-ahead"
    if base == local:
        return "remote-ahead-ff"
    return "diverged"


def sync_folder(folder, remote="granthi", branch="main"):
    """One sync pass for one folder. Returns (outcome, detail).

    outcome in {'synced', 'pushed', 'clean', 'diverged', 'error'}.
    Policy: ff-only pulls, plain pushes. NEVER --force, NEVER merge.
    """
    try:
        committed = autocommit(folder)
        git(folder, "fetch", remote)
        state = sync_state(folder, remote, branch)
        ff_pulled = False
        if state == "remote-ahead-ff":
            git(folder, "merge", "--ff-only", f"refs/remotes/{remote}/{branch}")
            ff_pulled = True
            state = sync_state(folder, remote, branch)
        if state == "diverged":
            return "diverged", "local and remote both advanced; skipping (no force, no merge)"
        if state in ("local-ahead", "no-remote-branch"):
            git(folder, "push", "-u", remote, branch)
            return "pushed", "committed+pushed" if committed else "pushed"
        # state == "in-sync"
        if ff_pulled:
            return "synced", "ff-pulled"
        return "clean", "in-sync"
    except RuntimeError as e:
        return "error", str(e)


# --------------------------------------------------------------------------
# credential helper (git calls back into this script; token never in URL)
# --------------------------------------------------------------------------

def cmd_git_credential(argv):
    op = argv[0] if argv else "get"
    if op != "get":
        return 0  # ignore store/erase
    attrs = {}
    for line in sys.stdin:
        line = line.strip()
        if not line:
            break
        k, _, v = line.partition("=")
        attrs[k] = v
    cfg = load_config()
    gitea = cfg.get("gitea_base", "")
    want = urllib.parse.urlparse(gitea)
    if attrs.get("host") == want.netloc:
        print(f"username={cfg.get('login', '')}")
        print(f"password={cfg.get('token', '')}")
    return 0


def install_credential_helper(folder):
    helper = f"!{sys.executable} {os.path.abspath(__file__)} git-credential"
    git(folder, "config", "credential.helper", helper)


# --------------------------------------------------------------------------
# commands
# --------------------------------------------------------------------------

def device_flow():
    """Zitadel OAuth 2.0 device authorization grant. Returns access token."""
    status, resp = http_json(
        "POST", f"{ZITADEL_BASE}/oauth/v2/device_authorization",
        form={"client_id": DEVICE_CLIENT_ID, "scope": DEVICE_SCOPE})
    if status != 200:
        raise SystemExit(f"device authorization failed (HTTP {status}): {resp}")
    print(f"\nTo link this device, open:\n\n  {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n")
    print(f"and enter code: {resp['user_code']}\n")
    interval = int(resp.get("interval", 5))
    deadline = time.time() + int(resp.get("expires_in", 300))
    while time.time() < deadline:
        time.sleep(interval)
        status, tok = http_json(
            "POST", f"{ZITADEL_BASE}/oauth/v2/token",
            form={"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                  "device_code": resp["device_code"],
                  "client_id": DEVICE_CLIENT_ID})
        if status == 200:
            return tok["access_token"]
        err = tok.get("error", "")
        if err == "authorization_pending":
            continue
        if err == "slow_down":
            interval += 5
            continue
        raise SystemExit(f"device flow failed: {tok}")
    raise SystemExit("device flow timed out (code expired)")


def cmd_link(args):
    zitadel_token = args.token or device_flow()
    status, resp = http_json(
        "POST", f"{args.server.rstrip('/')}/v1/link",
        body={"zitadel_access_token": zitadel_token,
              "device_name": args.device})
    if status != 200:
        raise SystemExit(f"link failed (HTTP {status}): {resp}")
    cfg = load_config()
    cfg.update({"server": args.server.rstrip("/"),
                "gitea_base": resp["gitea_base"],
                "login": resp["login"],
                "token": resp["token"],
                "token_name": resp["token_name"]})
    cfg.setdefault("folders", {})
    save_config(cfg)
    log(f"linked as {resp['login']} on {resp['gitea_base']} "
        f"(token {resp['token_name']}); config: {CONFIG_PATH}")
    return 0


def cmd_add(args):
    cfg = load_config()
    if "token" not in cfg:
        raise SystemExit("not linked yet -- run: granthi-sync link")
    folder = os.path.abspath(args.folder)
    if not os.path.isdir(folder):
        raise SystemExit(f"no such folder: {folder}")
    name = args.name or re.sub(r"[^a-zA-Z0-9._-]+", "-", os.path.basename(folder))
    ensure_repo(folder)
    install_credential_helper(folder)
    status, resp = http_json(
        "POST", f"{cfg['server']}/v1/repos",
        body={"token": cfg["token"], "name": name, "private": args.private})
    if status == 200:
        clone_url = resp["clone_url"]
    elif status == 409:
        clone_url = f"{cfg['gitea_base']}/{cfg['login']}/{name}.git"
        log(f"cloud repo {name} already exists; reusing")
    else:
        raise SystemExit(f"repo create failed (HTTP {status}): {resp}")
    rc, _ = git(folder, "remote", "get-url", "granthi", check=False)
    if rc == 0:
        git(folder, "remote", "set-url", "granthi", clone_url)
    else:
        git(folder, "remote", "add", "granthi", clone_url)
    autocommit(folder)
    git(folder, "push", "-u", "granthi", "main")
    cfg.setdefault("folders", {})[folder] = {
        "name": name, "branch": "main",
        "last_sync": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "diverged": False}
    save_config(cfg)
    log(f"linked folder {folder} -> {clone_url}")
    return 0


_STOP = False


def _sigterm(signum, frame):
    global _STOP
    _STOP = True
    log(f"signal {signum} received; finishing current pass then exiting")


def watch_pass():
    cfg = load_config()
    for folder, meta in sorted(cfg.get("folders", {}).items()):
        if not os.path.isdir(folder):
            log(f"SKIP {folder}: missing")
            continue
        outcome, detail = sync_folder(folder, branch=meta.get("branch", "main"))
        meta["diverged"] = outcome == "diverged"
        if outcome in ("pushed", "synced"):
            meta["last_sync"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
        if outcome == "diverged":
            log(f"DIVERGED {folder}: {detail}")
        elif outcome == "error":
            log(f"ERROR {folder}: {detail}")
        else:
            log(f"{outcome} {folder}: {detail}")
    save_config(cfg)


def cmd_watch(args):
    signal.signal(signal.SIGTERM, _sigterm)
    signal.signal(signal.SIGINT, _sigterm)
    log(f"granthi-sync watch starting (interval {args.interval}s)")
    while True:
        watch_pass()
        if args.once or _STOP:
            break
        deadline = time.time() + args.interval
        while time.time() < deadline and not _STOP:
            time.sleep(1)
        if _STOP:
            break
    log("watch stopped cleanly")
    return 0


def cmd_status(args):
    cfg = load_config()
    if "login" in cfg:
        print(f"linked: {cfg['login']} @ {cfg.get('gitea_base')} "
              f"(token {cfg.get('token_name')})")
    else:
        print("not linked")
    folders = cfg.get("folders", {})
    if not folders:
        print("no folders linked")
        return 0
    rows = [("FOLDER", "REPO", "LAST SYNC", "DIVERGED")]
    for folder, meta in sorted(folders.items()):
        rows.append((folder, meta.get("name", "?"),
                     meta.get("last_sync", "never"),
                     "YES" if meta.get("diverged") else "no"))
    widths = [max(len(r[i]) for r in rows) for i in range(4)]
    for r in rows:
        print("  ".join(c.ljust(w) for c, w in zip(r, widths)))
    return 0


def main(argv=None):
    p = argparse.ArgumentParser(prog="granthi-sync",
                                description="Granthi folder-to-cloud sync client")
    p.add_argument("--version", action="version", version=VERSION)
    sub = p.add_subparsers(dest="cmd", required=True)

    sp = sub.add_parser("link", help="link this device to your granthi account")
    sp.add_argument("--server", default=DEFAULT_SERVER)
    sp.add_argument("--token", help="ready Zitadel access token (headless path)")
    sp.add_argument("--device", default=os.uname().nodename.split(".")[0])
    sp.set_defaults(fn=cmd_link)

    sp = sub.add_parser("add", help="link a folder and push it to the cloud")
    sp.add_argument("folder")
    sp.add_argument("--name")
    grp = sp.add_mutually_exclusive_group()
    grp.add_argument("--private", dest="private", action="store_true", default=True)
    grp.add_argument("--public", dest="private", action="store_false")
    sp.set_defaults(fn=cmd_add)

    sp = sub.add_parser("watch", help="sync loop over linked folders")
    sp.add_argument("--interval", type=int, default=30)
    sp.add_argument("--once", action="store_true", help="single pass then exit")
    sp.set_defaults(fn=cmd_watch)

    sp = sub.add_parser("status", help="show linked folders")
    sp.set_defaults(fn=cmd_status)

    argv = list(sys.argv[1:] if argv is None else argv)
    if argv and argv[0] == "git-credential":  # hidden helper protocol
        return cmd_git_credential(argv[1:])
    args = p.parse_args(argv)
    return args.fn(args)


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