package service

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/gitstore"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const (
	mergeQueueEntryQueued         = "queued"
	mergeQueueEntryAwaitingChecks = "awaiting_checks"
	mergeQueueEntryMergeable      = "mergeable"
	mergeQueueEntryLocked         = "locked"
	mergeQueueEntryMerged         = "merged"
	mergeQueueEntryDequeued       = "dequeued"
	mergeQueueEntryUnmergeable    = "unmergeable"

	mergeGroupPublishing      = "publishing_ref"
	mergeGroupChecksRequested = "checks_requested"
	mergeGroupPassed          = "passed"
	mergeGroupMerging         = "merging"
	mergeGroupCascading       = "cascading"
	mergeGroupMerged          = "merged"
	mergeGroupFailed          = "failed"
	mergeGroupStale           = "stale"
)

var activeMergeQueueEntryStates = []string{
	mergeQueueEntryQueued,
	mergeQueueEntryAwaitingChecks,
	mergeQueueEntryMergeable,
	mergeQueueEntryLocked,
}

// startMergeQueueProcessor detaches speculative builds from the API response.
// GitHub's enqueue mutations and asynchronous merge endpoint return before CI
// has built or merged the queue entry.
func (s *Service) startMergeQueueProcessor(parent context.Context, queueID uint, actor db.User) {
	workerKey := s.workKey(parent, queueID)
	if _, loaded := s.mergeQueueProcessors.LoadOrStore(workerKey, struct{}{}); loaded {
		return
	}
	ctx, err := s.DetachContext(parent)
	if err != nil {
		s.mergeQueueProcessors.Delete(workerKey)
		slog.WarnContext(parent, "merge queue processor not started", "queue_id", queueID, "error", err)
		return
	}
	s.Wg.Add(1)
	go func() {
		defer s.Wg.Done()
		defer s.mergeQueueProcessors.Delete(workerKey)
		if _, err := s.BuildMergeGroupForQueue(ctx, queueID, actor); err != nil {
			slog.WarnContext(ctx, "merge queue processing failed", "queue_id", queueID, "error", err)
		}
	}()
}

// BuildMergeGroupForQueue fills the configured build-concurrency slots. Each
// build belongs to one queue entry and contains the base plus every entry ahead
// of it, which is the cumulative speculative topology exposed by GitHub's
// merge_group checks_requested event.
func (s *Service) BuildMergeGroupForQueue(ctx context.Context, queueID uint, actor db.User) (*db.MergeGroup, error) {
	var groups []db.MergeGroup
	var dequeuedPullRequestIDs []uint
	for attempts := 0; attempts < 100; attempts++ {
		built, ejectedPullRequestIDs, err := s.buildMergeGroups(ctx, queueID, actor)
		if err != nil {
			return nil, err
		}
		groups = append(groups, built...)
		dequeuedPullRequestIDs = append(dequeuedPullRequestIDs, ejectedPullRequestIDs...)
		if len(ejectedPullRequestIDs) == 0 {
			break
		}
	}
	s.notifyPullRequestQueueEvent(ctx, dequeuedPullRequestIDs, "dequeued")
	if len(dequeuedPullRequestIDs) > 0 {
		if err := s.cleanupTerminalMergeGroupRefs(ctx, queueID); err != nil {
			slog.WarnContext(ctx, "merge queue ejection lifecycle recovery failed", "queue_id", queueID, "error", err)
		}
	}
	for i := range groups {
		if err := s.publishMergeGroup(ctx, groups[i], actor); err != nil {
			slog.WarnContext(ctx, "merge_group publication failed", "group_id", groups[i].ID, "error", err)
			continue
		}
		groups[i].Status = mergeGroupChecksRequested
	}
	if len(groups) == 0 {
		return nil, nil
	}
	return &groups[0], nil
}

func (s *Service) buildMergeGroups(ctx context.Context, queueID uint, actor db.User) ([]db.MergeGroup, []uint, error) {
	gitStore, err := s.GitForCtx(ctx)
	if err != nil {
		return nil, nil, err
	}
	var groups []db.MergeGroup
	var ejectedPullRequestIDs []uint
	err = s.Transaction(ctx, func(txCtx context.Context, tx *gorm.DB) error {
		var queue db.MergeQueue
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Preload("Repository.Owner").First(&queue, queueID).Error; err != nil {
			return wrapErr(err)
		}
		config, enabled, err := s.MergeQueueConfigurationForBranch(txCtx, queue.RepositoryID, queue.BaseRef)
		if err != nil {
			return err
		}
		if !enabled {
			return fmt.Errorf("%w: merge queue is no longer configured for %s", ErrInvalidState, queue.BaseRef)
		}
		var building int64
		if err := tx.Model(&db.MergeGroup{}).Where("queue_id = ? AND status IN ?", queue.ID,
			[]string{mergeGroupPublishing, mergeGroupChecksRequested}).Count(&building).Error; err != nil {
			return err
		}
		slots := config.MaxEntriesToBuild - int(building)
		if slots <= 0 {
			return nil
		}
		var entries []db.MergeQueueEntry
		if err := tx.Where("queue_id = ? AND state IN ?", queue.ID, activeMergeQueueEntryStates).
			Order("position ASC").Preload("PullRequest").Find(&entries).Error; err != nil {
			return err
		}
		if len(entries) == 0 {
			return nil
		}
		baseSHA := entries[0].BaseSHA
		if sha, headErr := gitStore.HeadSHA(ctx, queue.Repository.FullName, queue.BaseRef); headErr == nil && sha != "" {
			baseSHA = sha
		}
		for targetIndex := range entries {
			if slots == 0 {
				break
			}
			target := entries[targetIndex]
			if target.State != mergeQueueEntryQueued {
				continue
			}
			headSHA := baseSHA
			targetBaseSHA := baseSHA
			for prefixIndex := 0; prefixIndex <= targetIndex; prefixIndex++ {
				entry := entries[prefixIndex]
				if prefixIndex == targetIndex {
					targetBaseSHA = headSHA
				}
				simulated, simErr := gitStore.SimulateMerge(ctx, queue.Repository.FullName, headSHA, entry.HeadSHA)
				if simErr != nil {
					ejectedIDs, err := s.ejectMergeQueueEntryTx(tx, entry, "pull request cannot be cleanly added to a merge group", actor.ID)
					if err != nil {
						return err
					}
					ejectedPullRequestIDs = append(ejectedPullRequestIDs, ejectedIDs...)
					return nil
				}
				headSHA = simulated
			}
			shortSHA := headSHA
			if len(shortSHA) > 12 {
				shortSHA = shortSHA[:12]
			}
			headRef := fmt.Sprintf("refs/heads/gh-readonly-queue/%s/pr-%d-%s", queue.BaseRef, target.PullRequest.Number, shortSHA)
			group := db.MergeGroup{
				QueueID: queue.ID, HeadEntryID: target.ID, EntryCount: targetIndex + 1,
				ActorID: actor.ID,
				HeadSHA: headSHA, HeadRef: headRef, BaseSHA: baseSHA,
				BaseRef: "refs/heads/" + queue.BaseRef, Status: mergeGroupPublishing,
			}
			if err := tx.Create(&group).Error; err != nil {
				return err
			}
			if err := tx.Model(&db.MergeQueueEntry{}).Where("id = ? AND state = ?", target.ID, mergeQueueEntryQueued).
				Updates(map[string]any{"state": mergeQueueEntryAwaitingChecks, "merge_group_id": group.ID,
					"base_sha": targetBaseSHA, "updated_at": time.Now().UTC()}).Error; err != nil {
				return err
			}
			groups = append(groups, group)
			slots--
		}
		return nil
	})
	return groups, ejectedPullRequestIDs, err
}

func (s *Service) dispatchMergeGroup(ctx context.Context, group db.MergeGroup, actor db.User) error {
	var current db.MergeGroup
	if err := s.DBForCtx(ctx).First(&current, group.ID).Error; err != nil {
		return err
	}
	if current.Status != mergeGroupChecksRequested {
		return nil
	}
	if actor.ID == 0 {
		actor.ID = current.ActorID
	}
	if err := s.ensureMergeGroupLifecycleEvent(ctx, current, mergeGroupLifecycleChecksRequested, "", actor.ID); err != nil {
		return err
	}
	return s.processMergeGroupLifecycleEvents(ctx, current.ID)
}

func (s *Service) dispatchMergeGroupWorkflows(ctx context.Context, repo db.Repository, group db.MergeGroup, actor db.User) error {
	gitStore, err := s.GitForCtx(ctx)
	if err != nil {
		return err
	}
	files, err := gitStore.ListTreeFilesAtRef(ctx, repo.FullName, group.HeadSHA)
	if err != nil {
		return err
	}
	for _, path := range files {
		if !isWorkflowDefinitionPath(path) {
			continue
		}
		content, err := gitStore.ReadFileAtRef(ctx, repo.FullName, path, group.HeadSHA)
		if err != nil {
			if errors.Is(err, gitstore.ErrContentNotFound) {
				continue
			}
			return fmt.Errorf("read merge_group workflow %s at %s: %w", path, group.HeadSHA, err)
		}
		if !workflowYAMLMatchesMergeGroup(content, mergeGroupLifecycleChecksRequested, group.BaseRef) {
			continue
		}
		workflow, active, err := s.workflowForEventDefinition(ctx, repo.ID, path, content)
		if err != nil {
			return err
		}
		if !active {
			continue
		}
		var count int64
		if err := s.DBForCtx(ctx).Model(&db.WorkflowRun{}).Where("workflow_id = ?", workflow.ID).Count(&count).Error; err != nil {
			return err
		}
		actorID := actor.ID
		run := db.WorkflowRun{
			RepositoryID: repo.ID, WorkflowID: workflow.ID, ActorID: &actorID,
			IntegrationID: githubActionsIntegrationIDPtr(), MergeGroupID: &group.ID,
			Name: ExtractWorkflowName(content, path), WorkflowPath: path,
			HeadBranch: group.HeadRef, HeadSHA: group.HeadSHA,
			Status: db.RunQueued, Conclusion: "", Event: "merge_group", RunNumber: int(count) + 1, RunAttempt: 1,
		}
		created, err := s.createWorkflowRunWithSnapshot(ctx, &run, content, true)
		if err != nil {
			return err
		}
		if !created {
			if err := s.DBForCtx(ctx).Where("merge_group_id = ? AND workflow_path = ?", group.ID, path).First(&run).Error; err != nil {
				return err
			}
		}
		if err := s.claimAndStartWorkflowRun(ctx, run.ID); err != nil {
			slog.WarnContext(ctx, "merge_group workflow execution kickoff failed", "workflow_run_id", run.ID, "error", err)
		}
	}
	return nil
}

func isWorkflowDefinitionPath(path string) bool {
	return strings.HasPrefix(path, ".github/workflows/") &&
		(strings.HasSuffix(path, ".yml") || strings.HasSuffix(path, ".yaml"))
}

// workflowForEventDefinition keeps the public Workflow ID relationship while
// making the event tree authoritative for discovery. A pre-existing disabled
// row still suppresses execution; a path present only in the event commit gets
// a catalog identity without requiring default-branch synchronization first.
func (s *Service) workflowForEventDefinition(ctx context.Context, repositoryID uint, path string, content []byte) (db.Workflow, bool, error) {
	var workflow db.Workflow
	err := s.DBForCtx(ctx).Where("repository_id = ? AND path = ?", repositoryID, path).First(&workflow).Error
	if err == nil {
		return workflow, workflow.State == db.WorkflowActive, nil
	}
	if !errors.Is(err, gorm.ErrRecordNotFound) {
		return db.Workflow{}, false, err
	}
	workflow = db.Workflow{
		RepositoryID: repositoryID,
		Name:         ExtractWorkflowName(content, path),
		Path:         path,
		State:        db.WorkflowActive,
	}
	result := s.DBForCtx(ctx).Clauses(clause.OnConflict{DoNothing: true}).Create(&workflow)
	if result.Error != nil {
		return db.Workflow{}, false, result.Error
	}
	if result.RowsAffected == 0 {
		if err := s.DBForCtx(ctx).Where("repository_id = ? AND path = ?", repositoryID, path).First(&workflow).Error; err != nil {
			return db.Workflow{}, false, err
		}
	}
	return workflow, workflow.State == db.WorkflowActive, nil
}

type mergeGroupCheckState int

const (
	mergeGroupCheckPending mergeGroupCheckState = iota
	mergeGroupCheckPassed
	mergeGroupCheckFailed
)

func (s *Service) ReevaluateMergeQueueForSHA(ctx context.Context, repoID uint, sha string) {
	var group db.MergeGroup
	if err := s.DBForCtx(ctx).Joins("JOIN merge_queues ON merge_queues.id = merge_groups.queue_id").
		Where("merge_queues.repository_id = ? AND merge_groups.head_sha = ? AND merge_groups.status = ?", repoID, sha, mergeGroupChecksRequested).
		First(&group).Error; err != nil {
		return
	}
	state, err := s.mergeGroupCheckResult(ctx, group)
	if err != nil || state == mergeGroupCheckPending {
		return
	}
	if state == mergeGroupCheckFailed {
		if err := s.failMergeGroup(ctx, group, "required status checks failed"); err != nil {
			slog.WarnContext(ctx, "merge queue failed group ejection failed", "group_id", group.ID, "error", err)
		}
		return
	}
	var entry db.MergeQueueEntry
	err = s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		result := tx.Model(&db.MergeGroup{}).Where("id = ? AND status = ?", group.ID, mergeGroupChecksRequested).
			Update("status", mergeGroupPassed)
		if result.Error != nil || result.RowsAffected == 0 {
			return result.Error
		}
		if err := tx.Preload("EnqueuedBy").First(&entry, group.HeadEntryID).Error; err != nil {
			return err
		}
		return tx.Model(&db.MergeQueueEntry{}).Where("id = ? AND merge_group_id = ?", entry.ID, group.ID).
			Update("state", mergeQueueEntryMergeable).Error
	})
	if err != nil {
		return
	}
	if err := s.tryMergeQueue(ctx, group.QueueID); err != nil {
		slog.WarnContext(ctx, "merge queue ready group merge failed", "group_id", group.ID, "error", err)
	}
	if entry.EnqueuedBy.ID != 0 {
		if _, err := s.BuildMergeGroupForQueue(ctx, group.QueueID, entry.EnqueuedBy); err != nil {
			slog.WarnContext(ctx, "merge queue build slot refill failed", "queue_id", group.QueueID, "error", err)
		}
	}
}

func (s *Service) mergeGroupCheckResult(ctx context.Context, group db.MergeGroup) (mergeGroupCheckState, error) {
	var queue db.MergeQueue
	if err := s.DBForCtx(ctx).First(&queue, group.QueueID).Error; err != nil {
		return mergeGroupCheckPending, err
	}
	checks, err := s.effectiveRequiredStatusChecks(ctx, queue.RepositoryID, queue.BaseRef)
	if err != nil {
		return mergeGroupCheckPending, err
	}
	if len(checks) == 0 {
		return mergeGroupCheckPassed, nil
	}
	states, err := s.latestStatusChecksForPR(ctx, db.PullRequest{RepositoryID: queue.RepositoryID, HeadSHA: group.HeadSHA})
	if err != nil {
		return mergeGroupCheckPending, err
	}
	for _, check := range checks {
		state, ok := states[requiredStatusCheckKey(check.Context, check.IntegrationID)]
		if !ok || !state.Completed {
			return mergeGroupCheckPending, nil
		}
		if !state.Passed {
			return mergeGroupCheckFailed, nil
		}
	}
	return mergeGroupCheckPassed, nil
}

func (s *Service) scheduleMergeGroupTimeout(parent context.Context, group db.MergeGroup) {
	timerKey := s.workKey(parent, group.ID)
	if _, loaded := s.mergeGroupTimeouts.LoadOrStore(timerKey, struct{}{}); loaded {
		return
	}
	var queue db.MergeQueue
	if err := s.DBForCtx(parent).First(&queue, group.QueueID).Error; err != nil {
		s.mergeGroupTimeouts.Delete(timerKey)
		return
	}
	config, enabled, err := s.MergeQueueConfigurationForBranch(parent, queue.RepositoryID, queue.BaseRef)
	if err != nil || !enabled {
		s.mergeGroupTimeouts.Delete(timerKey)
		return
	}
	delay := time.Until(group.CreatedAt.Add(time.Duration(config.CheckResponseTimeoutMinutes) * time.Minute))
	if delay < 0 {
		delay = 0
	}
	ctx, err := s.DetachContext(parent)
	if err != nil {
		s.mergeGroupTimeouts.Delete(timerKey)
		slog.WarnContext(parent, "merge group timeout not scheduled", "group_id", group.ID, "error", err)
		return
	}
	s.Wg.Add(1)
	go func() {
		defer s.Wg.Done()
		defer s.mergeGroupTimeouts.Delete(timerKey)
		timer := time.NewTimer(delay)
		defer timer.Stop()
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
		}
		var current db.MergeGroup
		if err := s.DBForCtx(ctx).First(&current, group.ID).Error; err != nil || current.Status != mergeGroupChecksRequested {
			return
		}
		state, err := s.mergeGroupCheckResult(ctx, current)
		if err == nil && state == mergeGroupCheckPassed {
			s.ReevaluateMergeQueueForSHA(ctx, queue.RepositoryID, current.HeadSHA)
			return
		}
		if err := s.failMergeGroup(ctx, current, "required status checks timed out"); err != nil {
			slog.WarnContext(ctx, "merge queue timeout ejection failed", "group_id", current.ID, "error", err)
		}
	}()
}

type mergeQueueCandidate struct {
	Group   db.MergeGroup
	Entries []db.MergeQueueEntry
	Config  MergeQueueConfiguration
}

type mergeQueueRecoverableError struct{ err error }

func (e *mergeQueueRecoverableError) Error() string { return e.err.Error() }
func (e *mergeQueueRecoverableError) Unwrap() error { return e.err }

func (s *Service) tryMergeQueue(ctx context.Context, queueID uint) error {
	candidate, waitUntil, err := s.lockMergeQueueCandidate(ctx, queueID)
	if err != nil {
		return err
	}
	if waitUntil != nil {
		s.scheduleMergeQueueMinimumWait(ctx, queueID, *waitUntil)
		return nil
	}
	if candidate == nil {
		return nil
	}
	if err := s.mergeQueueCandidate(ctx, *candidate); err != nil {
		var recoverable *mergeQueueRecoverableError
		if !errors.As(err, &recoverable) {
			_ = s.failMergeGroup(ctx, candidate.Group, err.Error())
		}
		return err
	}
	return nil
}

func (s *Service) lockMergeQueueCandidate(ctx context.Context, queueID uint) (*mergeQueueCandidate, *time.Time, error) {
	var candidate *mergeQueueCandidate
	var waitUntil *time.Time
	err := s.Transaction(ctx, func(txCtx context.Context, tx *gorm.DB) error {
		var queue db.MergeQueue
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&queue, queueID).Error; err != nil {
			return wrapErr(err)
		}
		config, enabled, err := s.MergeQueueConfigurationForBranch(txCtx, queue.RepositoryID, queue.BaseRef)
		if err != nil || !enabled {
			return err
		}
		var locked int64
		if err := tx.Model(&db.MergeQueueEntry{}).Where("queue_id = ? AND state = ?", queue.ID, mergeQueueEntryLocked).Count(&locked).Error; err != nil || locked > 0 {
			return err
		}
		var entries []db.MergeQueueEntry
		if err := tx.Where("queue_id = ? AND state IN ?", queue.ID, activeMergeQueueEntryStates).
			Order("position ASC").Preload("PullRequest.Repository.Owner").Preload("PullRequest.HeadRepository.Owner").
			Preload("PullRequest.Author").Preload("EnqueuedBy").Find(&entries).Error; err != nil {
			return err
		}
		if len(entries) == 0 {
			return nil
		}
		maxMerge := config.MaxEntriesToMerge
		if maxMerge > len(entries) {
			maxMerge = len(entries)
		}
		eligible := 0
		if config.GroupingStrategy == "ALLGREEN" {
			for eligible < maxMerge && entries[eligible].State == mergeQueueEntryMergeable {
				eligible++
			}
		} else {
			var passed []db.MergeGroup
			entryIDs := make([]uint, maxMerge)
			for i := 0; i < maxMerge; i++ {
				entryIDs[i] = entries[i].ID
			}
			if err := tx.Where("head_entry_id IN ? AND status = ?", entryIDs, mergeGroupPassed).Find(&passed).Error; err != nil {
				return err
			}
			passedByEntry := make(map[uint]bool, len(passed))
			for _, group := range passed {
				passedByEntry[group.HeadEntryID] = true
			}
			for i := 0; i < maxMerge; i++ {
				if passedByEntry[entries[i].ID] {
					eligible = i + 1
				}
			}
		}
		if eligible == 0 {
			return nil
		}
		selected := entries[:eligible]
		if eligible < config.MinEntriesToMerge {
			deadline := entries[0].CreatedAt.Add(time.Duration(config.MinEntriesToMergeWaitMinutes) * time.Minute)
			if time.Now().UTC().Before(deadline) {
				waitUntil = &deadline
				return nil
			}
		}
		var group db.MergeGroup
		if err := tx.Where("head_entry_id = ? AND status = ?", selected[len(selected)-1].ID, mergeGroupPassed).
			Order("created_at DESC").First(&group).Error; err != nil {
			return wrapErr(err)
		}
		ids := make([]uint, len(selected))
		for i := range selected {
			ids[i] = selected[i].ID
		}
		result := tx.Model(&db.MergeQueueEntry{}).Where("id IN ? AND state IN ?", ids, activeMergeQueueEntryStates).
			Update("state", mergeQueueEntryLocked)
		if result.Error != nil {
			return result.Error
		}
		if result.RowsAffected != int64(len(ids)) {
			return fmt.Errorf("%w: merge queue changed while locking a merge group", ErrConflict)
		}
		if err := tx.Model(&db.MergeGroup{}).Where("id = ? AND status = ?", group.ID, mergeGroupPassed).
			Update("status", mergeGroupMerging).Error; err != nil {
			return err
		}
		candidate = &mergeQueueCandidate{Group: group, Entries: selected, Config: config}
		return nil
	})
	return candidate, waitUntil, err
}

func (s *Service) scheduleMergeQueueMinimumWait(parent context.Context, queueID uint, deadline time.Time) {
	timerKey := s.workKey(parent, queueID)
	if _, loaded := s.mergeQueueMinimumWaits.LoadOrStore(timerKey, struct{}{}); loaded {
		return
	}
	delay := time.Until(deadline)
	if delay < 0 {
		delay = 0
	}
	ctx, err := s.DetachContext(parent)
	if err != nil {
		s.mergeQueueMinimumWaits.Delete(timerKey)
		slog.WarnContext(parent, "merge queue minimum wait not scheduled", "queue_id", queueID, "error", err)
		return
	}
	s.Wg.Add(1)
	go func() {
		defer s.Wg.Done()
		defer s.mergeQueueMinimumWaits.Delete(timerKey)
		timer := time.NewTimer(delay)
		defer timer.Stop()
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
			if err := s.tryMergeQueue(ctx, queueID); err != nil {
				slog.WarnContext(ctx, "merge queue minimum-size wait processing failed", "queue_id", queueID, "error", err)
			}
		}
	}()
}

func (s *Service) mergeQueueCandidate(ctx context.Context, candidate mergeQueueCandidate) error {
	group := candidate.Group
	entries := candidate.Entries
	if len(entries) == 0 {
		return nil
	}
	var queue db.MergeQueue
	if err := s.DBForCtx(ctx).Preload("Repository").First(&queue, group.QueueID).Error; err != nil {
		return err
	}
	actor := entries[0].EnqueuedBy
	for _, entry := range entries {
		if entry.PullRequest.HeadSHA != entry.HeadSHA {
			return fmt.Errorf("pull request head changed while queued")
		}
		policyPR := entry.PullRequest
		policyPR.Repository = entries[0].PullRequest.Repository
		policyPR.BaseRef = queue.BaseRef
		policyPR.HeadSHA = group.HeadSHA
		policyErr := s.enforceMergePolicy(ctx, actor, &policyPR)
		if stack, stackErr := s.GetPullRequestStackForPR(ctx, entry.PullRequestID); stackErr == nil && stack != nil {
			policyErr = s.enforceStackMergePolicy(ctx, actor, &policyPR)
		}
		if policyErr != nil {
			return policyErr
		}
	}
	method := strings.ToLower(candidate.Config.MergeMethod)
	resultSHA := group.HeadSHA
	var memberSHAs []string
	gitStore, err := s.GitForCtx(ctx)
	if err != nil {
		return err
	}
	members := make([]gitstore.StackMergeMember, len(entries))
	for i, entry := range entries {
		members[i] = gitstore.StackMergeMember{
			Number: entry.PullRequest.Number, BaseBranch: entry.PullRequest.BaseRef,
			HeadBranch: entry.PullRequest.HeadRef, Message: fmt.Sprintf("Merge pull request #%d", entry.PullRequest.Number),
			ExpectedHeadSHA: entry.HeadSHA,
		}
	}
	mergeResult, mergeErr := gitStore.MergeStackAtomicallyDetailed(ctx, gitstore.StackMergeOptions{
		FullName: queue.Repository.FullName, BaseBranch: queue.BaseRef,
		Members: members, MergeMethod: method, Committer: actor.Login, Email: actor.Email,
	})
	if mergeErr != nil {
		return mergeErr
	}
	resultSHA = mergeResult.FinalSHA
	memberSHAs = mergeResult.MemberSHAs
	if err := s.finalizeMergeQueueCandidate(ctx, candidate, resultSHA, memberSHAs); err != nil {
		return &mergeQueueRecoverableError{err: err}
	}
	return nil
}

func (s *Service) finalizeMergeQueueCandidate(ctx context.Context, candidate mergeQueueCandidate, resultSHA string, memberSHAs []string) error {
	group := candidate.Group
	entries := candidate.Entries
	if len(entries) == 0 {
		return nil
	}
	var queue db.MergeQueue
	if err := s.DBForCtx(ctx).Preload("Repository").First(&queue, group.QueueID).Error; err != nil {
		return err
	}
	actor := entries[0].EnqueuedBy
	if len(memberSHAs) == 0 {
		memberSHAs = make([]string, len(entries))
		for i := range memberSHAs {
			memberSHAs[i] = resultSHA
		}
	}
	if len(memberSHAs) != len(entries) {
		return fmt.Errorf("cannot finalize merge queue group without exact member SHAs")
	}
	for i := range memberSHAs {
		if strings.TrimSpace(memberSHAs[i]) == "" {
			return fmt.Errorf("cannot finalize merge queue group with an empty member SHA")
		}
	}
	prs := make([]db.PullRequest, len(entries))
	entryIDs := make([]uint, len(entries))
	entryMergeSHAs := make([]string, len(entries))
	for i := range entries {
		prs[i] = entries[i].PullRequest
		entryIDs[i] = entries[i].ID
		entryMergeSHAs[i] = memberSHAs[i]
	}
	if err := s.markStackPrefixMerged(ctx, prs, actor.Login, memberSHAs); err != nil {
		return err
	}
	now := time.Now().UTC()
	if err := s.DBForCtx(ctx).Model(&db.MergeGroup{}).Where("id = ? AND status IN ?", group.ID,
		[]string{mergeGroupMerging, mergeGroupCascading}).Updates(map[string]any{
		"status": mergeGroupCascading, "updated_at": now,
	}).Error; err != nil {
		return err
	}
	group.Status = mergeGroupCascading
	seenStacks := make(map[uint]struct{})
	for i := len(prs) - 1; i >= 0; i-- {
		stack, stackErr := s.GetPullRequestStackForPR(ctx, prs[i].ID)
		if stackErr != nil || stack == nil {
			continue
		}
		if _, done := seenStacks[stack.ID]; done {
			continue
		}
		seenStacks[stack.ID] = struct{}{}
		if cascadeErr := s.cascadeStackAfterMerge(ctx, prs[i].ID, resultSHA, actor); cascadeErr != nil {
			return fmt.Errorf("cascade Stack %d after merge queue merge: %w", stack.ID, cascadeErr)
		}
	}
	if err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		now := time.Now().UTC()
		if err := tx.Model(&db.MergeQueueEntry{}).Where("id IN ?", entryIDs).
			Updates(map[string]any{"state": mergeQueueEntryMerged, "updated_at": now}).Error; err != nil {
			return err
		}
		if err := terminalizeMergeGroupsForEntriesTx(tx, queue.ID, entryIDs,
			[]string{mergeGroupPublishing, mergeGroupChecksRequested, mergeGroupPassed, mergeGroupMerging, mergeGroupCascading},
			mergeGroupMerged, mergeGroupDestroyedMerged, actor.ID); err != nil {
			return err
		}
		var remaining []db.MergeQueueEntry
		if err := tx.Where("queue_id = ? AND state IN ?", queue.ID, []string{mergeQueueEntryAwaitingChecks, mergeQueueEntryMergeable}).
			Order("position ASC").Find(&remaining).Error; err != nil {
			return err
		}
		remainingIDs := make([]uint, len(remaining))
		for i := range remaining {
			remainingIDs[i] = remaining[i].ID
		}
		if len(remainingIDs) > 0 {
			if err := terminalizeMergeGroupsForEntriesTx(tx, queue.ID, remainingIDs,
				[]string{mergeGroupPublishing, mergeGroupChecksRequested, mergeGroupPassed},
				mergeGroupStale, mergeGroupDestroyedInvalidated, actor.ID); err != nil {
				return err
			}
			if err := tx.Model(&db.MergeQueueEntry{}).Where("id IN ?", remainingIDs).
				Updates(map[string]any{"state": mergeQueueEntryQueued, "merge_group_id": nil, "updated_at": now}).Error; err != nil {
				return err
			}
		}
		for i, entryID := range entryIDs {
			if err := tx.Model(&db.AsyncMergeRequest{}).Where("merge_queue_entry_id = ? AND status = ?", entryID, AsyncMergeEnqueued).
				Updates(map[string]any{"status": AsyncMergeMerged, "result_sha": entryMergeSHAs[i], "updated_at": now, "expires_at": now.Add(asyncMergeRetention)}).Error; err != nil {
				return err
			}
		}
		return compactMergeQueuePositionsTx(tx, queue.ID)
	}); err != nil {
		return err
	}
	prIDs := make([]uint, len(prs))
	for i := range prs {
		prIDs[i] = prs[i].ID
	}
	s.notifyPullRequestsMerged(ctx, prIDs)
	if err := s.cleanupTerminalMergeGroupRefs(ctx, queue.ID); err != nil {
		return err
	}
	s.startMergeQueueProcessor(ctx, queue.ID, actor)
	return nil
}

func (s *Service) recoverCascadingMergeGroup(ctx context.Context, group db.MergeGroup) error {
	var queue db.MergeQueue
	if err := s.DBForCtx(ctx).Preload("Repository").First(&queue, group.QueueID).Error; err != nil {
		return err
	}
	var entries []db.MergeQueueEntry
	if err := s.DBForCtx(ctx).Where("queue_id = ? AND state = ?", group.QueueID, mergeQueueEntryLocked).
		Order("position ASC").Preload("PullRequest.Repository.Owner").Preload("PullRequest.HeadRepository.Owner").
		Preload("PullRequest.Author").Preload("EnqueuedBy").Find(&entries).Error; err != nil {
		return err
	}
	if len(entries) == 0 {
		return fmt.Errorf("cascading merge group %d has no locked entries", group.ID)
	}
	config, enabled, err := s.MergeQueueConfigurationForBranch(ctx, queue.RepositoryID, queue.BaseRef)
	if err != nil {
		return err
	}
	if !enabled {
		return fmt.Errorf("merge queue is no longer configured for %s", queue.BaseRef)
	}
	memberSHAs := make([]string, len(entries))
	for i := range entries {
		memberSHAs[i] = entries[i].PullRequest.MergeCommitSHA
		if strings.TrimSpace(memberSHAs[i]) == "" {
			return fmt.Errorf("cascading merge group %d is missing member SHA at position %d", group.ID, i+1)
		}
	}
	resultSHA := memberSHAs[len(memberSHAs)-1]
	gitStore, err := s.GitForCtx(ctx)
	if err != nil {
		return err
	}
	currentBaseSHA, headErr := gitStore.HeadSHA(ctx, queue.Repository.FullName, queue.BaseRef)
	if headErr != nil {
		return headErr
	}
	ancestor, ancestryErr := gitStore.IsAncestor(ctx, queue.Repository.FullName, resultSHA, currentBaseSHA)
	if ancestryErr != nil {
		return ancestryErr
	}
	if !ancestor {
		return fmt.Errorf("cannot prove cascading merge group %d result is on the current base", group.ID)
	}
	return s.finalizeMergeQueueCandidate(ctx, mergeQueueCandidate{Group: group, Entries: entries, Config: config}, resultSHA, memberSHAs)
}

func (s *Service) failMergeGroup(ctx context.Context, group db.MergeGroup, reason string) error {
	var entry db.MergeQueueEntry
	if err := s.DBForCtx(ctx).Preload("EnqueuedBy").First(&entry, group.HeadEntryID).Error; err != nil {
		return err
	}
	var ejectedPullRequestIDs []uint
	err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		if err := terminalizeMergeGroupTx(tx, group,
			[]string{mergeGroupPublishing, mergeGroupChecksRequested, mergeGroupPassed, mergeGroupMerging, mergeGroupCascading},
			mergeGroupFailed, mergeGroupDestroyedDequeued, entry.EnqueuedByID); err != nil {
			return err
		}
		var err error
		ejectedPullRequestIDs, err = s.ejectMergeQueueEntryTx(tx, entry, reason, entry.EnqueuedByID)
		return err
	})
	if err == nil {
		s.notifyPullRequestQueueEvent(ctx, ejectedPullRequestIDs, "dequeued")
		if cleanupErr := s.cleanupTerminalMergeGroupRefs(ctx, group.QueueID); cleanupErr != nil {
			slog.WarnContext(ctx, "failed merge group ref cleanup failed", "group_id", group.ID, "error", cleanupErr)
		}
		s.startMergeQueueProcessor(ctx, group.QueueID, entry.EnqueuedBy)
	}
	return err
}

// cleanupTerminalMergeGroupRefs removes GitHub-shaped gh-readonly-queue refs
// after their durable group reaches a terminal state. It is intentionally
// idempotent so startup recovery can retry cleanup after a crash.
func (s *Service) cleanupTerminalMergeGroupRefs(ctx context.Context, queueID uint) error {
	var groups []db.MergeGroup
	if err := s.DBForCtx(ctx).Where("queue_id = ? AND status IN ?", queueID,
		[]string{mergeGroupMerged, mergeGroupFailed, mergeGroupStale}).Find(&groups).Error; err != nil {
		return err
	}
	for _, group := range groups {
		reason := group.TerminalReason
		if reason == "" {
			switch group.Status {
			case mergeGroupMerged:
				reason = mergeGroupDestroyedMerged
			case mergeGroupFailed:
				reason = mergeGroupDestroyedDequeued
			default:
				reason = mergeGroupDestroyedInvalidated
			}
			if err := s.DBForCtx(ctx).Model(&db.MergeGroup{}).Where("id = ? AND terminal_reason = ''", group.ID).
				Update("terminal_reason", reason).Error; err != nil {
				return err
			}
			group.TerminalReason = reason
		}
		if err := s.ensureMergeGroupLifecycleEvent(ctx, group, mergeGroupLifecycleDestroyed, reason, group.ActorID); err != nil {
			return err
		}
		if err := s.processMergeGroupLifecycleEvents(ctx, group.ID); err != nil {
			return err
		}
	}
	return nil
}

func (s *Service) ejectMergeQueueEntryTx(tx *gorm.DB, entry db.MergeQueueEntry, reason string, actorID uint) ([]uint, error) {
	query := tx.Model(&db.MergeQueueEntry{}).Where("id = ?", entry.ID)
	if entry.StackBatchID != "" {
		query = tx.Model(&db.MergeQueueEntry{}).Where("stack_batch_id = ? AND stack_position >= ?", entry.StackBatchID, entry.StackPosition)
	}
	var ejected []db.MergeQueueEntry
	if err := query.Order("position ASC").Find(&ejected).Error; err != nil {
		return nil, err
	}
	if len(ejected) == 0 {
		return nil, nil
	}
	ids := make([]uint, len(ejected))
	pullRequestIDs := make([]uint, len(ejected))
	minPosition := ejected[0].Position
	for i := range ejected {
		ids[i] = ejected[i].ID
		pullRequestIDs[i] = ejected[i].PullRequestID
		if ejected[i].Position < minPosition {
			minPosition = ejected[i].Position
		}
		if err := tx.Create(&db.PullRequestMergeQueueEvent{
			PullRequestID: ejected[i].PullRequestID, QueueID: ejected[i].QueueID,
			QueueEntryID: ejected[i].ID, ActorID: actorID, EventType: "removed",
			BeforeSHA: ejected[i].BaseSHA, Reason: reason,
		}).Error; err != nil {
			return nil, err
		}
	}
	now := time.Now().UTC()
	if err := terminalizeMergeGroupsForEntriesTx(tx, entry.QueueID, []uint{entry.ID},
		[]string{mergeGroupPublishing, mergeGroupChecksRequested, mergeGroupPassed, mergeGroupMerging, mergeGroupCascading},
		mergeGroupFailed, mergeGroupDestroyedDequeued, actorID); err != nil {
		return nil, err
	}
	invalidatedEjectedIDs := make([]uint, 0, len(ejected)-1)
	for i := range ejected {
		if ejected[i].ID != entry.ID {
			invalidatedEjectedIDs = append(invalidatedEjectedIDs, ejected[i].ID)
		}
	}
	if err := terminalizeMergeGroupsForEntriesTx(tx, entry.QueueID, invalidatedEjectedIDs,
		[]string{mergeGroupPublishing, mergeGroupChecksRequested, mergeGroupPassed, mergeGroupMerging, mergeGroupCascading},
		mergeGroupStale, mergeGroupDestroyedInvalidated, actorID); err != nil {
		return nil, err
	}
	if err := tx.Model(&db.MergeQueueEntry{}).Where("id IN ?", ids).
		Updates(map[string]any{"state": mergeQueueEntryUnmergeable, "updated_at": now}).Error; err != nil {
		return nil, err
	}
	var invalidated []db.MergeQueueEntry
	if err := tx.Where("queue_id = ? AND position >= ? AND state IN ?", entry.QueueID, minPosition, activeMergeQueueEntryStates).
		Find(&invalidated).Error; err != nil {
		return nil, err
	}
	invalidatedIDs := make([]uint, len(invalidated))
	for i := range invalidated {
		invalidatedIDs[i] = invalidated[i].ID
	}
	if len(invalidatedIDs) > 0 {
		if err := terminalizeMergeGroupsForEntriesTx(tx, entry.QueueID, invalidatedIDs,
			[]string{mergeGroupPublishing, mergeGroupChecksRequested, mergeGroupPassed, mergeGroupMerging, mergeGroupCascading},
			mergeGroupStale, mergeGroupDestroyedInvalidated, actorID); err != nil {
			return nil, err
		}
		if err := tx.Model(&db.MergeQueueEntry{}).Where("id IN ?", invalidatedIDs).
			Updates(map[string]any{"state": mergeQueueEntryQueued, "merge_group_id": nil, "updated_at": now}).Error; err != nil {
			return nil, err
		}
	}
	if err := tx.Model(&db.AsyncMergeRequest{}).Where("merge_queue_entry_id IN ? AND status = ?", ids, AsyncMergeEnqueued).
		Updates(map[string]any{"status": AsyncMergeFailed, "failure_message": reason, "updated_at": now, "expires_at": now.Add(asyncMergeRetention)}).Error; err != nil {
		return nil, err
	}
	if err := compactMergeQueuePositionsTx(tx, entry.QueueID); err != nil {
		return nil, err
	}
	return pullRequestIDs, nil
}

func compactMergeQueuePositionsTx(tx *gorm.DB, queueID uint) error {
	var entries []db.MergeQueueEntry
	if err := tx.Where("queue_id = ? AND state IN ?", queueID, activeMergeQueueEntryStates).Order("position ASC").Find(&entries).Error; err != nil {
		return err
	}
	for i := range entries {
		position := i + 1
		if entries[i].Position == position {
			continue
		}
		if err := tx.Model(&db.MergeQueueEntry{}).Where("id = ?", entries[i].ID).Update("position", position).Error; err != nil {
			return err
		}
	}
	return nil
}