Hold a task's merge while another task in the project is merging

internal/agents/mergein.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"slices"
 	"strings"
 	"unicode/utf8"
 
@@ -40,26 +41,169 @@ func (m *Manager) MergeIn(ctx context.Context, task api.Task, user api.User, tar
 	return m.create(ctx, task, user, prompt, "", "byre", nil)
 }
 
-func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project, user api.User) (api.AgentSession, error) {
+func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project, user api.User) (api.Task, error) {
 	if user.GitName == "" || user.GitEmail == "" {
-		return api.AgentSession{}, ErrNoGitIdentity
+		return api.Task{}, ErrNoGitIdentity
+	}
+	if task.MergeQueuedAt != nil {
+		return api.Task{}, fmt.Errorf("%w: task %s is already waiting to merge", ErrConflict, task.Id)
 	}
 
+	m.mergeMu.Lock()
+	defer m.mergeMu.Unlock()
+
+	switch m.mergeHolders[project.Id] {
+	case task.Id:
+		return api.Task{}, fmt.Errorf("%w: task %s is already merging", ErrConflict, task.Id)
+	case "":
+		err := m.startMerge(ctx, task, project, user)
+		if err != nil {
+			return api.Task{}, err
+		}
+	default:
+		err := m.store.QueueMerge(ctx, task.Id, user.Login)
+		if errors.Is(err, store.ErrNotFound) {
+			return api.Task{}, fmt.Errorf("%w: task %s is closed or already waiting to merge", ErrConflict, task.Id)
+		}
+		if err != nil {
+			return api.Task{}, err
+		}
+	}
+	return m.store.GetTask(ctx, task.Id)
+}
+
+func (m *Manager) startMerge(ctx context.Context, task api.Task, project api.Project, user api.User) error {
 	err := git.FetchFastForward(project.RepoPath, project.DefaultBranch)
 	if err != nil {
-		return api.AgentSession{}, err
+		return err
 	}
 
 	err = m.store.SetMergingBy(ctx, task.Id, user.Login)
 	if err != nil {
-		return api.AgentSession{}, err
+		return err
 	}
+	m.mergeHolders[project.Id] = task.Id
 
-	sess, err := m.MergeIn(ctx, task, user, project.DefaultBranch)
+	_, err = m.MergeIn(ctx, task, user, project.DefaultBranch)
 	if err != nil {
 		m.takeMerge(task.Id)
+		delete(m.mergeHolders, project.Id)
 	}
-	return sess, err
+	return err
+}
+
+func (m *Manager) recoverMerges(ctx context.Context) error {
+	merging, err := m.store.MergingTasks(ctx)
+	if err != nil {
+		return err
+	}
+
+	m.mergeMu.Lock()
+	defer m.mergeMu.Unlock()
+	for _, t := range merging {
+		m.mergeHolders[t.ProjectId] = t.Id
+	}
+	return nil
+}
+
+func (m *Manager) startHeldMerges(ctx context.Context) {
+	projects, err := m.store.ProjectsWithQueuedMerges(ctx)
+	if err != nil {
+		slog.Error("list projects with held merges", "err", err)
+		return
+	}
+	for _, p := range projects {
+		m.nextMerge(p)
+	}
+}
+
+func (m *Manager) abandonMerge(taskID, projectID string) {
+	if m.takeMerge(taskID) == "" {
+		return
+	}
+	go m.releaseMerge(projectID, taskID)
+}
+
+func (m *Manager) releaseMerge(projectID, taskID string) {
+	m.mergeMu.Lock()
+	defer m.mergeMu.Unlock()
+	if m.mergeHolders[projectID] == taskID {
+		delete(m.mergeHolders, projectID)
+	}
+	m.startNextMerge(projectID)
+}
+
+func (m *Manager) nextMerge(projectID string) {
+	m.mergeMu.Lock()
+	defer m.mergeMu.Unlock()
+	m.startNextMerge(projectID)
+}
+
+func (m *Manager) startNextMerge(projectID string) {
+	if m.mergeHolders[projectID] != "" {
+		return
+	}
+
+	ctx := context.Background()
+	held, err := m.store.QueuedMerges(ctx, projectID)
+	if err != nil {
+		slog.Error("list held merges", "project", projectID, "err", err)
+		return
+	}
+
+	for _, task := range held {
+		sessions, err := m.store.ListSessions(ctx, task.Id)
+		if err != nil {
+			slog.Error("list sessions for held merge", "task", task.Id, "err", err)
+			continue
+		}
+		if slices.ContainsFunc(sessions, func(s api.AgentSession) bool {
+			return s.State == api.AgentSessionStateRunning || s.State == api.AgentSessionStateQueued
+		}) {
+			continue
+		}
+
+		login, err := m.store.TakeQueuedMerge(ctx, task.Id)
+		if err != nil {
+			slog.Error("take held merge", "task", task.Id, "err", err)
+			continue
+		}
+		if login == "" {
+			continue
+		}
+
+		err = m.startHeldMerge(ctx, task, login)
+		if err == nil {
+			slog.Info("started held merge", "task", task.Id, "by", login)
+			return
+		}
+		slog.Error("start held merge", "task", task.Id, "err", err)
+		m.fileTaskError(ctx, task, sessions, "Starting the held merge failed: "+err.Error())
+	}
+}
+
+func (m *Manager) startHeldMerge(ctx context.Context, task api.Task, login string) error {
+	project, err := m.store.GetProject(ctx, task.ProjectId)
+	if err != nil {
+		return err
+	}
+	user, err := m.store.GetUser(ctx, login)
+	if err != nil {
+		return err
+	}
+	if user.GitName == "" || user.GitEmail == "" {
+		return ErrNoGitIdentity
+	}
+	return m.startMerge(ctx, task, project, user)
+}
+
+func (m *Manager) fileTaskError(ctx context.Context, task api.Task, sessions []api.AgentSession, msg string) {
+	if len(sessions) == 0 {
+		slog.Error("no session to file task error on", "task", task.Id, "error", msg)
+		return
+	}
+	last := sessions[len(sessions)-1]
+	m.fileError(&live{id: last.Id, taskID: task.Id, projectID: task.ProjectId}, msg)
 }
 
 func (m *Manager) takeMerge(taskID string) string {
@@ -71,6 +215,8 @@ func (m *Manager) takeMerge(taskID string) string {
 }
 
 func (m *Manager) finishMerge(l *live, login string) {
+	defer m.releaseMerge(l.projectID, l.taskID)
+
 	ctx := context.Background()
 	err := m.squashAndClose(ctx, l.taskID, login)
 	if err == nil {
@@ -170,7 +316,12 @@ func (m *Manager) Discard(ctx context.Context, task api.Task, project api.Projec
 	if err != nil {
 		return api.Task{}, err
 	}
-	return m.store.DiscardTask(ctx, task.Id, actor)
+	t, err := m.store.DiscardTask(ctx, task.Id, actor)
+	if err != nil {
+		return api.Task{}, err
+	}
+	go m.releaseMerge(project.Id, task.Id)
+	return t, nil
 }
 
 func (m *Manager) RemoveTask(ctx context.Context, task api.Task, project api.Project, actor string) error {