package main import ( "context" "encoding/json" "errors" "fmt" "os" "reflect" "slices" "strings" "sync" "sync/atomic" "time" tea "github.com/charmbracelet/bubbletea" ) const mutationQueueSchemaVersion = 1 type mutationKind string const ( mutationReply mutationKind = "reply" mutationResolution mutationKind = "resolution" mutationPREdit mutationKind = "pr-edit" ) type mutationOperation struct { ID string `json:"id"` Kind mutationKind `json:"kind"` Owner string `json:"owner"` Repository string `json:"repository"` Number int `json:"number"` PRID string `json:"pull_request_id"` ThreadID string `json:"thread_id,omitempty"` Body string `json:"body,omitempty"` ReplyID string `json:"reply_id,omitempty"` Resolved bool `json:"resolved,omitempty"` Viewer string `json:"viewer,omitempty"` Original PullRequestMetadata `json:"original,omitempty"` Update PullRequestMetadata `json:"update,omitempty"` Permissions ViewerPermissions `json:"permissions"` ThreadCanReply bool `json:"thread_can_reply,omitempty"` ThreadCanResolve bool `json:"thread_can_resolve,omitempty"` ThreadCanUnresolve bool `json:"thread_can_unresolve,omitempty"` PeopleDone bool `json:"people_done,omitempty"` Attempted bool `json:"attempted,omitempty"` AwaitingVerification bool `json:"awaiting_verification,omitempty"` Unverified bool `json:"unverified,omitempty"` Blocked bool `json:"blocked,omitempty"` Ambiguous bool `json:"ambiguous,omitempty"` LastError string `json:"last_error,omitempty"` EnqueuedAt time.Time `json:"enqueued_at"` } type mutationQueueEnvelope struct { Version int `json:"version"` Operations []mutationOperation `json:"operations"` } type mutationQueueStore struct { mu sync.Mutex path string operations []mutationOperation loadErr error } func loadMutationQueue(path string) *mutationQueueStore { store := &mutationQueueStore{path: path} if path == "" { return store } data, err := os.ReadFile(path) if errors.Is(err, os.ErrNotExist) { return store } if err != nil { store.loadErr = err return store } var envelope mutationQueueEnvelope if err := json.Unmarshal(data, &envelope); err != nil || envelope.Version != mutationQueueSchemaVersion { store.loadErr = errors.New("mutation queue is corrupt or has an unsupported schema version") return store } store.operations = envelope.Operations return store } var mutationSequence atomic.Uint64 func newMutationID() string { return fmt.Sprintf("%d-%d", time.Now().UnixNano(), mutationSequence.Add(1)) } func (s *mutationQueueStore) add(operation mutationOperation) error { if s == nil { return errors.New("mutation queue is unavailable") } s.mu.Lock() defer s.mu.Unlock() if s.loadErr != nil { return fmt.Errorf("mutation queue unavailable: %w", s.loadErr) } if operation.ID == "" { operation.ID = newMutationID() } if operation.EnqueuedAt.IsZero() { operation.EnqueuedAt = time.Now() } s.operations = append(s.operations, operation) if err := s.flushLocked(); err != nil { s.operations = s.operations[:len(s.operations)-1] return err } return nil } func (s *mutationQueueStore) front() (mutationOperation, bool) { if s == nil { return mutationOperation{}, false } s.mu.Lock() defer s.mu.Unlock() if len(s.operations) == 0 { return mutationOperation{}, false } return s.operations[0], true } func (s *mutationQueueStore) get(id string) (mutationOperation, bool) { for _, operation := range s.list() { if operation.ID == id { return operation, true } } return mutationOperation{}, false } func (s *mutationQueueStore) list() []mutationOperation { if s == nil { return nil } s.mu.Lock() defer s.mu.Unlock() return slices.Clone(s.operations) } func (s *mutationQueueStore) count() int { return len(s.list()) } func (s *mutationQueueStore) update(operation mutationOperation) error { if s == nil { return errors.New("mutation queue is unavailable") } s.mu.Lock() defer s.mu.Unlock() for index := range s.operations { if s.operations[index].ID == operation.ID { if reflect.DeepEqual(s.operations[index], operation) { return nil } previous := s.operations[index] s.operations[index] = operation if err := s.flushLocked(); err != nil { s.operations[index] = previous return err } return nil } } return errors.New("queued mutation no longer exists") } func (s *mutationQueueStore) remove(id string) error { if s == nil { return nil } s.mu.Lock() defer s.mu.Unlock() for index := range s.operations { if s.operations[index].ID == id { previous := slices.Clone(s.operations) s.operations = append(s.operations[:index], s.operations[index+1:]...) if err := s.flushLocked(); err != nil { s.operations = previous return err } return nil } } return nil } func (s *mutationQueueStore) removePR(owner, repo string, number int) error { if s == nil { return nil } s.mu.Lock() defer s.mu.Unlock() previous := slices.Clone(s.operations) filtered := make([]mutationOperation, 0, len(s.operations)) removed := false for _, operation := range s.operations { if operation.Owner != owner || operation.Repository != repo || operation.Number != number { filtered = append(filtered, operation) } else { removed = true } } if !removed { return nil } s.operations = filtered if err := s.flushLocked(); err != nil { s.operations = previous return err } return nil } func (s *mutationQueueStore) flushLocked() error { if s.path == "" { return nil } return atomicWriteJSON(s.path, mutationQueueEnvelope{ Version: mutationQueueSchemaVersion, Operations: s.operations, }, 0o600) } type mutationQueuedMsg struct { operation mutationOperation err error } type mutationReplayState int const ( mutationReplayApplied mutationReplayState = iota mutationReplayVerifying mutationReplayWaiting mutationReplayBlocked ) type mutationReplayMsg struct { operation mutationOperation state mutationReplayState details PRDetails reason string err error } func (m App) enqueueMutation(operation mutationOperation) tea.Cmd { return func() tea.Msg { operation.ID = newMutationID() operation.EnqueuedAt = time.Now() err := m.mutations.add(operation) return mutationQueuedMsg{operation: operation, err: err} } } func (m App) replaceMutation(operation mutationOperation) tea.Cmd { return func() tea.Msg { err := m.mutations.update(operation) return mutationQueuedMsg{operation: operation, err: err} } } func (m App) replayNextMutation() tea.Cmd { if m.mutations == nil { return nil } operation, ok := m.mutations.front() if !ok { return nil } service := m.service store := m.mutations return func() tea.Msg { ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() var details PRDetails var err error if live, ok := service.(liveGitHubService); ok { details, err = live.LivePullRequest(ctx, operation.Owner, operation.Repository, operation.Number) } else { details, err = service.GetPullRequest(ctx, operation.Owner, operation.Repository, operation.Number) } if err != nil || details.FromCache { return mutationReplayMsg{operation: operation, state: mutationReplayWaiting, err: err} } return executeQueuedMutation(ctx, service, store, operation, details) } } func (m *App) startMutationReplay() tea.Cmd { if m.mutations == nil || m.mutationReplayBusy || m.mutations.count() == 0 { return nil } m.mutationReplayBusy = true return m.replayNextMutation() } func executeQueuedMutation( ctx context.Context, service GitHubService, store *mutationQueueStore, operation mutationOperation, details PRDetails, ) mutationReplayMsg { blocked := func(reason string, ambiguous bool) mutationReplayMsg { operation.AwaitingVerification = false operation.Unverified = true if operation.Kind == mutationPREdit { operation.PeopleDone = false } operation.Blocked, operation.Ambiguous, operation.LastError = true, ambiguous, reason if err := store.update(operation); err != nil { return mutationReplayMsg{operation: operation, state: mutationReplayBlocked, details: details, reason: reason, err: err} } return mutationReplayMsg{operation: operation, state: mutationReplayBlocked, details: details, reason: reason} } verifying := func() mutationReplayMsg { operation.AwaitingVerification = true operation.Blocked, operation.Ambiguous, operation.LastError = false, false, "" if err := store.update(operation); err != nil { return blocked("could not checkpoint successful mutation for verification", true) } return mutationReplayMsg{ operation: operation, state: mutationReplayVerifying, details: details, } } markUnverified := func() *mutationReplayMsg { if !operation.AwaitingVerification { return nil } operation.AwaitingVerification = false operation.Unverified = true if operation.Kind == mutationPREdit { operation.PeopleDone = false } if err := store.update(operation); err != nil { result := blocked("could not record failed mutation verification", true) return &result } return nil } thread := findReviewThread(details.Threads, operation.ThreadID) switch operation.Kind { case mutationReply: if queuedReplyPresent(details, operation) { return mutationReplayMsg{operation: operation, state: mutationReplayApplied, details: details} } if result := markUnverified(); result != nil { return *result } if thread == nil { return blocked("the review thread no longer exists", false) } if !thread.ViewerCanReply { return blocked("GitHub no longer grants reply permission for this thread", false) } if operation.Attempted { return blocked("GitHub data cannot prove whether the previous reply attempt was applied", true) } writer, ok := service.(GitHubWriteService) if !ok { return blocked("configured GitHub service no longer supports replies", false) } operation.Attempted = true if err := store.update(operation); err != nil { operation.Attempted = false return mutationReplayMsg{ operation: operation, state: mutationReplayBlocked, details: details, reason: "could not checkpoint the reply attempt", err: err, } } comment, err := writer.ReplyToThread(ctx, operation.ThreadID, operation.Body) if err != nil { return mutationReplayMsg{operation: operation, state: mutationReplayWaiting, details: details, err: err} } operation.ReplyID = comment.ID return verifying() case mutationResolution: if thread == nil { return blocked("the review thread no longer exists", false) } if thread.IsResolved == operation.Resolved { return mutationReplayMsg{operation: operation, state: mutationReplayApplied, details: details} } if result := markUnverified(); result != nil { return *result } if operation.Resolved && !thread.ViewerCanResolve { return blocked("GitHub no longer grants resolve permission for this thread", false) } if !operation.Resolved && !thread.ViewerCanUnresolve { return blocked("GitHub no longer grants unresolve permission for this thread", false) } writer, ok := service.(GitHubWriteService) if !ok { return blocked("configured GitHub service no longer supports thread updates", false) } operation.Attempted = true if err := store.update(operation); err != nil { operation.Attempted = false return mutationReplayMsg{ operation: operation, state: mutationReplayBlocked, details: details, reason: "could not checkpoint the thread update attempt", err: err, } } if _, err := writer.SetThreadResolved(ctx, operation.ThreadID, operation.Resolved); err != nil { return mutationReplayMsg{operation: operation, state: mutationReplayWaiting, details: details, err: err} } return verifying() case mutationPREdit: if queuedPREditPresent(details, operation.Update) { return mutationReplayMsg{operation: operation, state: mutationReplayApplied, details: details} } if result := markUnverified(); result != nil { return *result } if !details.Permissions.CanUpdatePR { return blocked("GitHub no longer grants permission to update this pull request", false) } update, conflict := rebaseQueuedPREdit(operation.Original, operation.Update, currentMetadata(details)) if conflict != "" { return blocked("pull request fields changed on GitHub: "+conflict, false) } peopleWriter, peopleOK := service.(GitHubPullRequestPeopleWriteService) writer, writeOK := service.(GitHubPullRequestWriteService) if !peopleOK || !writeOK { return blocked("configured GitHub service no longer supports pull request updates", false) } if !operation.PeopleDone && (!slices.Equal(update.Reviewers, details.RequestedReviewers) || !slices.Equal(update.Assignees, details.Assignees)) { people, err := peopleWriter.UpdatePullRequestPeople(ctx, operation.Owner, operation.Repository, operation.Number, PullRequestPeopleUpdate{ CurrentReviewers: slices.Clone(details.RequestedReviewers), CurrentAssignees: slices.Clone(details.Assignees), Reviewers: slices.Clone(update.Reviewers), Assignees: slices.Clone(update.Assignees), }) if err != nil { return mutationReplayMsg{operation: operation, state: mutationReplayWaiting, details: details, err: err} } details.RequestedReviewers, details.Assignees = people.Reviewers, people.Assignees operation.PeopleDone = true if err := store.update(operation); err != nil { return blocked("could not checkpoint the applied reviewer and assignee update", true) } } if !samePRMetadataCore(update, currentMetadata(details)) { result, err := writer.UpdatePullRequest(ctx, operation.PRID, update) if err != nil { return mutationReplayMsg{operation: operation, state: mutationReplayWaiting, details: details, err: err} } details.Title, details.Body, details.BaseRef = result.Title, result.Body, result.BaseRef } return verifying() default: return blocked("queued mutation has an unsupported kind", false) } } func queuedPREditPresent(details PRDetails, update PullRequestMetadata) bool { return samePRMetadataCore(update, currentMetadata(details)) && slices.Equal(normalizedLogins(update.Reviewers), normalizedLogins(details.RequestedReviewers)) && slices.Equal(normalizedLogins(update.Assignees), normalizedLogins(details.Assignees)) } func findReviewThread(threads []ReviewThread, id string) *ReviewThread { for index := range threads { if threads[index].ID == id { return &threads[index] } } return nil } func queuedReplyPresent(details PRDetails, operation mutationOperation) bool { thread := findReviewThread(details.Threads, operation.ThreadID) if thread == nil { return false } matches := 0 for _, comment := range thread.Comments { if operation.ReplyID != "" && comment.ID == operation.ReplyID { return true } if comment.Body == operation.Body && strings.EqualFold(comment.Author, operation.Viewer) && !comment.CreatedAt.Before(operation.EnqueuedAt.Add(-time.Minute)) { matches++ } } return matches == 1 } func currentMetadata(details PRDetails) PullRequestMetadata { return PullRequestMetadata{ Title: details.Title, Body: details.Body, BaseRef: details.BaseRef, Reviewers: normalizedLogins(details.RequestedReviewers), Assignees: normalizedLogins(details.Assignees), Mergeable: details.Mergeable, MergeState: details.MergeState, UpdatedAt: details.UpdatedAt, } } func rebaseQueuedPREdit(base, desired, remote PullRequestMetadata) (PullRequestMetadata, string) { result := remote conflicts := []string{} rebaseString := func(name, before, want, current string) string { switch { case want == before: return current case current == before || current == want: return want default: conflicts = append(conflicts, name) return current } } result.Title = rebaseString("title", base.Title, desired.Title, remote.Title) result.Body = rebaseString("description", base.Body, desired.Body, remote.Body) result.BaseRef = rebaseString("target branch", base.BaseRef, desired.BaseRef, remote.BaseRef) rebaseLogins := func(name string, before, want, current []string) []string { switch { case slices.Equal(want, before): return current case slices.Equal(current, before) || slices.Equal(current, want): return want default: conflicts = append(conflicts, name) return current } } result.Reviewers = rebaseLogins("reviewers", base.Reviewers, desired.Reviewers, remote.Reviewers) result.Assignees = rebaseLogins("assignees", base.Assignees, desired.Assignees, remote.Assignees) return result, strings.Join(conflicts, ", ") } func (m App) projectQueuedMutations(details PRDetails) PRDetails { details.Threads = slices.Clone(details.Threads) for index := range details.Threads { details.Threads[index].Comments = slices.Clone(details.Threads[index].Comments) } details.RequestedReviewers = slices.Clone(details.RequestedReviewers) details.Assignees = slices.Clone(details.Assignees) for _, operation := range m.mutations.list() { if operation.Owner != details.Owner || operation.Repository != details.Repository || operation.Number != details.Number { continue } switch operation.Kind { case mutationReply: thread := findReviewThread(details.Threads, operation.ThreadID) if thread == nil { continue } pendingID := "pending:" + operation.ID found := false for index := range thread.Comments { if queuedReplyMatchesComment(thread.Comments[index], operation) { found = true continue } if thread.Comments[index].ID == pendingID { thread.Comments[index].Pending = mutationNeedsAttention(operation) found = true } } if !found { thread.Comments = append(thread.Comments, ReviewComment{ ID: pendingID, Author: operation.Viewer, Body: operation.Body, CreatedAt: operation.EnqueuedAt, Pending: mutationNeedsAttention(operation), }) } case mutationResolution: if thread := findReviewThread(details.Threads, operation.ThreadID); thread != nil { thread.IsResolved = operation.Resolved thread.Pending = mutationNeedsAttention(operation) } case mutationPREdit: details.Title, details.Body, details.BaseRef = operation.Update.Title, operation.Update.Body, operation.Update.BaseRef details.RequestedReviewers = slices.Clone(operation.Update.Reviewers) details.Assignees = slices.Clone(operation.Update.Assignees) details.Reviewers = projectRequestedReviewers(details.Reviewers, operation.Update.Reviewers) details.Pending = mutationNeedsAttention(operation) } } return details } func queuedReplyMatchesComment(comment ReviewComment, operation mutationOperation) bool { if operation.ReplyID != "" && comment.ID == operation.ReplyID { return true } return comment.Body == operation.Body && strings.EqualFold(comment.Author, operation.Viewer) && !comment.CreatedAt.Before(operation.EnqueuedAt.Add(-time.Minute)) } func mutationNeedsAttention(operation mutationOperation) bool { return operation.Blocked || operation.Unverified } func (m App) projectQueuedPullRequests(prs []PullRequest) []PullRequest { result := slices.Clone(prs) for _, operation := range m.mutations.list() { if operation.Kind != mutationPREdit { continue } for index := range result { if result[index].Owner == operation.Owner && result[index].Repository == operation.Repository && result[index].Number == operation.Number { result[index].Title = operation.Update.Title result[index].Pending = mutationNeedsAttention(operation) } } } return result } func projectRequestedReviewers(current []Reviewer, requested []string) []Reviewer { result := make([]Reviewer, 0, len(current)+len(requested)) known := make(map[string]bool) for _, reviewer := range current { if reviewer.State == "REVIEW_REQUESTED" { continue } result = append(result, reviewer) known[strings.ToLower(reviewer.Login)] = true } for _, login := range requested { if key := strings.ToLower(login); !known[key] { result = append(result, Reviewer{Login: login, State: "REVIEW_REQUESTED"}) known[key] = true } } return result } func blockedMutationChoices(operation mutationOperation) []string { choices := []string{"Keep queued and retry after refresh"} if operation.Kind == mutationPREdit { choices = append(choices, "Review queued edit against current GitHub state") } if operation.Ambiguous { choices = append(choices, "Retry this mutation now", "Treat this mutation as applied") } return append(choices, "Discard this mutation and continue", "Discard all queued mutations for this pull request", ) } func (m *App) resolveBlockedMutation(choice int) tea.Cmd { if m.blockedMutation == nil || m.mutations == nil { m.writeMode = writeNone return nil } operation := *m.blockedMutation label := blockedMutationChoices(operation)[choice] m.writeMode, m.blockedMutation, m.err = writeNone, nil, nil switch label { case "Keep queued and retry after refresh": return nil case "Review queued edit against current GitHub state": m.details = m.blockedMutationDetails m.editingMutationID = operation.ID command := m.startPREdit() m.prEditEditors[prEditTitleField].Text = operation.Update.Title m.prEditEditors[prEditBaseField].Text = operation.Update.BaseRef m.prEditEditors[prEditReviewersField].Text = strings.Join(operation.Update.Reviewers, ", ") m.prEditEditors[prEditAssigneesField].Text = strings.Join(operation.Update.Assignees, ", ") m.prEditEditors[prEditBodyField].Text = operation.Update.Body for index := range m.prEditEditors { m.prEditEditors[index].Cursor = len([]rune(m.prEditEditors[index].Text)) } return command case "Retry this mutation now": operation.Attempted, operation.AwaitingVerification = false, false operation.Unverified, operation.Blocked = false, false operation.Ambiguous, operation.LastError = false, "" if err := m.mutations.update(operation); err != nil { m.err = err return nil } return m.refreshAfterQueueChange(operation) case "Treat this mutation as applied", "Discard this mutation and continue": if err := m.mutations.remove(operation.ID); err != nil { m.err = err return nil } return m.startMutationReplay() default: if err := m.mutations.removePR(operation.Owner, operation.Repository, operation.Number); err != nil { m.err = err return nil } return m.refreshAfterQueueChange(operation) } } func (m *App) refreshAfterQueueChange(operation mutationOperation) tea.Cmd { if m.details.Owner == operation.Owner && m.details.Repository == operation.Repository && m.details.Number == operation.Number { m.loading = true return m.loadDetails(m.details.PullRequest, false) } return m.startMutationReplay() }