teabot/internal/dispatch/poll.go

package dispatch

import (
	"context"
	"time"

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

// pollRepo runs one poll cycle for a single repo: it seeds a baseline on first
// contact, then classifies and dispatches new issues, pull requests, and
// comments. Events authored by a bot personality are skipped (loop prevention),
// and anything already recorded in the state store is skipped (dedup).
func (d *Dispatcher) pollRepo(ctx context.Context, repo string) error {
	since := d.store.LastPoll(repo)
	now := time.Now()

	issues, err := d.client.ListIssues(ctx, repo, since)
	if err != nil {
		return err
	}
	pulls, err := d.client.ListPulls(ctx, repo)
	if err != nil {
		return err
	}
	comments, err := d.client.ListComments(ctx, repo, since)
	if err != nil {
		return err
	}

	// First contact: record everything currently open/recent as processed
	// without dispatching, so a fresh install doesn't stampede old items.
	if !d.store.Seeded(repo) {
		// Processed-only: baseline items are recorded so they don't stampede,
		// but are not "acted-on" (teabot never ran a session for them), so
		// later comments on them don't count as follow-ups on engaged threads.
		for _, i := range issues {
			d.store.MarkIssueProcessed(repo, i.Index)
		}
		for _, p := range pulls {
			d.store.MarkPullProcessed(repo, p.Index)
		}
		for _, c := range comments {
			d.store.MarkComment(repo, c.ID)
		}
		d.store.MarkSeeded(repo)
		d.store.SetLastPoll(repo, now)
		d.log.Info("seeded repo baseline",
			"repo", repo, "issues", len(issues), "pulls", len(pulls), "comments", len(comments))
		return nil
	}

	for _, issue := range issues {
		d.handleIssue(ctx, repo, issue)
	}
	for _, pull := range pulls {
		d.handlePull(ctx, repo, pull)
	}
	for _, comment := range comments {
		d.handleComment(ctx, repo, comment)
	}

	d.store.SetLastPoll(repo, now)
	return nil
}

// handleIssue dispatches an implementer session for a genuinely new issue.
func (d *Dispatcher) handleIssue(ctx context.Context, repo string, issue gitea.Issue) {
	if d.store.IssueProcessed(repo, issue.Index) {
		return
	}
	if d.botLogins[issue.Poster.Login] {
		d.store.MarkIssueProcessed(repo, issue.Index) // remember, but never act on our own
		return
	}
	if !d.cfg.IsAuthorAllowed(repo, issue.Poster.Login) {
		// Fail-closed: an untrusted author must never supply the prompt for a
		// skip-permissions container. Record (processed-only, not acted-on) so
		// it neither re-triggers nor counts as an engaged thread.
		d.store.MarkIssueProcessed(repo, issue.Index)
		d.log.Info("skipping issue from non-allowlisted author",
			"repo", repo, "index", issue.Index, "author", issue.Poster.Login)
		return
	}
	p := d.cfg.ImplementerFor()
	if p == nil {
		return
	}
	// Record before dispatch so a duplicate poll cannot double-launch.
	d.store.MarkIssue(repo, issue.Index)
	comments, _ := d.client.GetIssueComments(ctx, repo, issue.Index)
	d.dispatchIssue(ctx, repo, *p, issue, comments)
}

// handlePull dispatches a reviewer session for a genuinely new pull request.
func (d *Dispatcher) handlePull(ctx context.Context, repo string, pull gitea.PullRequest) {
	if d.store.PullProcessed(repo, pull.Index) {
		return
	}
	if d.botLogins[pull.Poster.Login] {
		d.store.MarkPullProcessed(repo, pull.Index)
		return
	}
	if !d.cfg.IsAuthorAllowed(repo, pull.Poster.Login) {
		d.store.MarkPullProcessed(repo, pull.Index)
		d.log.Info("skipping pull from non-allowlisted author",
			"repo", repo, "index", pull.Index, "author", pull.Poster.Login)
		return
	}
	p := d.cfg.ReviewerFor()
	if p == nil {
		return
	}
	d.store.MarkPull(repo, pull.Index)
	diff, _ := d.client.GetPullDiff(ctx, repo, pull.Index)
	comments, _ := d.client.GetIssueComments(ctx, repo, pull.Index)
	d.dispatchPull(ctx, repo, *p, pull, diff, comments)
}

// handleComment dispatches a follow-up session for a new comment on a thread
// teabot previously acted on.
func (d *Dispatcher) handleComment(ctx context.Context, repo string, comment gitea.Comment) {
	if d.store.CommentProcessed(repo, comment.ID) {
		return
	}
	// Always record the comment so it is not reconsidered next cycle.
	d.store.MarkComment(repo, comment.ID)
	if d.botLogins[comment.Poster.Login] {
		return // loop prevention: never react to our own comments
	}
	// Fail-closed: the NEW comment's author must be allowlisted. Being on an
	// acted-on thread is not enough — an untrusted comment on a bot thread must
	// not reopen the prompt-injection path.
	if !d.cfg.IsAuthorAllowed(repo, comment.Poster.Login) {
		d.log.Info("skipping comment from non-allowlisted author",
			"repo", repo, "comment_id", comment.ID, "author", comment.Poster.Login)
		return
	}
	index, ok := gitea.IssueIndexFromCommentURL(comment)
	if !ok {
		return
	}

	switch {
	case d.store.ActedOnPull(repo, index):
		p := d.cfg.ReviewerFor()
		if p == nil {
			return
		}
		pull, err := d.client.GetPull(ctx, repo, index)
		if err != nil {
			d.log.Warn("fetching pull for follow-up failed", "repo", repo, "index", index, "err", err)
			return
		}
		thread, _ := d.client.GetIssueComments(ctx, repo, index)
		d.dispatchPullFollowUp(ctx, repo, *p, pull, thread, comment)
	case d.store.ActedOnIssue(repo, index):
		p := d.cfg.ImplementerFor()
		if p == nil {
			return
		}
		issue, err := d.client.GetIssue(ctx, repo, index)
		if err != nil {
			d.log.Warn("fetching issue for follow-up failed", "repo", repo, "index", index, "err", err)
			return
		}
		thread, _ := d.client.GetIssueComments(ctx, repo, index)
		d.dispatchIssueFollowUp(ctx, repo, *p, issue, thread, comment)
	default:
		// Comment on a thread teabot never engaged with: ignore.
	}
}