gbr-games/cmd/gbrgames/worker.go

package main

import (
	"bufio"
	"bytes"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"
)

// worker runs the AI builder. It asks the panel for queued jobs (it only connects out,
// like node-agents do), runs research or build in a fresh clone of this repo, streams the output to the
// panel as the job's log, and finishes a build with a pull request opened as the bot account. CI then
// validates and boot-tests that pull request like any other.
//
// Panel API (headers X-Builder-Token and X-Builder-Worker):
//
//	POST /api/v1/builder/jobs/claim          {worker}                 -> 200 builderJob | 204 nothing queued
//	POST /api/v1/builder/jobs/{id}/progress  {stage, lines}           -> 200 {cancelled}
//	POST /api/v1/builder/jobs/{id}/result    {research | build+prUrl | error}

type builderJob struct {
	ID       string          `json:"id"`
	Kind     string          `json:"kind"` // "research" or "build"
	GameName string          `json:"gameName"`
	Notes    string          `json:"notes"`
	Variant  json.RawMessage `json:"variant"`
}

type jobResult struct {
	Research *researchResult `json:"research,omitempty"`
	Build    *buildResult    `json:"build,omitempty"`
	PRURL    string          `json:"prUrl,omitempty"`
	Error    string          `json:"error,omitempty"`
}

type workerConfig struct {
	panel, token          string
	gitea, repo, botToken string
	workDir, name         string
	poll                  time.Duration
	provider, model       string
	botLogin, botEmail    string // the token's account, used as the commit author
}

// checkBot reads which account GITEA_TOKEN belongs to and makes sure it can push to the games repo, so a
// wrong token or missing permission shows when the worker starts, not at the end of a 30-minute build.
func (c *workerConfig) checkBot() error {
	var user struct {
		Login string `json:"login"`
		Email string `json:"email"`
	}
	if err := c.giteaGet("/user", &user); err != nil {
		if !strings.HasPrefix(err.Error(), "HTTP 403") {
			return fmt.Errorf("GITEA_TOKEN: %w", err)
		}
		// A repository-only token (all the worker needs) may not read its own account: name the author from
		// BOT_NAME instead. Whether it can push is still checked below.
		user.Login = envOr("BOT_NAME", "gbr-bot")
	}
	var repo struct {
		Permissions struct {
			Push bool `json:"push"`
		} `json:"permissions"`
	}
	if err := c.giteaGet("/repos/"+c.repo, &repo); err != nil {
		return fmt.Errorf("%s: %w", c.repo, err)
	}
	if !repo.Permissions.Push {
		return fmt.Errorf("the account %s can't push to %s: give it Write access there", user.Login, c.repo)
	}
	c.botLogin, c.botEmail = user.Login, user.Email
	if c.botEmail == "" {
		c.botEmail = user.Login + "@users.noreply"
	}
	return nil
}

func (c *workerConfig) giteaGet(p string, out any) error {
	req, err := http.NewRequest(http.MethodGet, c.gitea+"/api/v1"+p, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "token "+c.botToken)
	resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode >= 300 {
		raw, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("HTTP %d: %.200s", resp.StatusCode, raw)
	}
	return json.NewDecoder(resp.Body).Decode(out)
}

func runWorker(args []string) error {
	fs := flag.NewFlagSet("worker", flag.ExitOnError)
	once := fs.Bool("once", false, "handle at most one job, then exit")
	fs.Parse(args)
	host, _ := os.Hostname()
	c := workerConfig{
		panel: strings.TrimRight(os.Getenv("PANEL_URL"), "/"), token: os.Getenv("BUILDER_TOKEN"),
		gitea: strings.TrimRight(envOr("GITEA_URL", "https://www.gitea.gbreed.com"), "/"), repo: envOr("GAMES_REPO", "gbreed-public/gbr-games"),
		botToken: os.Getenv("GITEA_TOKEN"), workDir: envOr("WORK_DIR", "/work"), name: envOr("WORKER_NAME", host),
		provider: envOr("MODEL_PROVIDER", "ollama"), model: os.Getenv("MODEL"),
	}
	var err error
	if c.poll, err = time.ParseDuration(envOr("POLL_INTERVAL", "20s")); err != nil {
		return err
	}
	for _, v := range []struct{ name, value string }{{"PANEL_URL", c.panel}, {"BUILDER_TOKEN", c.token}, {"GITEA_TOKEN", c.botToken}} {
		if v.value == "" {
			return fmt.Errorf("set %s", v.name)
		}
	}
	if err := c.checkBot(); err != nil {
		return err
	}
	fmt.Printf("[worker] %s polling %s every %s; pull requests as %s\n", c.name, c.panel, c.poll, c.botLogin)
	for {
		job, err := c.claim()
		if err != nil {
			fmt.Printf("[worker] claim: %v\n", err)
		}
		if job != nil {
			fmt.Printf("[worker] job %s: %s %q\n", job.ID, job.Kind, job.GameName)
			res := c.handle(job)
			if err := c.post("/jobs/"+job.ID+"/result", res, nil); err != nil {
				fmt.Printf("[worker] posting the result of %s: %v\n", job.ID, err)
			}
			if *once {
				return nil
			}
			continue
		}
		if *once {
			return nil
		}
		time.Sleep(c.poll)
	}
}

func envOr(name, def string) string {
	if v := os.Getenv(name); v != "" {
		return v
	}
	return def
}

func (c *workerConfig) claim() (*builderJob, error) {
	var job builderJob
	status, err := c.request("/jobs/claim", map[string]string{"worker": c.name}, &job)
	if err != nil || status == http.StatusNoContent {
		return nil, err
	}
	return &job, nil
}

func (c *workerConfig) post(p string, body, out any) error {
	_, err := c.request(p, body, out)
	return err
}

func (c *workerConfig) request(p string, body, out any) (int, error) {
	payload, err := json.Marshal(body)
	if err != nil {
		return 0, err
	}
	req, err := http.NewRequest(http.MethodPost, c.panel+"/api/v1/builder"+p, bytes.NewReader(payload))
	if err != nil {
		return 0, err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-Builder-Token", c.token)
	req.Header.Set("X-Builder-Worker", c.name)
	resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	raw, _ := io.ReadAll(resp.Body)
	if resp.StatusCode >= 300 {
		return resp.StatusCode, fmt.Errorf("HTTP %d: %.200s", resp.StatusCode, raw)
	}
	if out != nil && len(bytes.TrimSpace(raw)) > 0 {
		if err := json.Unmarshal(raw, out); err != nil {
			return resp.StatusCode, err
		}
	}
	return resp.StatusCode, nil
}

// handle runs one job in its own clone and returns what to report; it never panics the worker.
func (c *workerConfig) handle(job *builderJob) *jobResult {
	dir := filepath.Join(c.workDir, "job-"+job.ID)
	os.RemoveAll(dir)
	defer os.RemoveAll(dir)
	if out, err := runCmd("git", "clone", "-q", "--depth", "1", c.gitea+"/"+c.repo+".git", dir); err != nil {
		return &jobResult{Error: "git clone: " + out}
	}
	self, err := os.Executable()
	if err != nil {
		return &jobResult{Error: err.Error()}
	}
	resultFile := filepath.Join(dir, "job-result.json")
	modelArgs := []string{"-provider", c.provider}
	if c.model != "" {
		modelArgs = append(modelArgs, "-model", c.model)
	}

	switch job.Kind {
	case "research":
		args := append([]string{"research", "-name", job.GameName, "-notes", job.Notes, "-out", resultFile}, modelArgs...)
		if err := c.runStreamed(job.ID, dir, self, args, "RESEARCH"); err != nil {
			return &jobResult{Error: err.Error()}
		}
		res := &researchResult{}
		if err := readJSON(resultFile, res); err != nil {
			return &jobResult{Error: err.Error()}
		}
		return &jobResult{Research: res}

	case "build":
		variantFile := filepath.Join(dir, "job-variant.json")
		if err := os.WriteFile(variantFile, job.Variant, 0o644); err != nil {
			return &jobResult{Error: err.Error()}
		}
		args := append([]string{"build", "-name", job.GameName, "-notes", job.Notes, "-variant", "@" + variantFile, "-out", resultFile}, modelArgs...)
		runErr := c.runStreamed(job.ID, dir, self, args, "FIRST_BOOT")
		res := &buildResult{}
		if err := readJSON(resultFile, res); err != nil {
			msg := "the build produced no result"
			if runErr != nil {
				msg = runErr.Error()
			}
			return &jobResult{Error: msg}
		}
		// A draft that validates but didn't boot still goes up as a draft PR: a person can finish it, and
		// CI shows the same boot failure
		if _, err := os.Stat(filepath.Join(dir, res.Dir, "game.yaml")); err != nil {
			return &jobResult{Build: res, Error: "no template was written: " + strings.Join(res.Problems, "; ")}
		}
		c.progress(job.ID, "OPENING_PR", []string{"[worker] opening the pull request"})
		pr, err := c.openPR(dir, job, res)
		if err != nil {
			return &jobResult{Build: res, Error: "opening the pull request: " + err.Error()}
		}
		return &jobResult{Build: res, PRURL: pr}
	}
	return &jobResult{Error: "unknown job kind " + job.Kind}
}

// Output prefixes that mark a new stage of a build, shown in the panel's progress bar
var stagePrefixes = []struct{ prefix, stage string }{
	{"[build] first boot", "FIRST_BOOT"},
	{"[first-boot]", ""},
	{"[draft]", "DRAFTING"},
	{"[build] boot ", "BOOT_TEST"},
}

// runStreamed runs a gbrgames command and sends its output to the panel every few seconds (which also
// tells the panel the worker is alive). The panel can answer "cancelled" to stop the job.
func (c *workerConfig) runStreamed(jobID, dir, self string, args []string, stage string) error {
	cmd := exec.Command(self, args...)
	cmd.Dir = dir
	pr, pw := io.Pipe()
	cmd.Stdout, cmd.Stderr = pw, pw
	if err := cmd.Start(); err != nil {
		return err
	}
	var mu sync.Mutex
	var pending []string
	var lastErr string
	done := make(chan struct{})
	go func() {
		sc := bufio.NewScanner(pr)
		sc.Buffer(make([]byte, 1024*1024), 1024*1024)
		for sc.Scan() {
			line := strings.TrimRight(ansi.ReplaceAllString(sc.Text(), ""), "\r ")
			fmt.Println(line)
			mu.Lock()
			pending = append(pending, line)
			for _, s := range stagePrefixes {
				if s.stage != "" && strings.HasPrefix(line, s.prefix) {
					stage = s.stage
				}
			}
			if strings.HasPrefix(line, "error: ") {
				lastErr = strings.TrimPrefix(line, "error: ")
			}
			mu.Unlock()
		}
		close(done)
	}()
	wait := make(chan error, 1)
	go func() {
		err := cmd.Wait()
		pw.Close()
		<-done
		wait <- err
	}()

	tick := time.NewTicker(5 * time.Second)
	defer tick.Stop()
	flush := func() bool {
		mu.Lock()
		lines, st := pending, stage
		pending = nil
		mu.Unlock()
		return c.progress(jobID, st, lines)
	}
	for {
		select {
		case err := <-wait:
			flush()
			if err != nil {
				if lastErr != "" {
					return errors.New(lastErr)
				}
				return err
			}
			return nil
		case <-tick.C:
			if cancelled := flush(); cancelled {
				cmd.Process.Kill()
				<-wait
				return errors.New("cancelled from the panel")
			}
		}
	}
}

// progress sends log lines (possibly none, as a heartbeat) and reports whether the job was cancelled.
func (c *workerConfig) progress(jobID, stage string, lines []string) bool {
	var resp struct {
		Cancelled bool `json:"cancelled"`
	}
	if lines == nil {
		lines = []string{}
	}
	if err := c.post("/jobs/"+jobID+"/progress", map[string]any{"stage": stage, "lines": lines}, &resp); err != nil {
		fmt.Printf("[worker] progress: %v\n", err)
	}
	return resp.Cancelled
}

// openPR commits the new game folder to the game's draft branch as the bot and opens a pull request to
// main. A game has one draft at a time: an earlier draft's PR is closed and its branch deleted first.
func (c *workerConfig) openPR(dir string, job *builderJob, res *buildResult) (string, error) {
	branch := draftBranch(res.GameID)
	replaced, err := c.replaceOldDrafts(res.GameID)
	if err != nil {
		return "", fmt.Errorf("replacing the earlier draft: %w", err)
	}
	summary := res.Summary
	for _, n := range replaced {
		summary += fmt.Sprintf("\nReplaces #%d (closed, branch deleted).\n", n)
	}
	auth := "http.extraHeader=Authorization: token " + c.botToken
	steps := []struct {
		name string
		args []string
	}{
		{"checkout", []string{"checkout", "-q", "-b", branch}},
		{"add", []string{"add", filepath.ToSlash(res.Dir)}},
		{"commit", []string{"-c", "user.name=" + c.botLogin, "-c", "user.email=" + c.botEmail, "commit", "-q", "-m",
			fmt.Sprintf("feat(games): draft %s\n\nDrafted and boot-tested by the GBR builder (job %s).", job.GameName, job.ID)}},
		// Forced: the branch is the bot's own, and whatever was on it was the draft being replaced
		{"push", []string{"-c", auth, "push", "-q", "--force", "origin", branch}},
	}
	for _, s := range steps {
		cmd := exec.Command("git", s.args...)
		cmd.Dir = dir
		if out, err := cmd.CombinedOutput(); err != nil {
			// The push command holds the token: name the step only, and mask it in git's output
			return "", fmt.Errorf("git %s: %s", s.name, strings.ReplaceAll(string(out), c.botToken, "***"))
		}
	}
	title := "feat(games): add " + job.GameName
	if !res.Passed {
		title = "WIP: " + title + " (did not boot yet)"
	}
	var pr struct {
		HTMLURL string `json:"html_url"`
	}
	body := map[string]any{"title": title, "head": branch, "base": "main", "body": summary}
	if err := c.giteaCall(http.MethodPost, "/repos/"+c.repo+"/pulls", body, &pr); err != nil {
		return "", err
	}
	return pr.HTMLURL, nil
}

// replaceOldDrafts closes the game's open draft PRs and deletes their branches, so a rebuild leaves one
// draft to review instead of a pile. Only draft/ branches for this game are touched, never anyone's other
// work. Returns the numbers of the PRs it closed.
func (c *workerConfig) replaceOldDrafts(gameID string) ([]int, error) {
	var pulls []struct {
		Number int `json:"number"`
		Head   struct {
			Ref string `json:"ref"`
		} `json:"head"`
	}
	if err := c.giteaCall(http.MethodGet, "/repos/"+c.repo+"/pulls?state=open&limit=50", nil, &pulls); err != nil {
		return nil, err
	}
	branches := map[string]bool{draftBranch(gameID): true}
	var closed []int
	for _, p := range pulls {
		if !isDraftBranchOf(p.Head.Ref, gameID) {
			continue
		}
		if err := c.giteaCall(http.MethodPatch, fmt.Sprintf("/repos/%s/pulls/%d", c.repo, p.Number), map[string]string{"state": "closed"}, nil); err != nil {
			return closed, err
		}
		fmt.Printf("[worker] closed #%d, the earlier draft of %s\n", p.Number, gameID)
		closed = append(closed, p.Number)
		branches[p.Head.Ref] = true
	}
	for b := range branches {
		// Gone already is fine: not every game has had a draft
		if err := c.giteaCall(http.MethodDelete, "/repos/"+c.repo+"/branches/"+url.PathEscape(b), nil, nil); err != nil && !strings.HasPrefix(err.Error(), "HTTP 404") {
			return closed, fmt.Errorf("deleting branch %s: %w", b, err)
		}
	}
	return closed, nil
}

func draftBranch(gameID string) string {
	return "draft/" + gameID
}

// isDraftBranchOf matches draft/<id>, and draft/<id>-<job> as the worker named them before.
func isDraftBranchOf(ref, gameID string) bool {
	d := draftBranch(gameID)
	if ref == d {
		return true
	}
	rest, ok := strings.CutPrefix(ref, d+"-")
	return ok && regexp.MustCompile(`^[0-9a-f]{8}$`).MatchString(rest)
}

// giteaCall sends one API call as the bot; out may be nil.
func (c *workerConfig) giteaCall(method, p string, body, out any) error {
	var rd io.Reader
	if body != nil {
		payload, err := json.Marshal(body)
		if err != nil {
			return err
		}
		rd = bytes.NewReader(payload)
	}
	req, err := http.NewRequest(method, c.gitea+"/api/v1"+p, rd)
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "token "+c.botToken)
	resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	raw, _ := io.ReadAll(resp.Body)
	if resp.StatusCode >= 300 {
		return fmt.Errorf("HTTP %d: %.200s", resp.StatusCode, raw)
	}
	if out != nil && len(bytes.TrimSpace(raw)) > 0 {
		return json.Unmarshal(raw, out)
	}
	return nil
}

func readJSON(p string, out any) error {
	raw, err := os.ReadFile(p)
	if err != nil {
		return err
	}
	return json.Unmarshal(raw, out)
}