package service

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

	"github.com/google/uuid"
	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/gitstore"
	applog "github.com/pingcap/agent-git-service/internal/logging"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const workflowExecutionLeaseSafety = 30 * time.Second

// claimAndStartWorkflowRun atomically claims one durable queued/in-progress
// run. The lease prevents concurrent executors and makes process death
// recoverable without manufacturing another public WorkflowRun.
func (s *Service) claimAndStartWorkflowRun(parent context.Context, runID uint) error {
	ctx, err := s.DetachContext(parent)
	if err != nil {
		return err
	}
	policy, err := s.WorkflowPolicyForCtx(ctx)
	if err != nil {
		return err
	}

	now := time.Now().UTC()
	leaseID := uuid.NewString()
	leaseExpiresAt := now.Add(normalizeWorkflowPolicy(policy).Timeout + workflowExecutionLeaseSafety)
	result := s.DBForCtx(ctx).Model(&db.WorkflowRun{}).
		Where("id = ? AND status IN ? AND (execution_lease_expires_at IS NULL OR execution_lease_expires_at <= ?)",
			runID, []string{db.RunQueued, db.RunInProgress}, now).
		Updates(map[string]any{
			"status": db.RunInProgress, "execution_lease_id": leaseID,
			"execution_lease_expires_at": &leaseExpiresAt,
			"run_attempt":                gorm.Expr("CASE WHEN status = ? THEN run_attempt + 1 ELSE run_attempt END", db.RunInProgress),
		})
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected == 0 {
		return nil
	}

	var run db.WorkflowRun
	if err := s.DBForCtx(ctx).First(&run, runID).Error; err != nil {
		return err
	}
	var repo db.Repository
	if err := s.DBForCtx(ctx).Preload("Owner").First(&repo, run.RepositoryID).Error; err != nil {
		if !errors.Is(err, gorm.ErrRecordNotFound) {
			return fmt.Errorf("load workflow run repository: %w", err)
		}
		slog.ErrorContext(ctx, "workflow run repository unavailable", "workflow_run_id", run.ID, "error", err)
		s.completeRun(ctx, run.ID, leaseID, db.ConclusionFailure)
		return nil
	}
	snapshot, ready, err := s.ensureWorkflowRunSnapshot(ctx, &run, repo, leaseID)
	if err != nil {
		return err
	}
	if !ready {
		return nil
	}
	if run.RunAttempt > 1 {
		// The execution timeout ends before the lease, so an expired lease has
		// no live executor left. Replace the abandoned attempt's partial public
		// jobs/artifacts before rebuilding the same run on its next attempt.
		if err := s.DBForCtx(ctx).Where("run_id = ?", run.ID).Delete(&db.WorkflowRunJob{}).Error; err != nil {
			return fmt.Errorf("clear abandoned workflow jobs: %w", err)
		}
		if err := s.DBForCtx(ctx).Where("run_id = ?", run.ID).Delete(&db.Artifact{}).Error; err != nil {
			return fmt.Errorf("clear abandoned workflow artifacts: %w", err)
		}
	}
	applog.AddAttrs(ctx,
		slog.String("repo", repo.FullName),
		slog.Int("workflow_id", int(run.WorkflowID)),
		slog.Int("workflow_run_id", int(run.ID)),
	)
	s.Wg.Add(1)
	go func() {
		defer s.Wg.Done()
		s.executeWorkflow(ctx, run, repo, snapshot, leaseID)
	}()
	return nil
}

// ensureWorkflowRunSnapshot upgrades runs created before immutable workflow
// snapshots existed. It reads only the recorded HeadSHA: definitive missing
// history fails the run, while backend errors keep the lease retryable.
func (s *Service) ensureWorkflowRunSnapshot(ctx context.Context, run *db.WorkflowRun, repo db.Repository, leaseID string) (db.WorkflowRunSnapshot, bool, error) {
	var snapshot db.WorkflowRunSnapshot
	err := s.DBForCtx(ctx).First(&snapshot, "workflow_run_id = ?", run.ID).Error
	if err == nil {
		return snapshot, true, nil
	}
	if !errors.Is(err, gorm.ErrRecordNotFound) {
		return db.WorkflowRunSnapshot{}, false, fmt.Errorf("load workflow run snapshot: %w", err)
	}

	path := run.WorkflowPath
	if path == "" {
		var workflow db.Workflow
		if err := s.DBForCtx(ctx).First(&workflow, run.WorkflowID).Error; err != nil {
			if !errors.Is(err, gorm.ErrRecordNotFound) {
				return db.WorkflowRunSnapshot{}, false, fmt.Errorf("load workflow run definition: %w", err)
			}
			slog.ErrorContext(ctx, "workflow run definition unavailable", "workflow_run_id", run.ID, "workflow_id", run.WorkflowID)
			s.completeRun(ctx, run.ID, leaseID, db.ConclusionFailure)
			return db.WorkflowRunSnapshot{}, false, nil
		}
		path = workflow.Path
	}
	gitStore, err := s.GitForCtx(ctx)
	if err != nil {
		return db.WorkflowRunSnapshot{}, false, err
	}
	definition, err := gitStore.ReadFileAtRef(ctx, repo.FullName, path, run.HeadSHA)
	if err != nil {
		if !errors.Is(err, gitstore.ErrCommitNotFound) && !errors.Is(err, gitstore.ErrContentNotFound) {
			return db.WorkflowRunSnapshot{}, false, fmt.Errorf("read workflow run definition %s at %s: %w", path, run.HeadSHA, err)
		}
		slog.ErrorContext(ctx, "legacy workflow run definition unavailable", "workflow_run_id", run.ID, "workflow_id", run.WorkflowID, "error", err)
		s.completeRun(ctx, run.ID, leaseID, db.ConclusionFailure)
		return db.WorkflowRunSnapshot{}, false, nil
	}
	snapshot = db.WorkflowRunSnapshot{WorkflowRunID: run.ID, Definition: db.LargeText(definition)}
	owned := false
	err = s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		result := tx.Model(&db.WorkflowRun{}).
			Where("id = ? AND execution_lease_id = ?", run.ID, leaseID).
			Update("workflow_path", path)
		if result.Error != nil {
			return result.Error
		}
		if result.RowsAffected == 0 {
			return nil
		}
		owned = true
		return tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&snapshot).Error
	})
	if err != nil {
		return db.WorkflowRunSnapshot{}, false, fmt.Errorf("persist workflow run snapshot: %w", err)
	}
	if !owned {
		return db.WorkflowRunSnapshot{}, false, nil
	}
	run.WorkflowPath = path
	return snapshot, true, nil
}

func (s *Service) recoverWorkflowRuns(ctx context.Context) error {
	now := time.Now().UTC()
	var runIDs []uint
	if err := s.DBForCtx(ctx).Model(&db.WorkflowRun{}).
		Where("status IN ? AND (execution_lease_expires_at IS NULL OR execution_lease_expires_at <= ?)",
			[]string{db.RunQueued, db.RunInProgress}, now).
		Order("id ASC").Limit(100).Pluck("id", &runIDs).Error; err != nil {
		return fmt.Errorf("recover workflow runs: %w", err)
	}
	for _, runID := range runIDs {
		if err := s.claimAndStartWorkflowRun(ctx, runID); err != nil {
			return fmt.Errorf("recover workflow run %d: %w", runID, err)
		}
	}
	return nil
}