feat: queued mutations while reloading or offline

2026-08-03 12:14:31 +02:00
parent 63b6a645e0
commit 312b25fd39
11 changed files with 1247 additions and 28 deletions
--- a/mutation_queue.go
+++ b/mutation_queue.go
@@ -0,0 +1,624 @@
+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"`
+	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"`
+	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
+	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.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}
+	}
+	thread := findReviewThread(details.Threads, operation.ThreadID)
+	switch operation.Kind {
+	case mutationReply:
+		if queuedReplyPresent(details, operation) {
+			return mutationReplayMsg{operation: operation, state: mutationReplayApplied, details: details}
+		}
+		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,
+			}
+		}
+		if _, err := writer.ReplyToThread(ctx, operation.ThreadID, operation.Body); err != nil {
+			return mutationReplayMsg{operation: operation, state: mutationReplayWaiting, details: details, err: err}
+		}
+		return mutationReplayMsg{operation: operation, state: mutationReplayApplied, details: details}
+	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 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 mutationReplayMsg{operation: operation, state: mutationReplayApplied, details: details}
+	case mutationPREdit:
+		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 mutationReplayMsg{operation: operation, state: mutationReplayApplied, details: details}
+	default:
+		return blocked("queued mutation has an unsupported kind", false)
+	}
+}
+
+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 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 _, comment := range thread.Comments {
+				found = found || comment.ID == pendingID
+			}
+			if !found {
+				thread.Comments = append(thread.Comments, ReviewComment{
+					ID: pendingID, Author: operation.Viewer, Body: operation.Body,
+					CreatedAt: operation.EnqueuedAt, Pending: true,
+				})
+			}
+		case mutationResolution:
+			if thread := findReviewThread(details.Threads, operation.ThreadID); thread != nil {
+				thread.IsResolved, thread.Pending = operation.Resolved, true
+			}
+		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 = true
+		}
+	}
+	return details
+}
+
+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, result[index].Pending = operation.Update.Title, true
+			}
+		}
+	}
+	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.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()
+}