core/packages/git-board/script.star

# Git Board — Starlark Backend (v0.2.0)
#
# Library consumer — delegates all Gitea API work to gitea-client.
#
# Entry points:
#   on_request(req)              → surface API routes
#   on_tool_call(tool_name, params) → LLM tool execution
#
# Modules:
#   connections — resolve gitea connection (connections.read)
#   lib         — load gitea-client library
#   json        — encode/decode (universal)
#   settings    — user: default_owner, default_repo

gitea = lib.require("gitea-client")


def _conn():
    """Resolve the first available gitea connection."""
    return connections.get("gitea")


# ═══════════════════════════════════════════════
#  Surface API routes (/s/git-board/api/*)
# ═══════════════════════════════════════════════

def on_request(req):
    path   = req["path"]
    method = req["method"]

    conn = _conn()
    if conn == None:
        return _resp(400, {"error": "no gitea connection configured — add one in Settings > Connections"})

    if method == "GET" and path == "/repos":
        return _handle_repos(conn)
    elif method == "GET" and path == "/board":
        return _handle_board(conn, req)
    elif method == "GET" and path.startswith("/issue/"):
        return _handle_get_issue(conn, path[len("/issue/"):])
    elif method == "POST" and path.startswith("/issue/"):
        return _handle_post_issue(conn, path[len("/issue/"):], req)
    elif method == "GET" and path.startswith("/ci/"):
        return _handle_ci(conn, path[len("/ci/"):])

    return _resp(404, {"error": "not found"})


def _handle_repos(conn):
    """List repositories accessible via the connection."""
    repos = gitea.get_repos(conn)
    if repos == None:
        return _resp(502, {"error": "gitea request failed"})
    return _resp(200, {"data": repos})


def _handle_board(conn, req):
    """Combined issues + PRs for kanban board."""
    q = req.get("query", {})
    owner = _str(q.get("owner", "")) or settings.get("default_owner") or ""
    repo  = _str(q.get("repo", ""))  or settings.get("default_repo")  or ""
    if not owner or not repo:
        return _resp(400, {"error": "owner and repo required (query params or user settings)"})

    issues = gitea.get_issues(conn, owner, repo, "open") or []
    prs    = gitea.get_prs(conn, owner, repo, "open")    or []

    board = {"issues": issues, "pull_requests": prs}
    return _resp(200, board)


def _handle_get_issue(conn, issue_path):
    """Get issue detail: /issue/:owner/:repo/:number"""
    parts = issue_path.split("/", 2)
    if len(parts) < 3:
        return _resp(400, {"error": "path must be /issue/:owner/:repo/:number"})
    owner, repo, num = parts[0], parts[1], int(parts[2])
    data = gitea.get_issue(conn, owner, repo, num)
    if data == None:
        return _resp(404, {"error": "issue not found"})
    return _resp(200, data)


def _handle_post_issue(conn, issue_path, req):
    """Update issue or add comment: /issue/:owner/:repo/:number[/comment]"""
    parts = issue_path.split("/", 3)
    if len(parts) < 3:
        return _resp(400, {"error": "path must be /issue/:owner/:repo/:number"})
    owner, repo, num = parts[0], parts[1], int(parts[2])
    action = parts[3] if len(parts) > 3 else ""

    body = json.decode(req.get("body", "{}"))

    if action == "comment":
        text = body.get("body", "")
        if not text:
            return _resp(400, {"error": "comment body required"})
        result = gitea.add_comment(conn, owner, repo, num, text)
        if result == None:
            return _resp(502, {"error": "failed to add comment"})
        return _resp(201, result)

    # Default: update issue fields (state, title, body, assignee)
    title = body.get("title", "")
    issue_body = body.get("body", "")
    state = body.get("state", "")
    result = gitea.update_issue(conn, owner, repo, num, title, issue_body, state)
    if result == None:
        return _resp(502, {"error": "failed to update issue"})
    return _resp(200, result)


def _handle_ci(conn, ref_path):
    """CI status for owner/repo/ref."""
    parts = ref_path.split("/", 2)
    if len(parts) < 3:
        return _resp(400, {"error": "path must be /ci/:owner/:repo/:ref"})
    owner, repo, ref = parts[0], parts[1], parts[2]
    result = gitea.get_ci_status(conn, owner, repo, ref)
    if result == None:
        return _resp(502, {"error": "CI status request failed"})
    return _resp(200, result)


# ═══════════════════════════════════════════════
#  LLM Tools (called by AI during chat)
# ═══════════════════════════════════════════════

def on_tool_call(tool_name, params):
    conn = _conn()
    if conn == None:
        return {"error": "no gitea connection configured — add one in Settings > Connections"}

    owner = params.get("owner", "")
    repo  = params.get("repo", "")

    if tool_name == "list_issues":
        state = params.get("state", "open")
        data = gitea.get_issues(conn, owner, repo, state)
        if data == None:
            return {"error": "failed to list issues"}
        return {"issues": data, "count": len(data)}

    elif tool_name == "get_issue":
        num = int(params.get("number", 0))
        data = gitea.get_issue(conn, owner, repo, num)
        if data == None:
            return {"error": "issue not found"}
        return data

    elif tool_name == "create_issue":
        data = gitea.create_issue(conn, owner, repo,
                                  params.get("title", ""),
                                  params.get("body", ""),
                                  params.get("labels", ""))
        if data == None:
            return {"error": "failed to create issue"}
        return data

    elif tool_name == "update_issue":
        num = int(params.get("number", 0))
        data = gitea.update_issue(conn, owner, repo, num,
                                  params.get("title", ""),
                                  params.get("body", ""),
                                  params.get("state", ""))
        if data == None:
            return {"error": "failed to update issue"}
        return data

    elif tool_name == "add_comment":
        num = int(params.get("number", 0))
        data = gitea.add_comment(conn, owner, repo, num, params.get("body", ""))
        if data == None:
            return {"error": "failed to add comment"}
        return data

    elif tool_name == "list_pull_requests":
        state = params.get("state", "open")
        data = gitea.get_prs(conn, owner, repo, state)
        if data == None:
            return {"error": "failed to list PRs"}
        return {"pull_requests": data, "count": len(data)}

    elif tool_name == "get_ci_status":
        ref = params.get("ref", "")
        data = gitea.get_ci_status(conn, owner, repo, ref)
        if data == None:
            return {"error": "failed to get CI status"}
        return data

    return {"error": "unknown tool: " + tool_name}


# ═══════════════════════════════════════════════
#  Response helpers
# ═══════════════════════════════════════════════

def _resp(status, data):
    return {"status": status, "body": json.encode(data), "headers": {"Content-Type": "application/json"}}

def _str(v):
    """Coerce Starlark query param to string."""
    if v == None:
        return ""
    return str(v)