// 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" }