marketplace/skills/sync/scripts/_gitea.py

#!/usr/bin/env python3
"""
_gitea.py — transport. Everything that talks to Gitea, and nothing else.

Not a command. This module knows logins, HTTP verbs, pagination, and Gitea's
query quirks. It does NOT know what an issue is: no sections, no acceptance
criteria, no type taxonomy. Payload shapes come from map.py; the domain model
lives one layer further out in skills/issue/scripts/issue.py.

Login: resolved from .claude/settings.local.json (env.GITEA_LOGIN), walking up
from CWD — the same file /tea:auth writes and the tea-guard hook reads. No
script here accepts a login argument: the operator's pin is the only identity
they will use. No pin -> exit with a pointer to /tea:auth.

Also holds the id map (tmp/issues/.remote.json), which pairs a remote key with
a local slug, and the paths of the store-side files this layer writes. All of
it is transport bookkeeping, not domain data — the domain never reads any of
it, and losing the map still costs a re-pull and not information: the slug it
records also travels in the issue body as `<!-- tea:id … -->` (map.py), so a
pull rebuilds the entry from the tracker. See `rebuild_map`.
"""
import datetime
import json
import os
import subprocess
import re
import sys
import urllib.parse

PAYLOAD_DIR = ".payload"
REMOTE_MAP = ".remote.json"

# How far past the ideal page count a `keep`-bounded listing may scan before it
# gives up (see list_issues). The ideal is what `limit` would need if every
# payload counted; the slack pays for the ones that do not. It is a bound on
# requests, deliberately small: "fetch until N are kept" without one is "fetch
# the whole tracker" on any repo whose filter matches mostly closed issues.
PAGE_SLACK = 4


def die(msg, code=1):
    sys.stderr.write("%s: %s\n" % (os.path.basename(sys.argv[0]), msg))
    sys.exit(code)


def warn(msg):
    sys.stderr.write("warning: %s\n" % msg)


def now_iso():
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


# --------------------------------------------------------------------------
# login
# --------------------------------------------------------------------------

def find_pin(start_dir=None):
    """Walk up from start_dir; return the login from the first
    .claude/settings.local.json carrying a non-empty env.GITEA_LOGIN."""
    d = os.path.abspath(start_dir or ".")
    while True:
        p = os.path.join(d, ".claude", "settings.local.json")
        if os.path.isfile(p):
            try:
                with open(p) as f:
                    v = (json.load(f).get("env") or {}).get("GITEA_LOGIN")
                if isinstance(v, str) and v.strip():
                    return v.strip()
            except Exception:
                pass
        parent = os.path.dirname(d)
        if parent == d:
            return None
        d = parent


def require_login():
    login = find_pin(os.getcwd())
    if not login:
        die("no login pinned (.claude/settings.local.json env.GITEA_LOGIN). Run /tea:auth.")
    return login


# --------------------------------------------------------------------------
# api
# --------------------------------------------------------------------------

def api(login, endpoint, method="GET", payload=None, payload_name=None,
        out_root=None, allow_fail=False):
    """Call `tea api`; return parsed JSON (None on an empty body).

    payload (a dict) is written to <out_root>/.payload/<name>.json and passed
    as -d @file — the file survives the call for retries and debugging.
    allow_fail returns None instead of exiting when the call fails."""
    cmd = ["tea", "api", "--login", login]
    if method != "GET":
        cmd += ["-X", method]
    if payload is not None:
        pdir = os.path.join(out_root or ".", PAYLOAD_DIR)
        os.makedirs(pdir, exist_ok=True)
        path = os.path.join(pdir, "%s.json" % (payload_name or "request"))
        with open(path, "w") as f:
            json.dump(payload, f, ensure_ascii=False, indent=2)
        cmd += ["-d", "@" + path]
    cmd.append(endpoint)

    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode != 0:
        if allow_fail:
            return None
        die("`tea api %s %s` failed:\n%s" % (method, endpoint, (r.stderr or r.stdout).strip()))
    body = r.stdout.strip()
    if not body:
        return None
    try:
        return json.loads(body)
    except json.JSONDecodeError:
        if allow_fail:
            return None
        die("`tea api %s` returned non-JSON:\n%s" % (endpoint, body[:500]))


def pages(login, endpoint, limit=50, max_pages=40, **kw):
    """GET a list endpoint page by page, yielding each page as it arrives.

    A generator, because a caller whose budget is spent on what it *keeps*
    cannot be served by a function that fetches everything first: the page after
    the one that completed the budget must never be requested. Stop consuming
    and no further request is made."""
    sep = "&" if "?" in endpoint else "?"
    for page in range(1, max_pages + 1):
        batch = api(login, "%s%spage=%d&limit=%d" % (endpoint, sep, page, limit), **kw)
        if not isinstance(batch, list) or not batch:
            return
        yield batch
        if len(batch) < limit:
            return  # a short page is the last one


def paginate(login, endpoint, limit=50, max_pages=40, **kw):
    """GET a list endpoint page by page; return the concatenated list."""
    out = []
    for batch in pages(login, endpoint, limit=limit, max_pages=max_pages, **kw):
        out.extend(batch)
    return out


def repo_base(repo=None):
    """API prefix. Without --repo, let tea fill {owner}/{repo} from CWD."""
    return "repos/%s" % repo if repo else "repos/{owner}/{repo}"


def repo_slug(login, repo=None):
    """owner/repo as a literal string — needed for remote keys, which must not
    contain tea's {owner}/{repo} placeholder."""
    if repo:
        return repo
    got = api(login, "repos/{owner}/{repo}", allow_fail=True)
    if isinstance(got, dict) and got.get("full_name"):
        return got["full_name"]
    die("cannot determine owner/repo from the CWD — pass --repo owner/repo")


def parse_key(key):
    """Return (number, repo-or-None) from 42 / #42 / owner/repo#42 / a URL."""
    key = key.strip()
    m = re.match(r'^https?://[^/]+/([^/]+)/([^/]+)/issues/(\d+)/?$', key)
    if m:
        return int(m.group(3)), "%s/%s" % (m.group(1), m.group(2))
    m = re.match(r'^([\w.-]+/[\w.-]+)#(\d+)$', key)
    if m:
        return int(m.group(2)), m.group(1)
    m = re.match(r'^#?(\d+)$', key)
    if m:
        return int(m.group(1)), None
    die("cannot parse issue key %r (want 42, #42, owner/repo#42, or an issue URL)" % key)


# --------------------------------------------------------------------------
# filters
# --------------------------------------------------------------------------

def resolve_milestone(login, base, value):
    """(id, title) for a milestone given by id or title. Exits if unknown.

    Gitea silently IGNORES an unresolvable `milestones=` filter and returns the
    whole backlog, so the milestone must be resolved before it is trusted."""
    got = paginate(login, "%s/milestones?state=all" % base, limit=100)
    for m in got or []:
        if str(m.get("id")) == str(value) or m.get("title") == str(value):
            return m["id"], m.get("title", "")
    have = ", ".join("%s (id %d)" % (m.get("title", ""), m["id"]) for m in got or [])
    die("no milestone %r in this repo — have: %s" % (value, have or "none"))


def matches(payload, milestone_id=None, labels=()):
    """Client-side re-check of a server-side filter — see resolve_milestone."""
    if payload.get("pull_request"):
        return False
    if milestone_id is not None and (payload.get("milestone") or {}).get("id") != milestone_id:
        return False
    names = {l.get("name", "") for l in payload.get("labels") or []}
    return all(l in names for l in labels)


def list_issues(login, base, state="open", labels=(), query=None,
                milestone=None, limit=100, keep=None):
    """Filtered issue payloads. Returns (payloads, milestone_title).

    One request per page, and the payload already carries the issue bodies — a
    whole milestone costs one call per 50 issues, not one per issue.

    `limit` counts the payloads the CALLER cares about, not the ones the server
    returned. Without `keep` those are the same thing and this behaves as it
    always did. With it, `keep(payload)` says whether a payload counts, pages
    keep coming until `limit` of them have, and the returned list carries the
    ones that did not count too — they were enumerated, and a caller that has
    something to say about them (pull.py: "N closed, not stored") still can.

    What `keep` means is the caller's business; this module only counts. Two
    boundaries hold whatever it decides:

    - **Stop at the limit.** The page after the one that completed the budget
      is not requested — `pages` is a generator and this loop returns out of it.
    - **Stop at the page budget.** A predicate that rejects everything must not
      turn a bounded read into a walk of the whole tracker, so a filtered read
      may scan at most `PAGE_SLACK` times the pages `limit` would need if every
      payload counted. Hitting that with an unfilled budget is a warning, not a
      silent short answer: the caller asked for N and is told it got fewer."""
    if limit < 1:
        die("--limit must be 1 or more, got %d" % limit)
    ms_id, ms_title = (None, None)
    if milestone is not None:
        ms_id, ms_title = resolve_milestone(login, base, milestone)

    params = {"state": state, "type": "issues"}
    if labels:
        params["labels"] = ",".join(labels)
    if query:
        params["q"] = query
    if ms_title:
        params["milestones"] = ms_title
    endpoint = "%s/issues?%s" % (base, urllib.parse.urlencode(params))

    per_page = min(limit, 50)
    ideal = max(1, -(-limit // per_page))
    budget = ideal if keep is None else ideal * PAGE_SLACK

    got, kept, seen_pages, last_full = [], 0, 0, False
    for batch in pages(login, endpoint, limit=per_page, max_pages=budget):
        seen_pages += 1
        last_full = len(batch) == per_page
        for p in batch:
            if not matches(p, ms_id, labels):
                continue
            got.append(p)
            if keep is None or keep(p):
                kept += 1
                if kept >= limit:
                    return got, ms_title
    if keep is not None and seen_pages >= budget and last_full:
        warn("scanned %d page(s) and stopped %d short of --limit %d — there may"
             " be more; narrow the filter or raise --limit" % (budget, limit - kept, limit))
    return got, ms_title


def get_issue(login, base, number):
    payload = api(login, "%s/issues/%d" % (base, number))
    if not isinstance(payload, dict) or "number" not in payload:
        die("issue #%d not found" % number)
    return payload


def get_comments(login, base, number):
    return paginate(login, "%s/issues/%d/comments" % (base, number))


def native_deps(login, base, number):
    """Gitea's own issue-dependency links; empty when unsupported."""
    got = api(login, "%s/issues/%d/dependencies" % (base, number), allow_fail=True)
    return [i["number"] for i in got] if isinstance(got, list) else []


def native_dep_pairs(login, base, number):
    """The same links as {(owner/repo, number)} — what a repeat push compares
    against so it does not POST a link the tracker already has.

    A bare number is ambiguous the moment a dependency lives in another repo,
    and IssueMeta lets it, so the repo travels with it. The pair is a transport
    fact; formatting it as `owner/repo#42` is map.py's job, not this module's."""
    got = api(login, "%s/issues/%d/dependencies" % (base, number), allow_fail=True)
    out = set()
    for i in got if isinstance(got, list) else []:
        repo = (i.get("repository") or {}).get("full_name") or ""
        if "number" in i:
            out.add((repo, int(i["number"])))
    return out


def add_dependency(login, base, number, dep_repo, dep_number, out_root=None):
    """Make issue `number` depend on `dep_repo#dep_number`. True on success.

    Confirmed against the instance's own swagger.v1.json (Gitea 1.26.1):

        POST /repos/{owner}/{repo}/issues/{index}/dependencies
        body: IssueMeta — {"index": <int>, "owner": "<owner>", "repo": "<name>"}
        "Make the issue in the url depend on the issue in the form."

    The URL names the blocked issue and the body the blocker, which is the same
    direction native_deps reads back ("all issues that block this issue"). A
    link that already exists answers 409, so a failure here is reported and not
    fatal: one missing cross-link must not abort a push that has already
    created issues. Callers pre-filter with native_dep_pairs."""
    owner, _, name = (dep_repo or "").partition("/")
    if not owner or not name:
        return False
    payload = {"index": int(dep_number), "owner": owner, "repo": name}
    got = api(login, "%s/issues/%d/dependencies" % (base, number), "POST", payload,
              payload_name="dep-%d-%d" % (number, dep_number),
              out_root=out_root, allow_fail=True)
    return got is not None


# --------------------------------------------------------------------------
# labels
# --------------------------------------------------------------------------

def ensure_labels(login, base, specs, root):
    """Map label name -> id, creating what the repo is missing.

    `specs` is {name: {"color", "description", "exclusive"}} handed in by the
    caller — this module does not know which namespaces are exclusive or what
    they mean. Cached in <root>/.labels.json; the cache is refreshed from the
    API before anything is created."""
    cache_path = os.path.join(root, ".labels.json")
    cache = {}
    if os.path.isfile(cache_path):
        try:
            with open(cache_path) as f:
                cache = json.load(f)
        except Exception:
            cache = {}

    if any(n not in cache for n in specs):
        cache = {l["name"]: l["id"] for l in paginate(login, "%s/labels" % base, limit=100)}

    for name, spec in specs.items():
        if name in cache:
            continue
        payload = dict(spec, name=name)
        created = api(login, "%s/labels" % base, "POST", payload,
                      payload_name="label-%s" % name.replace("/", "-"), out_root=root)
        if not created or "id" not in created:
            die("could not create label %r" % name)
        cache[name] = created["id"]
        sys.stderr.write("created label %s%s\n"
                         % (name, " (exclusive)" if spec.get("exclusive") else ""))

    os.makedirs(root, exist_ok=True)
    with open(cache_path, "w") as f:
        json.dump(cache, f, indent=2, sort_keys=True)
    return {n: cache[n] for n in specs}


def resolve_milestone_id(login, base, title):
    """Milestone id for a title, or None when the repo has no such milestone."""
    if not title or title == "none":
        return None
    for m in paginate(login, "%s/milestones?state=all" % base, limit=100) or []:
        if m.get("title") == title:
            return m["id"]
    return None


# --------------------------------------------------------------------------
# store-side files this layer owns
# --------------------------------------------------------------------------
# The issue file itself is the domain's (`issue.path_of`). The one file the sync
# layer puts beside it is named here, in one place, because three commands have
# to agree on it: pull.py writes the thread, comment.py refetches it, push.py
# deletes it along with the issue it just sent.

def comments_path(root, id):
    """An issue's comment thread — beside it, under the same slug.

    A path, not a concept the domain needs: a thread is pulled from Gitea and
    never pushed back, so the domain has no reason to know the file exists."""
    return os.path.join(root, "%s.comments.md" % id)


# --------------------------------------------------------------------------
# id map: remote key <-> local slug
# --------------------------------------------------------------------------

def map_path(root):
    return os.path.join(root, REMOTE_MAP)


def load_map(root):
    """{"owner/repo#42": "wire-sqlc-appclick"} — the local slug ledger.

    Entries outlive the files they name, and that is now the normal case rather
    than a leak: `push.py` deletes an issue's file the moment Gitea confirms it,
    and the entry it leaves behind is what lets the next `pull.py 42` land on
    the same slug. Nothing prunes them, because "no file" no longer means "no
    such issue". A stale entry costs one json line and is corrected the next
    time that number is pulled."""
    p = map_path(root)
    if not os.path.isfile(p):
        return {}
    try:
        with open(p) as f:
            got = json.load(f)
        return got if isinstance(got, dict) else {}
    except Exception:
        return {}


def save_map(root, m):
    os.makedirs(root, exist_ok=True)
    with open(map_path(root), "w") as f:
        json.dump(m, f, indent=2, sort_keys=True)


def rebuild_map(root, issues):
    """Fold the `gitea:` fields still on disk into the id map. Returns it.

    This used to say "the files are the source of truth; .remote.json is only an
    index over them", and that stopped being true the day push started deleting
    the file it had just sent. A pushed issue leaves no `gitea:` field behind to
    read, so the files are now a SUBSET of what the map knows, and a rebuild
    from them alone would throw away every entry it cannot see.

    So the contradiction is resolved by moving the source of truth, not by
    keeping this function honest about files:

        Gitea            the issue, and — in `<!-- tea:id … -->` — its slug
        .remote.json     a local number -> slug ledger, a cache of that marker
        tmp/issues/*.md  whatever happens to be checked out right now

    Which makes this a MERGE and never a replacement: it starts from what is
    already recorded and adds what the remaining files say. What it cannot
    recover — a pushed-and-dropped issue whose ledger entry was also lost — is
    not lost either; the next `pull.py <n>` reads the slug off the marker and
    writes the entry back."""
    m = load_map(root)
    for id, iss in issues.items():
        key = iss.extra.get("gitea")
        if key:
            m[key] = id
    save_map(root, m)
    return m