teabot/internal/dispatch/dispatch.go

// Package dispatch is teabot's core loop: it polls watched repos, filters
// events (dedup + loop prevention), and dispatches one-shot Claude sessions.
package dispatch

import (
	"context"
	"log/slog"
	"net/url"
	"strings"
	"sync"
	"time"

	"git.unkin.net/unkin/teabot/internal/config"
	"git.unkin.net/unkin/teabot/internal/docker"
	"git.unkin.net/unkin/teabot/internal/gitea"
)

// GiteaClient is the read surface of the Gitea API that the dispatcher needs.
// It is an interface so tests can supply a fake without network access.
type GiteaClient interface {
	ListIssues(ctx context.Context, repo string, since time.Time) ([]gitea.Issue, error)
	ListPulls(ctx context.Context, repo string) ([]gitea.PullRequest, error)
	ListComments(ctx context.Context, repo string, since time.Time) ([]gitea.Comment, error)
	GetIssueComments(ctx context.Context, repo string, index int64) ([]gitea.Comment, error)
	GetIssue(ctx context.Context, repo string, index int64) (gitea.Issue, error)
	GetPull(ctx context.Context, repo string, index int64) (gitea.PullRequest, error)
	GetPullDiff(ctx context.Context, repo string, index int64) (string, error)
}

// StateStore is the persistence surface the dispatcher needs.
type StateStore interface {
	IssueProcessed(repo string, index int64) bool
	PullProcessed(repo string, index int64) bool
	CommentProcessed(repo string, id int64) bool
	ActedOnIssue(repo string, index int64) bool
	ActedOnPull(repo string, index int64) bool
	MarkIssue(repo string, index int64)
	MarkPull(repo string, index int64)
	MarkIssueProcessed(repo string, index int64)
	MarkPullProcessed(repo string, index int64)
	MarkComment(repo string, id int64)
	Seeded(repo string) bool
	MarkSeeded(repo string)
	LastPoll(repo string) time.Time
	SetLastPoll(repo string, t time.Time)
	Save() error
}

// Dispatcher wires configuration, state, the Gitea client, and the job runner.
type Dispatcher struct {
	cfg    *config.Config
	store  StateStore
	client GiteaClient
	runner docker.Runner
	log    *slog.Logger

	sem       chan struct{}
	wg        sync.WaitGroup
	botLogins map[string]bool
	gitHost   string
}

// New builds a Dispatcher.
func New(cfg *config.Config, store StateStore, client GiteaClient, runner docker.Runner, log *slog.Logger) *Dispatcher {
	host := cfg.GiteaURL
	if u, err := url.Parse(cfg.GiteaURL); err == nil {
		host = u.Host
	}
	return &Dispatcher{
		cfg:       cfg,
		store:     store,
		client:    client,
		runner:    runner,
		log:       log,
		sem:       make(chan struct{}, cfg.MaxConcurrent),
		botLogins: cfg.BotLogins(),
		gitHost:   host,
	}
}

// Run polls on an interval until ctx is cancelled, then waits for in-flight jobs.
func (d *Dispatcher) Run(ctx context.Context) error {
	ticker := time.NewTicker(d.cfg.PollInterval)
	defer ticker.Stop()
	// Poll immediately, then on each tick.
	d.pollAll(ctx)
	for {
		select {
		case <-ctx.Done():
			d.log.Info("shutting down, waiting for in-flight jobs")
			d.wg.Wait()
			return d.store.Save()
		case <-ticker.C:
			d.pollAll(ctx)
		}
	}
}

// PollOnce runs a single poll cycle. When wait is true it blocks until every
// job dispatched during the cycle has finished (used by `run --once`).
func (d *Dispatcher) PollOnce(ctx context.Context, wait bool) error {
	d.pollAll(ctx)
	if wait {
		d.wg.Wait()
	}
	return d.store.Save()
}

// Wait blocks until all in-flight jobs finish.
func (d *Dispatcher) Wait() { d.wg.Wait() }

func (d *Dispatcher) pollAll(ctx context.Context) {
	for _, repo := range d.cfg.Repos {
		if ctx.Err() != nil {
			return
		}
		if err := d.pollRepo(ctx, repo); err != nil {
			d.log.Warn("poll failed", "repo", repo, "err", err)
		}
	}
	if err := d.store.Save(); err != nil {
		d.log.Warn("saving state failed", "err", err)
	}
}

// cloneURL returns the plain HTTPS clone URL for a repo.
func (d *Dispatcher) cloneURL(repo string) string {
	return d.cfg.GiteaURL + "/" + strings.Trim(repo, "/") + ".git"
}