package service

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

	"github.com/pingcap/agent-git-service/internal/db"
	"gorm.io/gorm"
)

const (
	defaultPullRequestMergeRecoveryInterval = 10 * time.Second
	defaultPullRequestMergeRecoveryGrace    = time.Minute
	maxPullRequestMergeRecoveryBackoff      = time.Minute
)

// StartPullRequestMergeRecovery continuously resumes durable asynchronous
// merge, Merge Queue, lifecycle-outbox, ref-cleanup, and workflow-run work.
// The first pass handles restart recovery immediately; later passes apply a
// grace period and exponential backoff so runtime transient failures recover
// without racing fresh in-process Git operations or hot-looping dependencies.
func (s *Service) StartPullRequestMergeRecovery(ctx context.Context) error {
	workerCtx, err := s.DetachContext(ctx)
	if err != nil {
		return err
	}
	key, err := s.tenantStateKey(workerCtx, "pull-request-merge-recovery")
	if err != nil {
		return err
	}
	if _, loaded := s.pullRequestMergeRecoveryWorkers.LoadOrStore(key, struct{}{}); loaded {
		return nil
	}
	s.Wg.Add(1)
	go func() {
		defer s.Wg.Done()
		interval := s.pullRequestMergeRecoveryIntervalValue()
		maxBackoff := maxPullRequestMergeRecoveryBackoff
		if interval > maxBackoff {
			maxBackoff = interval
		}
		backoff := interval
		firstPass := true
		for {
			var staleBefore *time.Time
			if !firstPass {
				cutoff := time.Now().UTC().Add(-s.pullRequestMergeRecoveryGracePeriodValue())
				staleBefore = &cutoff
			}
			err := s.recoverPullRequestMergeWork(workerCtx, staleBefore)
			delay := interval
			if err != nil {
				slog.WarnContext(workerCtx, "pull request merge recovery failed", "error", err, "retry_in", backoff)
				delay = backoff
				backoff *= 2
				if backoff > maxBackoff {
					backoff = maxBackoff
				}
			} else {
				backoff = interval
			}
			firstPass = false
			timer := time.NewTimer(delay)
			select {
			case <-workerCtx.Done():
				timer.Stop()
				return
			case <-timer.C:
			}
		}
	}()
	return nil
}

func (s *Service) pullRequestMergeRecoveryIntervalValue() time.Duration {
	if s.PullRequestMergeRecoveryInterval > 0 {
		return s.PullRequestMergeRecoveryInterval
	}
	return defaultPullRequestMergeRecoveryInterval
}

func (s *Service) pullRequestMergeRecoveryGracePeriodValue() time.Duration {
	if s.PullRequestMergeRecoveryGracePeriod > 0 {
		return s.PullRequestMergeRecoveryGracePeriod
	}
	return defaultPullRequestMergeRecoveryGrace
}

// RecoverPullRequestMergeWork reloads the durable work states which normally
// have an in-process worker or timer attached to them.
func (s *Service) RecoverPullRequestMergeWork(ctx context.Context) error {
	return s.recoverPullRequestMergeWork(ctx, nil)
}

func (s *Service) recoverPullRequestMergeWork(ctx context.Context, staleBefore *time.Time) error {
	if err := s.recoverWorkflowRuns(ctx); err != nil {
		return err
	}
	now := time.Now().UTC()
	var requests []db.AsyncMergeRequest
	requestQuery := s.DBForCtx(ctx).Where("status IN ?", recoverableAsyncMergeStatuses)
	if staleBefore != nil {
		requestQuery = requestQuery.Where("updated_at <= ?", *staleBefore)
	}
	if err := requestQuery.
		Preload("Actor").Find(&requests).Error; err != nil {
		return fmt.Errorf("recover active async merges: %w", err)
	}
	for i := range requests {
		marked, err := markAsyncMergeRecoveryAttempt(s.DBForCtx(ctx), requests[i].ID, now)
		if err != nil {
			return fmt.Errorf("mark active async merge %d recovery attempt: %w", requests[i].ID, err)
		}
		if !marked {
			continue
		}
		requests[i].UpdatedAt = now
		requests[i].ExpiresAt = now.Add(asyncMergeRetention)
		if requests[i].Status == AsyncMergePending {
			s.startAsyncMergeWorker(ctx, requests[i].ID, requests[i].Actor)
			continue
		}
		if err := s.recoverDirectStackMerge(ctx, requests[i]); err != nil {
			slog.WarnContext(ctx, "recover direct Stack merge failed", "request_id", requests[i].ID, "error", err)
		}
	}

	var groups []db.MergeGroup
	groupQuery := s.DBForCtx(ctx).Where("status IN ?", []string{mergeGroupPublishing, mergeGroupChecksRequested, mergeGroupPassed, mergeGroupMerging, mergeGroupCascading})
	if staleBefore != nil {
		groupQuery = groupQuery.Where("updated_at <= ?", *staleBefore)
	}
	if err := groupQuery.
		Find(&groups).Error; err != nil {
		return fmt.Errorf("recover merge groups: %w", err)
	}
	for i := range groups {
		var queue db.MergeQueue
		if err := s.DBForCtx(ctx).First(&queue, groups[i].QueueID).Error; err != nil {
			return fmt.Errorf("recover merge group %d queue: %w", groups[i].ID, err)
		}
		if groups[i].Status == mergeGroupPublishing {
			var actor db.User
			if groups[i].ActorID != 0 {
				_ = s.DBForCtx(ctx).First(&actor, groups[i].ActorID).Error
			}
			if err := s.publishMergeGroup(ctx, groups[i], actor); err != nil {
				slog.WarnContext(ctx, "recover merge group publication failed", "group_id", groups[i].ID, "error", err)
			}
			continue
		}
		if groups[i].Status == mergeGroupChecksRequested {
			if err := s.dispatchMergeGroup(ctx, groups[i], db.User{}); err != nil {
				slog.WarnContext(ctx, "recover merge group checks dispatch failed", "group_id", groups[i].ID, "error", err)
			}
			continue
		}
		if groups[i].Status == mergeGroupCascading {
			if err := s.recoverCascadingMergeGroup(ctx, groups[i]); err != nil {
				slog.WarnContext(ctx, "recover merge group cascade failed", "group_id", groups[i].ID, "error", err)
			}
			continue
		}
		if groups[i].Status == mergeGroupMerging {
			if err := s.recoverMergingMergeGroup(ctx, groups[i]); err != nil {
				slog.WarnContext(ctx, "recover merging merge group failed", "group_id", groups[i].ID, "error", err)
			}
			continue
		}
		if err := s.tryMergeQueue(ctx, groups[i].QueueID); err != nil {
			slog.WarnContext(ctx, "recover ready merge group failed", "group_id", groups[i].ID, "error", err)
		}
	}
	var terminalQueueIDs []uint
	terminalQuery := s.DBForCtx(ctx).Model(&db.MergeGroup{}).Distinct("queue_id").
		Where("status IN ? AND ref_deleted_at IS NULL", []string{mergeGroupMerged, mergeGroupFailed, mergeGroupStale})
	if staleBefore != nil {
		terminalQuery = terminalQuery.Where("updated_at <= ?", *staleBefore)
	}
	// Include legacy rows without ref_published_at: cleanup inspects Git and
	// deletes a real pre-marker ref, while leaving a never-published row
	// unmarked as required by the lifecycle invariant.
	if err := terminalQuery.
		Pluck("queue_id", &terminalQueueIDs).Error; err != nil {
		return fmt.Errorf("recover terminal merge groups: %w", err)
	}
	for _, queueID := range terminalQueueIDs {
		if err := s.cleanupTerminalMergeGroupRefs(ctx, queueID); err != nil {
			slog.WarnContext(ctx, "recover merge group ref cleanup failed", "queue_id", queueID, "error", err)
		}
	}

	var entries []db.MergeQueueEntry
	entryQuery := s.DBForCtx(ctx).Where("state IN ?", activeMergeQueueEntryStates)
	if staleBefore != nil {
		entryQuery = entryQuery.Where("updated_at <= ?", *staleBefore)
	}
	if err := entryQuery.
		Order("queue_id ASC, position ASC").Preload("EnqueuedBy").Find(&entries).Error; err != nil {
		return fmt.Errorf("recover merge queue entries: %w", err)
	}
	queueActors := make(map[uint]db.User)
	for i := range entries {
		if _, exists := queueActors[entries[i].QueueID]; !exists {
			queueActors[entries[i].QueueID] = entries[i].EnqueuedBy
		}
	}
	for queueID, actor := range queueActors {
		if err := s.cleanupTerminalMergeGroupRefs(ctx, queueID); err != nil {
			slog.WarnContext(ctx, "recover merge group ref cleanup failed", "queue_id", queueID, "error", err)
		}
		s.startMergeQueueProcessor(ctx, queueID, actor)
	}
	return nil
}

// markAsyncMergeRecoveryAttempt records real recovery activity. Periodic scans
// use UpdatedAt as their grace/backoff clock, while ExpiresAt independently
// controls how long the result remains publicly visible.
func markAsyncMergeRecoveryAttempt(query *gorm.DB, requestID uint, now time.Time) (bool, error) {
	result := query.Model(&db.AsyncMergeRequest{}).
		Where("id = ? AND status IN ?", requestID, recoverableAsyncMergeStatuses).
		Updates(map[string]any{"updated_at": now, "expires_at": now.Add(asyncMergeRetention)})
	if result.Error != nil {
		return false, result.Error
	}
	return result.RowsAffected == 1, nil
}

func (s *Service) recoverMergingMergeGroup(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 {
		if err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
			return terminalizeMergeGroupTx(tx, group, []string{mergeGroupMerging}, mergeGroupStale, mergeGroupDestroyedInvalidated, group.ActorID)
		}); err != nil {
			return err
		}
		return s.cleanupTerminalMergeGroupRefs(ctx, group.QueueID)
	}
	config, enabled, err := s.MergeQueueConfigurationForBranch(ctx, queue.RepositoryID, queue.BaseRef)
	if err != nil {
		return err
	}
	if !enabled {
		return s.staleRecoveredMergeGroup(ctx, group, entries)
	}
	candidate := mergeQueueCandidate{Group: group, Entries: entries, Config: config}
	gitStore, err := s.GitForCtx(ctx)
	if err != nil {
		return err
	}
	baseSHA, err := gitStore.HeadSHA(ctx, queue.Repository.FullName, queue.BaseRef)
	if err != nil {
		return err
	}
	if baseSHA == group.BaseSHA {
		return s.retryRecoveredMergeGroup(ctx, group, entries)
	}
	baseCommit, baseErr := gitStore.GetCommit(ctx, queue.Repository.FullName, baseSHA)
	groupCommit, groupErr := gitStore.GetCommit(ctx, queue.Repository.FullName, group.HeadSHA)
	if baseErr == nil && groupErr == nil && baseCommit.TreeSHA == groupCommit.TreeSHA {
		memberSHAs, mapErr := s.recoveredMergeGroupMemberSHAs(ctx, queue, group, entries, baseSHA)
		if mapErr != nil {
			return fmt.Errorf("recover merge group member SHAs: %w", mapErr)
		}
		return s.finalizeMergeQueueCandidate(ctx, candidate, baseSHA, memberSHAs)
	}
	return s.staleRecoveredMergeGroup(ctx, group, entries)
}

// recoveredMergeGroupMemberSHAs reconstructs the per-PR merge_commit_sha values
// after Git advanced but the durable queue transaction did not commit. Each
// speculative prefix group records the tree expected after its PR; matching
// those trees along the new base's first-parent chain works for merge, squash,
// and multi-commit rebase strategies without replaying the Git write.
func (s *Service) recoveredMergeGroupMemberSHAs(ctx context.Context, queue db.MergeQueue, group db.MergeGroup, entries []db.MergeQueueEntry, baseSHA string) ([]string, error) {
	gitStore, err := s.GitForCtx(ctx)
	if err != nil {
		return nil, err
	}
	expectedTrees := make([]string, len(entries))
	for i, entry := range entries {
		var prefix db.MergeGroup
		if entry.ID == group.HeadEntryID {
			prefix = group
		} else {
			if err := s.DBForCtx(ctx).Where("queue_id = ? AND head_entry_id = ? AND base_sha = ?", group.QueueID, entry.ID, group.BaseSHA).
				Order("created_at DESC").First(&prefix).Error; err != nil {
				return nil, err
			}
		}
		expected, err := gitStore.GetCommit(ctx, queue.Repository.FullName, prefix.HeadSHA)
		if err != nil {
			return nil, err
		}
		expectedTrees[i] = expected.TreeSHA
	}
	return s.recoverMemberSHAsFromExpectedTrees(ctx, queue.Repository.FullName, group.BaseSHA, baseSHA, expectedTrees)
}

func (s *Service) recoverMemberSHAsFromExpectedTrees(ctx context.Context, repositoryFullName, oldBaseSHA, newBaseSHA string, expectedTrees []string) ([]string, error) {
	gitStore, err := s.GitForCtx(ctx)
	if err != nil {
		return nil, err
	}
	const maxFirstParentCommits = 100000
	history := make([]string, 0, len(expectedTrees))
	current := newBaseSHA
	for current != oldBaseSHA {
		if len(history) == maxFirstParentCommits {
			return nil, fmt.Errorf("first-parent history exceeds %d commits", maxFirstParentCommits)
		}
		commit, err := gitStore.GetCommit(ctx, repositoryFullName, current)
		if err != nil {
			return nil, err
		}
		history = append(history, current)
		if len(commit.ParentSHAs) == 0 {
			return nil, fmt.Errorf("base %s is not descended from original base %s", newBaseSHA, oldBaseSHA)
		}
		current = commit.ParentSHAs[0]
	}
	for left, right := 0, len(history)-1; left < right; left, right = left+1, right-1 {
		history[left], history[right] = history[right], history[left]
	}
	memberSHAs := make([]string, len(expectedTrees))
	historyIndex := 0
	for i, expectedTree := range expectedTrees {
		for historyIndex < len(history) {
			actual, err := gitStore.GetCommit(ctx, repositoryFullName, history[historyIndex])
			if err != nil {
				return nil, err
			}
			historyIndex++
			if actual.TreeSHA == expectedTree {
				memberSHAs[i] = actual.SHA
				break
			}
		}
		if memberSHAs[i] == "" {
			return nil, fmt.Errorf("no first-parent commit matches expected member tree at position %d", i+1)
		}
	}
	return memberSHAs, nil
}

func (s *Service) retryRecoveredMergeGroup(ctx context.Context, group db.MergeGroup, entries []db.MergeQueueEntry) error {
	ids := mergeQueueEntryIDs(entries)
	if err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		if err := tx.Model(&db.MergeGroup{}).Where("id = ? AND status = ?", group.ID, mergeGroupMerging).
			Update("status", mergeGroupPassed).Error; err != nil {
			return err
		}
		return tx.Model(&db.MergeQueueEntry{}).Where("id IN ? AND state = ?", ids, mergeQueueEntryLocked).
			Update("state", mergeQueueEntryMergeable).Error
	}); err != nil {
		return err
	}
	return s.tryMergeQueue(ctx, group.QueueID)
}

func (s *Service) staleRecoveredMergeGroup(ctx context.Context, group db.MergeGroup, entries []db.MergeQueueEntry) error {
	ids := mergeQueueEntryIDs(entries)
	if err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		if err := terminalizeMergeGroupTx(tx, group, []string{mergeGroupMerging}, mergeGroupStale, mergeGroupDestroyedInvalidated, group.ActorID); err != nil {
			return err
		}
		return tx.Model(&db.MergeQueueEntry{}).Where("id IN ? AND state = ?", ids, mergeQueueEntryLocked).
			Updates(map[string]any{"state": mergeQueueEntryQueued, "merge_group_id": nil, "updated_at": time.Now().UTC()}).Error
	}); err != nil {
		return err
	}
	if err := s.cleanupTerminalMergeGroupRefs(ctx, group.QueueID); err != nil {
		return err
	}
	return nil
}

func mergeQueueEntryIDs(entries []db.MergeQueueEntry) []uint {
	ids := make([]uint, len(entries))
	for i := range entries {
		ids[i] = entries[i].ID
	}
	return ids
}