package worker import ( "context" "encoding/json" "fmt" "log/slog" "strings" "sync" "time" "gitea-codex-bot/internal/commands" "gitea-codex-bot/internal/config" "gitea-codex-bot/internal/domain" "gitea-codex-bot/internal/gitea" "gitea-codex-bot/internal/review" "gitea-codex-bot/internal/store" ) type Worker struct { settings config.Settings store store.Store gitea *gitea.Client runner domain.ReviewRunner logger *slog.Logger } func New(settings config.Settings, st store.Store, client *gitea.Client, runner domain.ReviewRunner, logger *slog.Logger) *Worker { return &Worker{settings: settings, store: st, gitea: client, runner: runner, logger: logger} } func (w *Worker) Run(ctx context.Context) error { var wg sync.WaitGroup for i := 0; i < w.settings.Concurrency; i++ { wg.Add(1) go func() { defer wg.Done(); w.loop(ctx) }() } wg.Wait() return nil } func (w *Worker) loop(ctx context.Context) { for { if ctx.Err() != nil { return } job, run, err := w.store.ClaimNextJob(ctx, time.Now().UTC(), 5*time.Minute, 2) if err != nil { w.logger.Error("claim job", "error", err) sleep(ctx, time.Second) continue } if job == nil { sleep(ctx, time.Second) continue } if err := w.process(ctx, *job, *run); err != nil { w.logger.Error("process job", "job_id", job.ID, "error", err) } } } func (w *Worker) process(ctx context.Context, job domain.Job, run domain.ReviewRun) error { cmd := commandFromJob(job, w.settings.Aliases()) if cmd.Name == "help" || cmd.Name == "ignore" || cmd.Name == "explain" { return w.processNonReview(ctx, job, run, cmd) } pr, err := w.gitea.GetPullRequest(ctx, job.Repo, job.PRNumber) if err != nil { return w.fail(ctx, job, run, err) } if pr.IsFork && !w.settings.AllowUntrustedForks { message := "Skipped review for fork PR because `ALLOW_UNTRUSTED_FORKS=false`." _, postErr := w.gitea.PostIssueComment(ctx, job.Repo, job.PRNumber, message) if postErr != nil { return w.fail(ctx, job, run, postErr) } return w.store.FinishJob(ctx, job.ID, run.ID, true, true, &domain.ReviewResult{Verdict: "correct", Confidence: 1, Summary: message, Findings: []domain.Finding{}}, nil) } cfg := review.MissingRepoConfig() text, configured, cfgErr := w.gitea.GetFileContent(ctx, job.Repo, ".codex-review.yml", pr.HeadSHA) if cfgErr != nil { return w.fail(ctx, job, run, cfgErr) } if configured { cfg, err = review.ParseRepoConfig(text) if err != nil { return w.fail(ctx, job, run, err) } } if !cfg.Enabled { _, postErr := w.gitea.PostIssueComment(ctx, job.Repo, job.PRNumber, review.FormatDisabledAck()) if postErr != nil { return w.fail(ctx, job, run, postErr) } return w.store.FinishJob(ctx, job.ID, run.ID, true, true, &domain.ReviewResult{Verdict: "correct", Confidence: 1, Summary: review.FormatDisabledAck(), Findings: []domain.Finding{}}, nil) } review.ResolveMode(&cmd, cfg) result, err := w.runner.Run(ctx, pr, cmd, cfg) if err != nil { return w.fail(ctx, job, run, err) } body := review.FormatResultComment(pr.HeadSHA, result, cfg.Configured) commentID, err := w.gitea.PostIssueComment(ctx, job.Repo, job.PRNumber, body) if err != nil { return w.fail(ctx, job, run, err) } if err := w.store.UpsertBotComment(ctx, job.Repo, job.PRNumber, "codex-review", commentID, pr.HeadSHA); err != nil { return w.fail(ctx, job, run, err) } return w.store.FinishJob(ctx, job.ID, run.ID, true, false, &result, nil) } func (w *Worker) processNonReview(ctx context.Context, job domain.Job, run domain.ReviewRun, cmd domain.ParsedCommand) error { switch cmd.Name { case "ignore": result := domain.ReviewResult{Verdict: "correct", Confidence: 1, Summary: "Ignore command acknowledged. No review run executed.", Findings: []domain.Finding{}} return w.store.FinishJob(ctx, job.ID, run.ID, true, true, &result, nil) case "explain": latest, err := w.store.LatestSuccessfulReview(ctx, job.Repo, job.PRNumber) if err != nil { return w.fail(ctx, job, run, err) } message := "## Codex Explain\n\nNo previous result found for this command." if latest != nil && len(latest.ResultJSON) > 0 { var result domain.ReviewResult if json.Unmarshal(latest.ResultJSON, &result) == nil { message = "## Codex Explain\n\n" + result.Summary } } if _, err := w.gitea.PostIssueComment(ctx, job.Repo, job.PRNumber, message); err != nil { return w.fail(ctx, job, run, err) } return w.store.FinishJob(ctx, job.ID, run.ID, true, true, &domain.ReviewResult{Verdict: "correct", Confidence: 1, Summary: message, Findings: []domain.Finding{}}, nil) case "help": comments, err := w.gitea.GetIssueComments(ctx, job.Repo, job.PRNumber) if err != nil { return w.fail(ctx, job, run, err) } pending, err := w.store.PendingCount(ctx, job.Repo, job.PRNumber) if err != nil { return w.fail(ctx, job, run, err) } message := helpComment(comments, w.settings.GiteaBotUsername, pending) if _, err := w.gitea.PostIssueComment(ctx, job.Repo, job.PRNumber, message); err != nil { return w.fail(ctx, job, run, err) } return w.store.FinishJob(ctx, job.ID, run.ID, true, true, &domain.ReviewResult{Verdict: "correct", Confidence: 1, Summary: "Help/status summary posted.", Findings: []domain.Finding{}}, nil) } return w.fail(ctx, job, run, fmt.Errorf("unsupported worker command %q", cmd.Name)) } func (w *Worker) fail(ctx context.Context, job domain.Job, run domain.ReviewRun, err error) error { errorText := strings.TrimSpace(err.Error()) if errorText == "" { errorText = "review failed" } if _, postErr := w.gitea.PostIssueComment(ctx, job.Repo, job.PRNumber, review.FailureComment(job.HeadSHA, errorText)); postErr != nil { w.logger.Error("post failure comment", "job_id", job.ID, "error", postErr) } return w.store.FinishJob(ctx, job.ID, run.ID, false, false, nil, fmt.Errorf("%s", errorText)) } func commandFromJob(job domain.Job, aliases map[string]bool) domain.ParsedCommand { if parsed, ok := commands.Parse(job.TriggerCommentBody, aliases); ok { return parsed } args := strings.Fields(job.CommandArgs) return domain.ParsedCommand{Name: job.Command, Raw: job.TriggerCommentBody, Arguments: args, Mode: "summary", Full: contains(args, "--full")} } func contains(items []string, needle string) bool { for _, item := range items { if item == needle { return true } } return false } func helpComment(comments []map[string]any, bot string, pending int) string { bot = strings.ToLower(strings.TrimSpace(bot)) human, bots := 0, 0 lines := []string{"## Codex Help", "", "Supported commands:", "- `@codex review [security|performance|tests] [--full]`", "- `@codex rerun`", "- `@codex explain`", "- `@codex ignore`", "- `@codex -h` / `@codex --help` / `@codex help`", "", "Status note:", fmt.Sprintf("- Pending jobs on this PR: `%d`", pending), "", fmt.Sprintf("Discussion summary (%d comments):", len(comments))} for _, c := range comments { user := "unknown" if obj, ok := c["user"].(map[string]any); ok { if v, ok := obj["username"].(string); ok && v != "" { user = v } else if v, ok := obj["login"].(string); ok { user = v } } if strings.ToLower(user) == bot { bots++ } else { human++ } body, _ := c["body"].(string) body = strings.Join(strings.Fields(body), " ") if body != "" { if len(body) > 180 { body = body[:180] + "..." } lines = append(lines, fmt.Sprintf("- @%s: %s", user, body)) } } lines[11] = fmt.Sprintf("Discussion summary (%d comments, human `%d`, bot `%d`):", len(comments), human, bots) return strings.Join(lines, "\n") } func sleep(ctx context.Context, duration time.Duration) { timer := time.NewTimer(duration) defer timer.Stop() select { case <-ctx.Done(): case <-timer.C: } }