diple/mutation_queue.go

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()
}