package service

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/agent-git-service/internal/db"
	reposync "github.com/pingcap/agent-git-service/internal/sync"
	"github.com/pingcap/agent-git-service/internal/sync/canonical"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const defaultRepositorySyncLeaseTTL = 2 * time.Minute

type RepositorySyncClaim struct {
	Run             db.SyncRun
	Binding         db.SyncBinding
	Endpoints       []db.SyncBindingEndpoint
	LeaseOwner      string
	LeaseGeneration uint64
}

// EnqueueDueRepositorySyncRuns coalesces each due schedule into at most one
// active Run. It advances NextRunAt even when another Run already owns the
// Binding, preventing an outage from creating an unbounded queue.
func (s *Service) EnqueueDueRepositorySyncRuns(ctx context.Context, now time.Time, limit int) (int, error) {
	if limit <= 0 || limit > 1000 {
		limit = 100
	}
	now = now.UTC()
	var bindings []db.SyncBinding
	if err := s.DBForCtx(ctx).
		Where("status = ? AND sync_enabled = ? AND schedule_mode = ? AND next_run_at IS NOT NULL AND next_run_at <= ?", SyncBindingStatusActive, true, SyncScheduleScheduled, now).
		Order("next_run_at ASC, id ASC").Limit(limit).Find(&bindings).Error; err != nil {
		return 0, wrapErr(err)
	}
	enqueued := 0
	fullAuditInterval := s.repositorySyncFullAuditInterval(ctx)
	for _, candidate := range bindings {
		err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
			var binding db.SyncBinding
			if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&binding, "id = ?", candidate.ID).Error; err != nil {
				return err
			}
			if binding.Status != SyncBindingStatusActive || !binding.SyncEnabled || binding.ScheduleMode != SyncScheduleScheduled || binding.NextRunAt == nil || binding.NextRunAt.After(now) {
				return nil
			}
			if err := requireNoRepositorySyncSafetyFenceTx(tx, binding.ID); err != nil {
				// A safety fence intentionally leaves the schedule due. Once an
				// operator resolves it, exactly one coalesced run is queued.
				return nil
			}
			schedule, err := reposync.ParseCron(binding.ScheduleCron)
			if err != nil {
				return fmt.Errorf("binding %d schedule: %w", binding.ID, err)
			}
			next, err := schedule.Next(now)
			if err != nil {
				return err
			}
			var active int64
			if err := tx.Model(&db.SyncRun{}).Where("binding_id = ? AND status IN ?", binding.ID, syncActiveRunStatuses()).Count(&active).Error; err != nil {
				return err
			}
			if active != 0 {
				// Leave the due instant in place as the single coalesced pending
				// request. Once the active Run finishes, the next scheduler pass
				// creates exactly one Run and advances to the next future cron time.
				return nil
			}
			if err := tx.Model(&binding).Update("next_run_at", &next).Error; err != nil {
				return err
			}
			auditMode := SyncAuditModeCheckpoint
			if binding.LastFullAuditAt == nil || !binding.LastFullAuditAt.Add(fullAuditInterval).After(now) {
				auditMode = SyncAuditModeFull
			}
			run := db.SyncRun{
				BindingID: binding.ID, Kind: SyncRunKindSync, Mode: SyncRunModeApply,
				AuthorityEpoch: binding.AuthorityEpoch, SourceSide: binding.AuthoritySide,
				TargetSide: otherSyncSide(binding.AuthoritySide), Status: SyncRunStatusQueued,
				Phase: SyncRunPhaseSync, Stage: SyncRunStageQueued, PlannerMode: SyncPlannerModeAuto,
				AuditMode: auditMode, MapperVersion: canonical.MapperVersion,
				VisibilityConfirmation: binding.VisibilityConfirmation,
			}
			if err := tx.Create(&run).Error; err != nil {
				return err
			}
			if err := tx.Model(&binding).Update("last_run_id", run.ID).Error; err != nil {
				return err
			}
			enqueued++
			return nil
		})
		if err != nil {
			return enqueued, wrapErr(err)
		}
	}
	return enqueued, nil
}

// ClaimNextRepositorySyncRun acquires the Binding lease and either starts a
// queued Run or reclaims a running Run whose previous worker lease expired.
func (s *Service) ClaimNextRepositorySyncRun(ctx context.Context, workerID string, leaseTTL time.Duration) (RepositorySyncClaim, bool, error) {
	workerID = strings.TrimSpace(workerID)
	if workerID == "" {
		return RepositorySyncClaim{}, false, fmt.Errorf("%w: worker id is required", ErrValidation)
	}
	if leaseTTL <= 0 {
		leaseTTL = defaultRepositorySyncLeaseTTL
	}
	now := time.Now().UTC()
	var claim RepositorySyncClaim
	for attempts := 0; attempts < 8; attempts++ {
		found := false
		err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
			var run db.SyncRun
			err := tx.Table("sync_runs AS r").Select("r.*").
				Joins("JOIN sync_bindings AS b ON b.id = r.binding_id").
				Where("b.status = ? AND ((r.status = ? AND (r.retry_after_at IS NULL OR r.retry_after_at <= ?) AND (b.lease_expires_at IS NULL OR b.lease_expires_at <= ?)) OR (r.status = ? AND (b.lease_expires_at IS NULL OR b.lease_expires_at <= ?)))",
					SyncBindingStatusActive, SyncRunStatusQueued, now, now, SyncRunStatusRunning, now).
				Order("CASE WHEN r.kind = 'handoff' THEN 0 ELSE 1 END, r.id ASC").
				Clauses(clause.Locking{Strength: "UPDATE"}).Take(&run).Error
			if err != nil {
				if err == gorm.ErrRecordNotFound {
					return nil
				}
				return err
			}
			var binding db.SyncBinding
			if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Preload("Endpoints").First(&binding, "id = ?", run.BindingID).Error; err != nil {
				return err
			}
			if run.AuthorityEpoch != binding.AuthorityEpoch || run.SourceSide != binding.AuthoritySide {
				completed := now
				if err := tx.Model(&run).Updates(map[string]any{"status": SyncRunStatusSuperseded, "stage": SyncRunStageCompleted, "error_code": "authority_epoch_changed", "completed_at": &completed}).Error; err != nil {
					return err
				}
				if _, err := releaseUnstartedRepositorySyncBindingTx(tx, binding, run, SyncRunStatusSuperseded); err != nil {
					return err
				}
				return nil
			}
			if run.MapperVersion != canonical.MapperVersion {
				var unfinished int64
				if err := tx.Model(&db.SyncRunItem{}).Where("run_id = ? AND state IN ?", run.ID,
					[]string{SyncRunItemStateApplying, SyncRunItemStateVerifying}).Count(&unfinished).Error; err != nil {
					return err
				}
				completed := now
				if unfinished == 0 {
					if err := tx.Model(&run).Updates(map[string]any{"status": SyncRunStatusSuperseded, "stage": SyncRunStageCompleted, "error_code": "mapping_version_changed", "completed_at": &completed,
						"error_summary": "mapping version changed before execution"}).Error; err != nil {
						return err
					}
					if _, err := releaseUnstartedRepositorySyncBindingTx(tx, binding, run, SyncRunStatusSuperseded); err != nil {
						return err
					}
					return nil
				}
				if err := tx.Model(&db.SyncRunItem{}).Where("run_id = ? AND state IN ?", run.ID,
					[]string{SyncRunItemStateApplying, SyncRunItemStateVerifying}).Updates(map[string]any{
					"state": SyncRunItemStateBlocked, "error_code": SyncRunItemErrorMapperChanged,
					"error_message": "mapping version changed while an external intent may still be in flight",
				}).Error; err != nil {
					return err
				}
				if err := tx.Model(&run).Updates(map[string]any{"status": SyncRunStatusBlocked, "stage": SyncRunStageCompleted, "error_code": SyncRunItemErrorMapperChanged, "completed_at": &completed,
					"error_summary": "mapping version changed with unfinished external intents"}).Error; err != nil {
					return err
				}
				if _, err := releaseUnstartedRepositorySyncBindingTx(tx, binding, run, SyncRunStatusBlocked); err != nil {
					return err
				}
				return nil
			}
			generation := binding.LeaseGeneration + 1
			expires := now.Add(leaseTTL)
			result := tx.Model(&db.SyncBinding{}).Where("id = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?)", binding.ID, now).
				Updates(map[string]any{"lease_owner": workerID, "lease_generation": generation, "lease_expires_at": &expires})
			if result.Error != nil {
				return result.Error
			}
			if result.RowsAffected != 1 {
				return nil
			}
			updates := map[string]any{"status": SyncRunStatusRunning, "lease_generation": generation, "last_progress_at": &now, "retry_after_at": nil}
			// A rate-limit pause is an operational state, not a lasting Run
			// failure. Once the Worker reacquires the Run, clear that temporary
			// status so a later successful completion is healthy rather than
			// permanently shown as "needs attention".
			if run.ErrorCode == "rate_limited" {
				updates["error_code"] = ""
				updates["error_summary"] = ""
				run.ErrorCode = ""
				run.ErrorSummary = ""
			}
			if run.StartedAt == nil {
				updates["started_at"] = &now
			}
			if err := tx.Model(&run).Updates(updates).Error; err != nil {
				return err
			}
			run.Status, run.LeaseGeneration = SyncRunStatusRunning, generation
			if run.StartedAt == nil {
				run.StartedAt = &now
			}
			binding.LeaseOwner, binding.LeaseGeneration, binding.LeaseExpiresAt = workerID, generation, &expires
			claim = RepositorySyncClaim{Run: run, Binding: binding, Endpoints: binding.Endpoints, LeaseOwner: workerID, LeaseGeneration: generation}
			found = true
			return nil
		})
		if err != nil {
			return RepositorySyncClaim{}, false, wrapErr(err)
		}
		if found {
			return claim, true, nil
		}
	}
	return RepositorySyncClaim{}, false, nil
}

// DeferRepositorySyncRun safely pauses a provider-limited Run without
// completing or closing its Binding. Its durable plan and completed RunItems
// remain intact; the next worker automatically claims it once retryAt passes.
func (s *Service) DeferRepositorySyncRun(ctx context.Context, claim RepositorySyncClaim, retryAt time.Time, errorCode, summary string) (db.SyncRun, error) {
	now := time.Now().UTC()
	if retryAt.IsZero() || retryAt.Before(now.Add(10*time.Second)) {
		retryAt = now.Add(5 * time.Minute)
	}
	retryAt = retryAt.UTC()
	if err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		var run db.SyncRun
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&run, "id = ?", claim.Run.ID).Error; err != nil {
			return err
		}
		if run.Status != SyncRunStatusRunning || run.LeaseGeneration != claim.LeaseGeneration {
			return fmt.Errorf("%w: sync run is not running", ErrInvalidState)
		}
		var binding db.SyncBinding
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&binding, "id = ?", run.BindingID).Error; err != nil {
			return err
		}
		if binding.LeaseOwner != claim.LeaseOwner || binding.LeaseGeneration != claim.LeaseGeneration {
			return fmt.Errorf("%w: repository sync lease was lost before pause", ErrConflict)
		}
		return tx.Model(&run).Updates(map[string]any{
			"status":           SyncRunStatusQueued,
			"stage":            SyncRunStageQueued,
			"retry_after_at":   &retryAt,
			"error_code":       strings.TrimSpace(errorCode),
			"error_summary":    strings.TrimSpace(summary),
			"last_progress_at": &now,
		}).Error
	}); err != nil {
		return db.SyncRun{}, wrapErr(err)
	}
	return loadRepositorySyncRunWithoutAuthorization(s.DBForCtx(ctx), claim.Run.ID)
}

// UpdateRepositorySyncRunStage records a user-visible execution stage while
// fencing the update to the worker that currently owns the Binding lease.
func (s *Service) UpdateRepositorySyncRunStage(ctx context.Context, claim RepositorySyncClaim, stage string) error {
	switch stage {
	case SyncRunStageCheckingAccess, SyncRunStageInventorySource, SyncRunStageInventoryTarget,
		SyncRunStagePlanning, SyncRunStageApplying, SyncRunStageVerifying, SyncRunStageFinalizing:
	default:
		return fmt.Errorf("%w: unsupported repository sync stage %q", ErrValidation, stage)
	}
	result := s.DBForCtx(ctx).Model(&db.SyncRun{}).
		Where("id = ? AND status = ? AND lease_generation = ?", claim.Run.ID, SyncRunStatusRunning, claim.LeaseGeneration).
		Updates(map[string]any{"stage": stage, "last_progress_at": time.Now().UTC()})
	if result.Error != nil {
		return wrapErr(result.Error)
	}
	if result.RowsAffected != 1 {
		return fmt.Errorf("%w: repository sync run stage lease was lost", ErrConflict)
	}
	return nil
}

func (s *Service) RenewRepositorySyncLease(ctx context.Context, bindingID uint, workerID string, generation uint64, leaseTTL time.Duration) error {
	_, err := s.RenewRepositorySyncLeaseUntil(ctx, bindingID, workerID, generation, leaseTTL)
	return err
}

// RenewRepositorySyncLeaseUntil renews the fenced binding lease and returns
// its new expiry. Callers that run a long-lived executor use the expiry to
// guarantee that a blocked renewal never keeps operating beyond its lease.
func (s *Service) RenewRepositorySyncLeaseUntil(ctx context.Context, bindingID uint, workerID string, generation uint64, leaseTTL time.Duration) (time.Time, error) {
	if leaseTTL <= 0 {
		leaseTTL = defaultRepositorySyncLeaseTTL
	}
	now := time.Now().UTC()
	expires := now.Add(leaseTTL)
	// A lease renewal is one fenced SQL statement. Avoid GORM's default
	// transaction wrapper here: it cannot make a single UPDATE more atomic, and
	// a cancelled wrapper can surface a misleading "transaction rolled back"
	// error instead of the retryable TiDB lock-wait result.
	database := s.DBForCtx(ctx).Session(&gorm.Session{SkipDefaultTransaction: true})
	result := database.Model(&db.SyncBinding{}).
		Where("id = ? AND lease_owner = ? AND lease_generation = ? AND lease_expires_at > ?", bindingID, workerID, generation, now).
		Update("lease_expires_at", &expires)
	if result.Error != nil {
		return time.Time{}, wrapErr(result.Error)
	}
	if result.RowsAffected != 1 {
		return time.Time{}, fmt.Errorf("%w: repository sync lease was lost", ErrConflict)
	}
	// Keep this separate from the lease row update. The heartbeat must not make
	// the Run's status depend on a second write, but successful progress makes
	// stalled/orphaned Runs observable to the watchdog and UI.
	// The binding lease is the correctness fence. Progress is observability only
	// and must not revoke a successfully-renewed lease if this independent write
	// races with a transient database timeout.
	if err := database.Model(&db.SyncRun{}).
		Where("binding_id = ? AND status = ? AND lease_generation = ?", bindingID, SyncRunStatusRunning, generation).
		Update("last_progress_at", now).Error; err != nil {
		return expires, nil
	}
	return expires, nil
}

func (s *Service) ReleaseRepositorySyncLease(ctx context.Context, bindingID uint, workerID string, generation uint64) error {
	result := s.DBForCtx(ctx).Model(&db.SyncBinding{}).
		Where("id = ? AND lease_owner = ? AND lease_generation = ?", bindingID, workerID, generation).
		Updates(map[string]any{"lease_owner": "", "lease_expires_at": nil})
	if result.Error != nil {
		return wrapErr(result.Error)
	}
	if result.RowsAffected != 1 {
		return fmt.Errorf("%w: repository sync lease was lost", ErrConflict)
	}
	return nil
}

func (s *Service) AssertRepositorySyncLease(ctx context.Context, claim RepositorySyncClaim) error {
	now := time.Now().UTC()
	var count int64
	err := s.DBForCtx(ctx).Table("sync_bindings AS b").Joins("JOIN sync_runs AS r ON r.binding_id = b.id").
		Where("b.id = ? AND b.lease_owner = ? AND b.lease_generation = ? AND b.lease_expires_at > ? AND b.authority_epoch = ? AND b.authority_side = ? AND r.id = ? AND r.mapper_version = ?",
			claim.Binding.ID, claim.LeaseOwner, claim.LeaseGeneration, now, claim.Run.AuthorityEpoch, claim.Run.SourceSide, claim.Run.ID, canonical.MapperVersion).
		Count(&count).Error
	if err != nil {
		return wrapErr(err)
	}
	if count != 1 {
		return fmt.Errorf("%w: repository sync lease, authority epoch, or mapping version was lost", ErrConflict)
	}
	return nil
}

// RepositorySyncLinkViews returns endpoint-oriented ObjectLinks for one run
// direction. sourceSide may be reversed during Handoff dry-run while the
// persisted authority remains unchanged until the final epoch switch.
func (s *Service) RepositorySyncLinkViews(ctx context.Context, bindingID uint, sourceSide string) ([]reposync.LinkView, error) {
	if sourceSide != SyncSideA && sourceSide != SyncSideB {
		return nil, fmt.Errorf("%w: invalid sync source side", ErrValidation)
	}
	var links []db.SyncObjectLink
	if err := s.DBForCtx(ctx).Where("binding_id = ?", bindingID).Order("id ASC").Find(&links).Error; err != nil {
		return nil, wrapErr(err)
	}
	views := make([]reposync.LinkView, 0, len(links))
	for _, link := range links {
		var sourceID, targetID *string
		var sourceRevision, sourceHash, targetRevision, targetHash string
		if sourceSide == SyncSideA {
			sourceID, targetID = link.SideAID, link.SideBID
			sourceRevision, sourceHash = link.SideARevision, link.SideAHash
			targetRevision, targetHash = link.SideBRevision, link.SideBHash
		} else {
			sourceID, targetID = link.SideBID, link.SideAID
			sourceRevision, sourceHash = link.SideBRevision, link.SideBHash
			targetRevision, targetHash = link.SideARevision, link.SideAHash
		}
		if sourceID == nil || strings.TrimSpace(*sourceID) == "" {
			continue
		}
		view := reposync.LinkView{
			ID: link.ID, ObjectType: link.LogicalObjectType, LogicalKey: link.LogicalKey,
			SourceIdentity: *sourceID, Status: link.Status,
			SourceRevision: sourceRevision, SourceHash: sourceHash,
			TargetRevision: targetRevision, TargetHash: targetHash,
			MappingVersion: link.VerifiedMappingVersion,
			Metadata:       append(json.RawMessage(nil), []byte(link.MetadataJSON)...),
		}
		if targetID != nil {
			view.TargetIdentity = *targetID
		}
		views = append(views, view)
	}
	return views, nil
}

func (s *Service) TerminateRepositorySyncRun(ctx context.Context, claim RepositorySyncClaim, status, errorCode, summary string) (db.SyncRun, error) {
	if !isRepositorySyncTerminalStatus(status) || status == SyncRunStatusCompleted || status == SyncRunStatusCompletedWithPendingChanges {
		return db.SyncRun{}, fmt.Errorf("%w: invalid exceptional terminal status", ErrValidation)
	}
	now := time.Now().UTC()
	err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		var run db.SyncRun
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&run, "id = ?", claim.Run.ID).Error; err != nil {
			return err
		}
		if run.Status != SyncRunStatusRunning || run.LeaseGeneration != claim.LeaseGeneration {
			return fmt.Errorf("%w: sync run is not running", ErrInvalidState)
		}
		var binding db.SyncBinding
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&binding, "id = ?", run.BindingID).Error; err != nil {
			return err
		}
		// Generation fencing is sufficient for exceptional termination. A
		// heartbeat can time out just before this transaction; a replacement
		// Worker cannot write this Run without first changing the generation.
		if binding.LeaseOwner != claim.LeaseOwner || binding.LeaseGeneration != claim.LeaseGeneration {
			return fmt.Errorf("%w: repository sync lease was lost before termination", ErrConflict)
		}
		if err := tx.Model(&run).Updates(map[string]any{"status": status, "stage": SyncRunStageCompleted, "error_code": strings.TrimSpace(errorCode), "error_summary": strings.TrimSpace(summary), "completed_at": &now}).Error; err != nil {
			return err
		}
		if _, err := releaseUnstartedRepositorySyncBindingTx(tx, binding, run, status); err != nil {
			return err
		}
		if run.Kind == SyncRunKindHandoff && run.SourceSide == SyncSideA {
			return setAGSRepositorySyncWriteLockTx(tx, run.BindingID, false)
		}
		return nil
	})
	if err != nil {
		return db.SyncRun{}, wrapErr(err)
	}
	return loadRepositorySyncRunWithoutAuthorization(s.DBForCtx(ctx), claim.Run.ID)
}

func (s *Service) RepositorySyncRunCancellationRequested(ctx context.Context, runID uint) (bool, error) {
	var requested bool
	if err := s.DBForCtx(ctx).Model(&db.SyncRun{}).Select("cancel_requested").Where("id = ?", runID).Scan(&requested).Error; err != nil {
		return false, wrapErr(err)
	}
	return requested, nil
}