diple/mutation_queue_test.go

package main

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
	"time"
)

func failingMutationQueuePath(t *testing.T) string {
	t.Helper()
	blocker := filepath.Join(t.TempDir(), "not-a-directory")
	if err := os.WriteFile(blocker, []byte("block"), 0o600); err != nil {
		t.Fatal(err)
	}
	return filepath.Join(blocker, "mutation-queue.json")
}

func TestMutationQueuePersistsFIFOAndCapturedGates(t *testing.T) {
	path := filepath.Join(t.TempDir(), "mutation-queue.json")
	store := loadMutationQueue(path)
	first := mutationOperation{
		ID: "first", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "reply", ThreadCanReply: true,
		Permissions: ViewerPermissions{CanReplyAny: true}, EnqueuedAt: time.Now(),
	}
	second := mutationOperation{
		ID: "second", Kind: mutationResolution, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Resolved: true, ThreadCanResolve: true, EnqueuedAt: time.Now(),
	}
	if err := store.add(first); err != nil {
		t.Fatal(err)
	}
	if err := store.add(second); err != nil {
		t.Fatal(err)
	}
	reloaded := loadMutationQueue(path)
	operations := reloaded.list()
	if len(operations) != 2 || operations[0].ID != "first" || operations[1].ID != "second" ||
		!operations[0].ThreadCanReply || !operations[0].Permissions.CanReplyAny {
		t.Fatalf("reloaded queue = %#v", operations)
	}
}

func TestMutationQueueRollsBackMemoryWhenPersistenceFails(t *testing.T) {
	path := filepath.Join(t.TempDir(), "mutation-queue.json")
	store := loadMutationQueue(path)
	first := mutationOperation{ID: "first", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1}
	second := mutationOperation{ID: "second", Kind: mutationResolution, Owner: "o", Repository: "r", Number: 2}
	if err := store.add(first); err != nil {
		t.Fatal(err)
	}
	if err := store.add(second); err != nil {
		t.Fatal(err)
	}
	store.path = failingMutationQueuePath(t)

	changed := first
	changed.Attempted = true
	if err := store.update(changed); err == nil {
		t.Fatal("update unexpectedly succeeded")
	}
	if got, _ := store.get("first"); got.Attempted {
		t.Fatalf("failed update remained in memory: %#v", got)
	}

	if err := store.remove("first"); err == nil {
		t.Fatal("remove unexpectedly succeeded")
	}
	if operations := store.list(); len(operations) != 2 || operations[0].ID != "first" || operations[1].ID != "second" {
		t.Fatalf("failed remove changed memory: %#v", operations)
	}

	if err := store.removePR("o", "r", 1); err == nil {
		t.Fatal("removePR unexpectedly succeeded")
	}
	if operations := store.list(); len(operations) != 2 || operations[0].ID != "first" || operations[1].ID != "second" {
		t.Fatalf("failed removePR changed memory: %#v", operations)
	}

	reloaded := loadMutationQueue(path)
	if operations := reloaded.list(); len(operations) != 2 || operations[0].ID != "first" || operations[1].ID != "second" {
		t.Fatalf("durable queue changed after failed writes: %#v", operations)
	}
}

func TestMutationQueueSkipsPersistenceForNoOpChanges(t *testing.T) {
	path := filepath.Join(t.TempDir(), "mutation-queue.json")
	store := loadMutationQueue(path)
	operation := mutationOperation{
		ID: "first", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	store.path = failingMutationQueuePath(t)
	if err := store.update(operation); err != nil {
		t.Fatalf("identical update attempted persistence: %v", err)
	}
	if err := store.removePR("different", "repository", 99); err != nil {
		t.Fatalf("non-matching removePR attempted persistence: %v", err)
	}
	if stored, ok := store.front(); !ok || !reflect.DeepEqual(stored, operation) {
		t.Fatalf("no-op changes altered the queue: %#v", stored)
	}
}

func TestCorruptMutationQueueRefusesToOverwriteUserData(t *testing.T) {
	path := filepath.Join(t.TempDir(), "mutation-queue.json")
	if err := atomicWriteJSON(path, map[string]any{"broken": true}, 0o600); err != nil {
		t.Fatal(err)
	}
	store := loadMutationQueue(path)
	if store.loadErr == nil {
		t.Fatal("corrupt queue was accepted")
	}
	if err := store.add(mutationOperation{Kind: mutationReply}); err == nil {
		t.Fatal("corrupt queue was overwritten by a new mutation")
	}
}

func TestQueuedMutationsProjectWithoutChangingSnapshot(t *testing.T) {
	store := loadMutationQueue("")
	now := time.Now()
	for _, operation := range []mutationOperation{
		{ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
			ThreadID: "thread", Body: "pending body", Viewer: "me", EnqueuedAt: now},
		{ID: "resolve", Kind: mutationResolution, Owner: "o", Repository: "r", Number: 1,
			ThreadID: "thread", Resolved: true, EnqueuedAt: now.Add(time.Second)},
		{ID: "edit", Kind: mutationPREdit, Owner: "o", Repository: "r", Number: 1,
			Update: PullRequestMetadata{Title: "queued", Body: "body", BaseRef: "next",
				Reviewers: []string{"reviewer"}, Assignees: []string{"assignee"}}, EnqueuedAt: now},
	} {
		if err := store.add(operation); err != nil {
			t.Fatal(err)
		}
	}
	m := NewApp(nil, "o", "r", false, 50, time.Minute)
	m.mutations = store
	snapshot := PRDetails{
		PullRequest: PullRequest{Owner: "o", Repository: "r", Number: 1, Title: "remote"},
		Threads:     []ReviewThread{{ID: "thread"}},
	}
	projected := m.projectQueuedMutations(snapshot)
	if snapshot.Title != "remote" || snapshot.Threads[0].IsResolved || len(snapshot.Threads[0].Comments) != 0 {
		t.Fatalf("source snapshot was mutated: %#v", snapshot)
	}
	if projected.Title != "queued" || projected.Pending || !projected.Threads[0].IsResolved ||
		projected.Threads[0].Pending || len(projected.Threads[0].Comments) != 1 ||
		projected.Threads[0].Comments[0].Pending {
		t.Fatalf("projection = %#v", projected)
	}
}

func TestCachedGrantedPermissionQueuesReply(t *testing.T) {
	store := loadMutationQueue("")
	settings := defaultAppSettings()
	settings.Mutations = store
	m := NewAppWithSettings(&recordingService{}, "o", "r", false, 50, time.Minute, settings)
	m.loading, m.screen = false, threadScreen
	m.details = PRDetails{
		PullRequest: PullRequest{ID: "pr", Owner: "o", Repository: "r", Number: 1}, FromCache: true,
		ViewerLogin: "me", Permissions: ViewerPermissions{CanReplyAny: true},
		Threads: []ReviewThread{{ID: "thread", ViewerCanReply: true}},
	}
	m.writeMode, m.writeThreadID, m.replyDraft = writeReplyBusy, "thread", "offline reply"
	message := m.submitReply()()
	updated, command := m.Update(message)
	m = updated.(App)
	if store.count() != 1 || m.writeMode != writeNone ||
		len(m.details.Threads[0].Comments) != 1 || m.details.Threads[0].Comments[0].Pending {
		t.Fatalf("queued cached reply: count=%d mode=%d details=%#v command=%v",
			store.count(), m.writeMode, m.details, command)
	}
}

func TestReplayChecksLivePermissionBeforeMutation(t *testing.T) {
	service := &recordingService{}
	store := loadMutationQueue("")
	operation := mutationOperation{
		ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", Viewer: "me", ThreadCanReply: true, EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	details := PRDetails{
		PullRequest: PullRequest{Owner: "o", Repository: "r", Number: 1},
		Threads:     []ReviewThread{{ID: "thread", ViewerCanReply: false}},
	}
	result := executeQueuedMutation(context.Background(), service, store, operation, details)
	if result.state != mutationReplayBlocked || service.writeBody != "" ||
		!strings.Contains(result.reason, "no longer grants reply permission") {
		t.Fatalf("permission replay result = %#v service=%#v", result, service)
	}
}

func TestReplyIsNotSentUnlessAttemptCheckpointIsDurable(t *testing.T) {
	path := filepath.Join(t.TempDir(), "mutation-queue.json")
	store := loadMutationQueue(path)
	operation := mutationOperation{
		ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", Viewer: "me", EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	store.path = failingMutationQueuePath(t)
	service := &recordingService{}
	details := PRDetails{Threads: []ReviewThread{{ID: "thread", ViewerCanReply: true}}}
	result := executeQueuedMutation(context.Background(), service, store, operation, details)
	if result.state != mutationReplayBlocked || result.err == nil ||
		result.reason != "could not checkpoint the reply attempt" || service.writeBody != "" {
		t.Fatalf("reply checkpoint result=%#v service=%#v", result, service)
	}
	stored, _ := store.front()
	if stored.Attempted {
		t.Fatalf("failed attempt checkpoint remained in memory: %#v", stored)
	}
}

func TestRepeatedBlockedReplayDoesNotRewriteSnapshot(t *testing.T) {
	path := filepath.Join(t.TempDir(), "mutation-queue.json")
	store := loadMutationQueue(path)
	operation := mutationOperation{
		ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", Viewer: "me", Attempted: true, EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	details := PRDetails{Threads: []ReviewThread{{ID: "thread", ViewerCanReply: true}}}
	first := executeQueuedMutation(context.Background(), &recordingService{}, store, operation, details)
	if first.state != mutationReplayBlocked || first.err != nil {
		t.Fatalf("initial blocked replay = %#v", first)
	}
	blocked, _ := store.front()
	if !blocked.Blocked || !blocked.Ambiguous {
		t.Fatalf("blocked state was not persisted: %#v", blocked)
	}

	store.path = failingMutationQueuePath(t)
	repeated := executeQueuedMutation(context.Background(), &recordingService{}, store, blocked, details)
	if repeated.state != mutationReplayBlocked || repeated.err != nil {
		t.Fatalf("identical blocked replay attempted persistence: %#v", repeated)
	}
}

func TestResolutionIsNotSentUnlessAttemptCheckpointIsDurable(t *testing.T) {
	path := filepath.Join(t.TempDir(), "mutation-queue.json")
	store := loadMutationQueue(path)
	operation := mutationOperation{
		ID: "resolution", Kind: mutationResolution, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Resolved: true, EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	store.path = failingMutationQueuePath(t)
	service := &recordingService{}
	details := PRDetails{Threads: []ReviewThread{{ID: "thread", ViewerCanResolve: true}}}
	result := executeQueuedMutation(context.Background(), service, store, operation, details)
	if result.state != mutationReplayBlocked || result.err == nil ||
		result.reason != "could not checkpoint the thread update attempt" || service.writeThreadID != "" {
		t.Fatalf("resolution checkpoint result=%#v service=%#v", result, service)
	}
	stored, _ := store.front()
	if stored.Attempted {
		t.Fatalf("failed attempt checkpoint remained in memory: %#v", stored)
	}
}

func TestReplayReconcilesReplyBeforeAskingAboutAmbiguousDelivery(t *testing.T) {
	store := loadMutationQueue("")
	enqueued := time.Now().Add(-time.Minute)
	operation := mutationOperation{
		ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", Viewer: "me", Attempted: true, EnqueuedAt: enqueued,
	}
	details := PRDetails{Threads: []ReviewThread{{
		ID: "thread", ViewerCanReply: true,
		Comments: []ReviewComment{{Author: "me", Body: "body", CreatedAt: enqueued.Add(time.Second)}},
	}}}
	result := executeQueuedMutation(context.Background(), &recordingService{}, store, operation, details)
	if result.state != mutationReplayApplied {
		t.Fatalf("reconciled reply = %#v", result)
	}
	details.Threads[0].Comments = append(details.Threads[0].Comments, details.Threads[0].Comments[0])
	result = executeQueuedMutation(context.Background(), &recordingService{}, store, operation, details)
	if result.state != mutationReplayBlocked || !result.operation.Ambiguous {
		t.Fatalf("duplicate matching replies were not treated as ambiguous: %#v", result)
	}

	details.Threads[0].Comments = nil
	result = executeQueuedMutation(context.Background(), &recordingService{}, store, operation, details)
	if result.state != mutationReplayBlocked || !result.operation.Ambiguous {
		t.Fatalf("ambiguous reply = %#v", result)
	}
}

func TestReplyProjectionUsesRemoteCommentWithoutTemporaryDuplicate(t *testing.T) {
	store := loadMutationQueue("")
	enqueued := time.Now().Add(-time.Second)
	operation := mutationOperation{
		ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", Viewer: "me", EnqueuedAt: enqueued,
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	m := NewApp(nil, "o", "r", false, 50, time.Minute)
	m.mutations = store
	details := PRDetails{
		PullRequest: PullRequest{Owner: "o", Repository: "r", Number: 1},
		Threads: []ReviewThread{{ID: "thread", Comments: []ReviewComment{
			{ID: "remote", Author: "me", Body: "body", CreatedAt: enqueued.Add(time.Second)},
		}}},
	}

	projected := m.projectQueuedMutations(details)
	if len(projected.Threads[0].Comments) != 1 || projected.Threads[0].Comments[0].ID != "remote" {
		t.Fatalf("single remote reply was duplicated: %#v", projected.Threads[0].Comments)
	}

	details.Threads[0].Comments = append(details.Threads[0].Comments, ReviewComment{
		ID: "actual-duplicate", Author: "me", Body: "body", CreatedAt: enqueued.Add(2 * time.Second),
	})
	projected = m.projectQueuedMutations(details)
	if len(projected.Threads[0].Comments) != 2 {
		t.Fatalf("remote duplicates were not preserved exactly: %#v", projected.Threads[0].Comments)
	}
}

func TestSuccessfulReplyCheckpointsRemoteCommentID(t *testing.T) {
	store := loadMutationQueue("")
	operation := mutationOperation{
		ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", Viewer: "me", EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	details := PRDetails{Threads: []ReviewThread{{ID: "thread", ViewerCanReply: true}}}
	result := executeQueuedMutation(context.Background(), &recordingService{}, store, operation, details)
	stored, ok := store.front()
	if result.state != mutationReplayVerifying || !ok || stored.ReplyID != "new-comment" {
		t.Fatalf("reply verification identity was not checkpointed: result=%#v stored=%#v", result, stored)
	}
}

func TestSuccessfulResolutionWaitsForLiveVerification(t *testing.T) {
	store := loadMutationQueue("")
	operation := mutationOperation{
		ID: "resolution", Kind: mutationResolution, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Resolved: true, EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	service := &recordingService{}
	details := PRDetails{Threads: []ReviewThread{{ID: "thread", ViewerCanResolve: true}}}

	result := executeQueuedMutation(context.Background(), service, store, operation, details)
	if result.state != mutationReplayVerifying || !service.writeResolved {
		t.Fatalf("successful resolution = %#v service=%#v", result, service)
	}
	stored, ok := store.front()
	if !ok || !stored.AwaitingVerification || stored.Unverified || stored.Blocked {
		t.Fatalf("resolution awaiting verification = %#v", stored)
	}

	details.Threads[0].IsResolved = true
	result = executeQueuedMutation(context.Background(), service, store, stored, details)
	if result.state != mutationReplayApplied {
		t.Fatalf("verified resolution = %#v", result)
	}
}

func TestRetryableReplyFailureRemainsOptimisticallyApplied(t *testing.T) {
	store := loadMutationQueue("")
	operation := mutationOperation{
		ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", Viewer: "me", EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	details := PRDetails{
		PullRequest: PullRequest{Owner: "o", Repository: "r", Number: 1},
		Threads:     []ReviewThread{{ID: "thread", ViewerCanReply: true}},
	}
	result := executeQueuedMutation(
		context.Background(), &failingReplyService{}, store, operation, details,
	)
	if result.state != mutationReplayWaiting || result.err == nil {
		t.Fatalf("retryable reply = %#v", result)
	}
	m := NewApp(nil, "o", "r", false, 50, time.Minute)
	m.mutations = store
	projected := m.projectQueuedMutations(details)
	if len(projected.Threads[0].Comments) != 1 || projected.Threads[0].Comments[0].Pending {
		t.Fatalf("retryable reply was not optimistic: %#v", projected.Threads[0].Comments)
	}
}

func TestFailedReplyVerificationMarksOptimisticCommentForAttention(t *testing.T) {
	store := loadMutationQueue("")
	operation := mutationOperation{
		ID: "reply", Kind: mutationReply, Owner: "o", Repository: "r", Number: 1,
		ThreadID: "thread", Body: "body", Viewer: "me", Attempted: true,
		AwaitingVerification: true, EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	details := PRDetails{
		PullRequest: PullRequest{Owner: "o", Repository: "r", Number: 1},
		Threads:     []ReviewThread{{ID: "thread", ViewerCanReply: true}},
	}
	result := executeQueuedMutation(context.Background(), &recordingService{}, store, operation, details)
	if result.state != mutationReplayBlocked || !result.operation.Unverified {
		t.Fatalf("failed verification = %#v", result)
	}
	m := NewApp(nil, "o", "r", false, 50, time.Minute)
	m.mutations = store
	projected := m.projectQueuedMutations(details)
	if len(projected.Threads[0].Comments) != 1 || !projected.Threads[0].Comments[0].Pending {
		t.Fatalf("unverified reply was not marked for attention: %#v", projected.Threads[0].Comments)
	}
}

func TestQueuedPREditThreeWayMergeOnlyBlocksConflictingFields(t *testing.T) {
	base := PullRequestMetadata{Title: "old", Body: "old body", BaseRef: "main"}
	desired := base
	desired.Title = "queued title"
	remote := base
	remote.Body = "remote body"
	merged, conflict := rebaseQueuedPREdit(base, desired, remote)
	if conflict != "" || merged.Title != "queued title" || merged.Body != "remote body" {
		t.Fatalf("non-conflicting merge = %#v conflict=%q", merged, conflict)
	}
	remote.Title = "remote title"
	_, conflict = rebaseQueuedPREdit(base, desired, remote)
	if conflict != "title" {
		t.Fatalf("conflict = %q, want title", conflict)
	}
}

func TestBlockedPREditCanBeReviewedAndReplacedInPlace(t *testing.T) {
	store := loadMutationQueue("")
	operation := mutationOperation{
		ID: "edit", Kind: mutationPREdit, Owner: "o", Repository: "r", Number: 1, PRID: "pr",
		Original: PullRequestMetadata{Title: "old", Body: "body", BaseRef: "main"},
		Update:   PullRequestMetadata{Title: "queued", Body: "body", BaseRef: "main"},
		Blocked:  true, LastError: "title changed", EnqueuedAt: time.Now(),
	}
	if err := store.add(operation); err != nil {
		t.Fatal(err)
	}
	settings := defaultAppSettings()
	settings.Mutations = store
	m := NewAppWithSettings(&recordingPRService{}, "o", "r", false, 50, time.Minute, settings)
	m.loading, m.blockedMutation, m.writeMode = false, &operation, writeQueueBlocked
	m.blockedMutationDetails = PRDetails{
		PullRequest: PullRequest{ID: "pr", Owner: "o", Repository: "r", Number: 1, Title: "remote"},
		Body:        "body", BaseRef: "main", Permissions: ViewerPermissions{CanUpdatePR: true},
	}
	command := m.resolveBlockedMutation(1)
	if command == nil || m.writeMode != writePREdit || m.editingMutationID != "edit" ||
		m.prEditEditors[prEditTitleField].Text != "queued" || m.prEditOriginal.Title != "remote" {
		t.Fatalf("reviewed edit mode=%d id=%q title=%q original=%#v command=%v",
			m.writeMode, m.editingMutationID, m.prEditEditors[prEditTitleField].Text, m.prEditOriginal, command)
	}
	m.prEditEditors[prEditTitleField].Text = "reconciled"
	message := m.submitPREdit()()
	if queued, ok := message.(mutationQueuedMsg); !ok || queued.err != nil {
		t.Fatalf("replacement message = %#v", message)
	}
	replaced, ok := store.front()
	if !ok || store.count() != 1 || replaced.ID != "edit" || replaced.Blocked ||
		replaced.Original.Title != "remote" || replaced.Update.Title != "reconciled" {
		t.Fatalf("replaced operation = %#v", replaced)
	}
}

type failingReplyService struct{ recordingService }

func (s *failingReplyService) ReplyToThread(context.Context, string, string) (ReviewComment, error) {
	return ReviewComment{}, errors.New("connection lost")
}