homelab/review-bot/main.go

// review-bot reviews pull requests with an agent that reads the repository: a Mistral agent with
// the Gitea connector, or Claude with the Gitea MCP server; TypeSafe can pick the model per pull
// request. Gitea's pull_request webhook queues the review at once; a periodic sweep over the
// configured repositories catches deliveries that were lost. The result is posted as the review
// bot. Nothing is stored: the state is the marker in the bot's own review.
package main

import (
	"context"
	"errors"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"
)

func main() {
	log := slog.New(slog.NewJSONHandler(os.Stdout, nil))
	cfg, err := loadConfig(os.Getenv, os.ReadFile)
	if err != nil {
		log.Error("invalid configuration", "err", err)
		os.Exit(2)
	}
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()
	if err := run(ctx, cfg, log); err != nil {
		log.Error("stopped", "err", err)
		os.Exit(1)
	}
}

func run(ctx context.Context, cfg Config, log *slog.Logger) error {
	metrics := newMetrics()
	gitea := &Gitea{base: cfg.GiteaURL, token: cfg.ReviewToken, http: &http.Client{Timeout: time.Minute}}
	status := newTracker(gitea, log, metrics)
	go status.Run(ctx)
	for _, w := range cfg.Warnings {
		log.Warn("configuration", "warning", w)
	}
	reviewer := &Reviewer{
		cfg:   cfg,
		gitea: gitea,
		// The conversations stream; the review timeout bounds them, not the client.
		backends: map[string]Backend{
			"mistral": &Mistral{base: cfg.MistralURL, key: cfg.MistralKey, agentPrefix: cfg.AgentName,
				connectorID: cfg.ConnectorID, http: &http.Client{}, recoverTries: 3, recoverWait: 30 * time.Second},
			"anthropic": newAnthropic(cfg),
		},
		log:     log,
		metrics: metrics,
		status:  status,
	}
	if cfg.Selector == "typesafe" {
		reviewer.selector = &TypeSafe{url: cfg.TypeSafeURL, key: cfg.TypeSafeKey, model: cfg.TypeSafeModel,
			http: &http.Client{Timeout: time.Minute}}
	}
	queue := newQueue(reviewer, cfg.QueueSize, log, metrics)
	queue.Start(ctx, cfg.Workers)

	for _, b := range backends {
		if models := modelsOf(cfg.Models, b); len(models) > 0 {
			go prepare(ctx, b, models, reviewer, log)
		}
	}
	go sweepLoop(ctx, cfg, reviewer, queue)

	srv := &http.Server{
		Addr:              cfg.Listen,
		Handler:           (&Server{cfg: cfg, queue: queue, reviewer: reviewer, metrics: metrics, log: log}).routes(),
		ReadHeaderTimeout: 10 * time.Second,
		ReadTimeout:       30 * time.Second,
		WriteTimeout:      30 * time.Second,
	}
	failed := make(chan error, 1)
	go func() { failed <- srv.ListenAndServe() }()
	log.Info("listening", "addr", cfg.Listen, "repos", cfg.Repos)

	select {
	case <-ctx.Done():
	case err := <-failed:
		if !errors.Is(err, http.ErrServerClosed) {
			return err
		}
	}
	log.Info("shutting down")
	shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_ = srv.Shutdown(shutdown)
	queue.Stop(cfg.ShutdownWait)
	return nil
}

// prepare readies one backend for its models, and retries until it works.
func prepare(ctx context.Context, backend string, models []string, r *Reviewer, log *slog.Logger) {
	for wait := 10 * time.Second; ; wait = min(wait*2, 5*time.Minute) {
		attempt, cancel := context.WithTimeout(ctx, time.Minute)
		err := r.backends[backend].Prepare(attempt, models)
		cancel()
		if err == nil {
			r.setReady(backend, true)
			log.Info("backend ready", "backend", backend, "models", models)
			return
		}
		log.Error("cannot prepare the backend", "backend", backend, "err", err, "retry_in", wait.String())
		select {
		case <-ctx.Done():
			return
		case <-time.After(wait):
		}
	}
}

func sweepLoop(ctx context.Context, cfg Config, r *Reviewer, q *Queue) {
	select {
	case <-ctx.Done():
		return
	case <-time.After(time.Minute):
	}
	for {
		r.Sweep(ctx, q)
		select {
		case <-ctx.Done():
			return
		case <-time.After(cfg.Sweep):
		}
	}
}