package service

import (
	"context"
	"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 (
	SyncRunItemStatePlanned    = "planned"
	SyncRunItemStateApplying   = "applying"
	SyncRunItemStateVerifying  = "verifying"
	SyncRunItemStateCompleted  = "completed"
	SyncRunItemStateBlocked    = "blocked"
	SyncRunItemStateFailed     = "failed"
	SyncRunItemStateSuperseded = "superseded"
)

type CompleteRepositorySyncRunItemInput struct {
	TargetIdentity  string
	TargetRevision  string
	TargetHash      string
	TargetNumber    int
	TargetURL       string
	LeaseOwner      string
	LeaseGeneration uint64
}

type FailRepositorySyncRunItemInput struct {
	ErrorCode    string
	ErrorMessage string
	Blocked      bool
}

// RefreshRepositorySyncEndpointLocator persists a provider-validated rename
// without changing the endpoint's immutable repository identity.
func (s *Service) RefreshRepositorySyncEndpointLocator(ctx context.Context, bindingID uint, side, repositoryID, fullName string) error {
	fullName = strings.TrimSpace(fullName)
	if fullName == "" {
		return fmt.Errorf("%w: repository full name is required", ErrValidation)
	}
	result := s.DBForCtx(ctx).Model(&db.SyncBindingEndpoint{}).
		Where("binding_id = ? AND side = ? AND repository_id = ?", bindingID, side, repositoryID).
		Update("repository_full_name", fullName)
	if result.Error != nil {
		return wrapErr(result.Error)
	}
	if result.RowsAffected != 1 {
		return fmt.Errorf("%w: sync endpoint identity changed", ErrConflict)
	}
	return nil
}

// BeginRepositorySyncRunItem persists intent execution before any external
// write. Reclaimed workers may call it for applying/verifying rows; the update
// is idempotent and increments the attempt counter only when a new apply is
// started.
func (s *Service) BeginRepositorySyncRunItem(ctx context.Context, itemID uint) (db.SyncRunItem, error) {
	result := s.DBForCtx(ctx).Model(&db.SyncRunItem{}).
		Where("id = ? AND state IN ?", itemID, []string{SyncRunItemStatePlanned, SyncRunItemStateApplying, SyncRunItemStateVerifying}).
		Updates(map[string]any{
			"state": SyncRunItemStateApplying,
			// Reclaimed applying/verifying intents are read-back recovery of the
			// same attempt. Only the planned -> applying edge starts a new
			// external attempt and increments the audit counter.
			"attempt_count": gorm.Expr("attempt_count + CASE WHEN state = ? THEN 1 ELSE 0 END", SyncRunItemStatePlanned),
		})
	if result.Error != nil {
		return db.SyncRunItem{}, wrapErr(result.Error)
	}
	if result.RowsAffected != 1 {
		return db.SyncRunItem{}, fmt.Errorf("%w: sync run item cannot begin", ErrInvalidState)
	}
	var item db.SyncRunItem
	if err := s.DBForCtx(ctx).First(&item, "id = ?", itemID).Error; err != nil {
		return db.SyncRunItem{}, wrapErr(err)
	}
	return item, nil
}

func (s *Service) BeginRepositorySyncRunItemVerification(ctx context.Context, itemID uint) error {
	result := s.DBForCtx(ctx).Model(&db.SyncRunItem{}).Where("id = ? AND state IN ?", itemID,
		[]string{SyncRunItemStateApplying, SyncRunItemStateVerifying}).Update("state", SyncRunItemStateVerifying)
	if result.Error != nil {
		return wrapErr(result.Error)
	}
	if result.RowsAffected != 1 {
		return fmt.Errorf("%w: sync run item cannot enter verification", ErrInvalidState)
	}
	return nil
}

func (s *Service) SupersedeRepositorySyncRunItem(ctx context.Context, itemID uint, reason string) error {
	result := s.DBForCtx(ctx).Model(&db.SyncRunItem{}).Where("id = ? AND state IN ?", itemID, []string{SyncRunItemStatePlanned, SyncRunItemStateApplying, SyncRunItemStateVerifying}).Updates(map[string]any{"state": SyncRunItemStateSuperseded, "error_code": "source_changed", "error_message": strings.TrimSpace(reason)})
	if result.Error != nil {
		return wrapErr(result.Error)
	}
	if result.RowsAffected != 1 {
		return fmt.Errorf("%w: sync run item cannot be superseded", ErrInvalidState)
	}
	return nil
}

type CompleteRepositorySyncRunInput struct {
	Status                    string
	AuthorityStable           bool
	AllManagedObjectsVerified bool
	SourceInventoryDigest     string
	TargetInventoryDigest     string
	CountersJSON              db.LargeText
	ReportJSON                db.LargeText
	ErrorSummary              string
	LeaseOwner                string
	LeaseGeneration           uint64
}

type CompleteRepositorySyncHandoffPhaseInput struct {
	ExpectedPhase         string
	AuthorityStable       bool
	SourceInventoryDigest string
	TargetInventoryDigest string
	CountersJSON          db.LargeText
	ReportJSON            db.LargeText
	LeaseOwner            string
	LeaseGeneration       uint64
}

// CompleteRepositorySyncRunItem records one verified external side effect and
// updates the corresponding ObjectLink in the same transaction.
func (s *Service) CompleteRepositorySyncRunItem(ctx context.Context, itemID uint, in CompleteRepositorySyncRunItemInput) (db.SyncRunItem, error) {
	database := s.DBForCtx(ctx)
	err := database.Transaction(func(tx *gorm.DB) error {
		item, run, err := loadRepositorySyncRunItemForUpdate(tx, itemID)
		if err != nil {
			return err
		}
		if item.State == SyncRunItemStateCompleted {
			return nil
		}
		if item.State != SyncRunItemStatePlanned && item.State != SyncRunItemStateApplying && item.State != SyncRunItemStateVerifying {
			return fmt.Errorf("%w: sync run item is not completable", ErrInvalidState)
		}
		if err := verifyRepositorySyncRunEpochTx(tx, run); err != nil {
			return err
		}
		var binding db.SyncBinding
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&binding, "id = ?", run.BindingID).Error; err != nil {
			return err
		}
		if err := verifyRepositorySyncRunLeaseTx(binding, run, in.LeaseOwner, in.LeaseGeneration, time.Now().UTC()); err != nil {
			return err
		}
		if item.Action == reposync.ActionCreate && strings.TrimSpace(in.TargetIdentity) == "" {
			return fmt.Errorf("%w: create verification requires target identity", ErrValidation)
		}
		if item.ObjectLinkID != nil {
			var link db.SyncObjectLink
			if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&link, "id = ? AND binding_id = ?", *item.ObjectLinkID, run.BindingID).Error; err != nil {
				return err
			}
			now := time.Now().UTC()
			updates := map[string]any{
				"status": "active", "verified_mapping_version": run.MapperVersion, "last_verified_at": &now,
			}
			if run.SourceSide == SyncSideA {
				updates["side_a_revision"], updates["side_a_hash"] = item.SourceRevision, repositorySyncRunItemSourceHash(item)
				updates["side_b_revision"], updates["side_b_hash"] = in.TargetRevision, in.TargetHash
			} else {
				updates["side_b_revision"], updates["side_b_hash"] = item.SourceRevision, repositorySyncRunItemSourceHash(item)
				updates["side_a_revision"], updates["side_a_hash"] = in.TargetRevision, in.TargetHash
			}
			if item.Action == reposync.ActionDelete {
				updates["status"] = "tombstoned"
				if run.SourceSide == SyncSideA {
					updates["side_a_revision"], updates["side_a_hash"] = "", ""
					updates["side_b_revision"], updates["side_b_hash"] = "", ""
				} else {
					updates["side_b_revision"], updates["side_b_hash"] = "", ""
					updates["side_a_revision"], updates["side_a_hash"] = "", ""
				}
			} else if strings.TrimSpace(in.TargetIdentity) != "" {
				if run.TargetSide == SyncSideA {
					updates["side_a_object_type"], updates["side_a_id"] = item.ObjectType, strings.TrimSpace(in.TargetIdentity)
					if in.TargetNumber > 0 {
						updates["side_a_number"] = in.TargetNumber
					}
					if strings.TrimSpace(in.TargetURL) != "" {
						updates["side_a_url"] = strings.TrimSpace(in.TargetURL)
					}
				} else {
					updates["side_b_object_type"], updates["side_b_id"] = item.ObjectType, strings.TrimSpace(in.TargetIdentity)
					if in.TargetNumber > 0 {
						updates["side_b_number"] = in.TargetNumber
					}
					if strings.TrimSpace(in.TargetURL) != "" {
						updates["side_b_url"] = strings.TrimSpace(in.TargetURL)
					}
				}
			}
			if err := tx.Model(&db.SyncObjectLink{}).Where("id = ?", link.ID).Updates(updates).Error; err != nil {
				return err
			}
		}
		now := time.Now().UTC()
		return tx.Model(&db.SyncRunItem{}).Where("id = ?", item.ID).Updates(map[string]any{
			"state": SyncRunItemStateCompleted, "expected_target_revision": in.TargetRevision,
			"expected_target_hash": in.TargetHash, "completed_at": &now,
			"error_code": "", "error_message": "",
		}).Error
	})
	if err != nil {
		if isDuplicateErr(err) {
			return db.SyncRunItem{}, fmt.Errorf("%w: target identity is already linked", ErrConflict)
		}
		return db.SyncRunItem{}, wrapErr(err)
	}
	var item db.SyncRunItem
	if err := database.First(&item, "id = ?", itemID).Error; err != nil {
		return db.SyncRunItem{}, wrapErr(err)
	}
	return item, nil
}

func (s *Service) FailRepositorySyncRunItem(ctx context.Context, itemID uint, in FailRepositorySyncRunItemInput) (db.SyncRunItem, error) {
	state := SyncRunItemStateFailed
	if in.Blocked {
		state = SyncRunItemStateBlocked
	}
	result := s.DBForCtx(ctx).Model(&db.SyncRunItem{}).
		Where("id = ? AND state IN ?", itemID, []string{SyncRunItemStatePlanned, SyncRunItemStateApplying, SyncRunItemStateVerifying}).
		Updates(map[string]any{
			"state":      state,
			"error_code": strings.TrimSpace(in.ErrorCode), "error_message": in.ErrorMessage,
		})
	if result.Error != nil {
		return db.SyncRunItem{}, wrapErr(result.Error)
	}
	if result.RowsAffected != 1 {
		return db.SyncRunItem{}, fmt.Errorf("%w: sync run item is not fail-able", ErrInvalidState)
	}
	var item db.SyncRunItem
	if err := s.DBForCtx(ctx).First(&item, "id = ?", itemID).Error; err != nil {
		return db.SyncRunItem{}, wrapErr(err)
	}
	return item, nil
}

// CompleteRepositorySyncRun finalizes a normal sync or dry-run. A converged
// status is rejected unless every RunItem is complete and the worker confirms
// both source stability and target verification.
func (s *Service) CompleteRepositorySyncRun(ctx context.Context, runID uint, in CompleteRepositorySyncRunInput) (db.SyncRun, error) {
	status := strings.TrimSpace(in.Status)
	if !isRepositorySyncTerminalStatus(status) {
		return db.SyncRun{}, fmt.Errorf("%w: invalid terminal sync run status", ErrValidation)
	}
	database := s.DBForCtx(ctx)
	err := database.Transaction(func(tx *gorm.DB) error {
		var run db.SyncRun
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&run, "id = ?", runID).Error; err != nil {
			return err
		}
		if run.Kind != SyncRunKindSync || run.Status != SyncRunStatusRunning {
			return fmt.Errorf("%w: normal sync run is not running", ErrInvalidState)
		}
		if err := verifyRepositorySyncRunEpochTx(tx, run); err != nil {
			return err
		}
		var binding db.SyncBinding
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&binding, "id = ?", run.BindingID).Error; err != nil {
			return err
		}
		if err := verifyRepositorySyncRunLeaseTx(binding, run, in.LeaseOwner, in.LeaseGeneration, time.Now().UTC()); err != nil {
			return err
		}
		if status == SyncRunStatusCompleted || status == SyncRunStatusCompletedWithPendingChanges {
			if !in.AllManagedObjectsVerified {
				return fmt.Errorf("%w: managed objects were not fully verified", ErrInvalidState)
			}
			if status == SyncRunStatusCompleted && !in.AuthorityStable {
				return fmt.Errorf("%w: authority inventory changed during the run", ErrConflict)
			}
			allowed := []string{SyncRunItemStateCompleted}
			if status == SyncRunStatusCompletedWithPendingChanges {
				allowed = append(allowed, SyncRunItemStateSuperseded)
			}
			if err := requireRepositorySyncItemsInStatesTx(tx, run.ID, run.Phase, allowed); err != nil {
				return err
			}
		}
		now := time.Now().UTC()
		if err := tx.Model(&db.SyncRun{}).Where("id = ?", run.ID).Updates(map[string]any{
			"status": status, "stage": SyncRunStageCompleted, "source_inventory_digest": in.SourceInventoryDigest,
			"target_inventory_digest": in.TargetInventoryDigest, "counters_json": in.CountersJSON,
			"report_json": in.ReportJSON, "error_code": "", "error_summary": in.ErrorSummary, "completed_at": &now,
		}).Error; err != nil {
			return err
		}
		bindingUpdates := map[string]any{"last_sync_at": &now}
		if status == SyncRunStatusCompleted && run.Mode == SyncRunModeApply {
			bindingUpdates["last_converged_run_id"] = run.ID
			if run.AuditMode == SyncAuditModeFull {
				bindingUpdates["last_full_audit_at"] = &now
			}
		}
		if err := tx.Model(&db.SyncBinding{}).Where("id = ?", run.BindingID).Updates(bindingUpdates).Error; err != nil {
			return err
		}
		released, err := releaseUnstartedRepositorySyncBindingTx(tx, binding, run, status)
		if err != nil {
			return err
		}
		if released {
			return nil
		}
		if status == SyncRunStatusCompletedWithPendingChanges {
			next := db.SyncRun{BindingID: run.BindingID, Kind: SyncRunKindSync, Mode: run.Mode, AuthorityEpoch: run.AuthorityEpoch, SourceSide: run.SourceSide, TargetSide: run.TargetSide, Status: SyncRunStatusQueued, Phase: SyncRunPhaseSync, Stage: SyncRunStageQueued, PlannerMode: SyncPlannerModeAuto, AuditMode: run.AuditMode, MapperVersion: canonical.MapperVersion, VisibilityConfirmation: run.VisibilityConfirmation}
			if err := tx.Create(&next).Error; err != nil {
				return err
			}
			return tx.Model(&db.SyncBinding{}).Where("id = ?", run.BindingID).Update("last_run_id", next.ID).Error
		}
		return nil
	})
	if err != nil {
		return db.SyncRun{}, wrapErr(err)
	}
	return loadRepositorySyncRunWithoutAuthorization(database, runID)
}

// CompleteRepositorySyncHandoffPhase advances final_sync -> reverse_dry_run,
// then atomically switches authority and epoch after a clean reverse dry-run.
func (s *Service) CompleteRepositorySyncHandoffPhase(ctx context.Context, runID uint, in CompleteRepositorySyncHandoffPhaseInput) (db.SyncRun, error) {
	database := s.DBForCtx(ctx)
	err := database.Transaction(func(tx *gorm.DB) error {
		var run db.SyncRun
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&run, "id = ?", runID).Error; err != nil {
			return err
		}
		if run.Kind != SyncRunKindHandoff || run.Status != SyncRunStatusRunning || run.Phase != in.ExpectedPhase {
			return fmt.Errorf("%w: handoff phase is not running", ErrInvalidState)
		}
		if err := verifyRepositorySyncRunEpochTx(tx, run); err != nil {
			return err
		}
		var binding db.SyncBinding
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&binding, "id = ?", run.BindingID).Error; err != nil {
			return err
		}
		if err := verifyRepositorySyncRunLeaseTx(binding, run, in.LeaseOwner, in.LeaseGeneration, time.Now().UTC()); err != nil {
			return err
		}
		if !in.AuthorityStable {
			return fmt.Errorf("%w: authority inventory changed during handoff", ErrConflict)
		}
		if err := requireRepositorySyncItemsCompletedTx(tx, run.ID, run.Phase); err != nil {
			return err
		}
		switch run.Phase {
		case SyncRunPhaseHandoffFinal:
			return tx.Model(&db.SyncRun{}).Where("id = ?", run.ID).Updates(map[string]any{
				"phase": SyncRunPhaseHandoffReverseDry, "status": SyncRunStatusQueued, "stage": SyncRunStageQueued,
				"source_inventory_digest": in.SourceInventoryDigest, "target_inventory_digest": in.TargetInventoryDigest,
				"counters_json": in.CountersJSON, "report_json": in.ReportJSON,
			}).Error
		case SyncRunPhaseHandoffReverseDry:
			now := time.Now().UTC()
			if err := setRepositorySyncTargetClaimTx(tx, run.BindingID, otherSyncSide(run.HandoffToSide)); err != nil {
				return err
			}
			result := tx.Model(&db.SyncBinding{}).Where("id = ? AND authority_epoch = ? AND authority_side = ?", run.BindingID, run.AuthorityEpoch, run.SourceSide).
				Updates(map[string]any{
					"authority_side": run.HandoffToSide, "authority_epoch": run.AuthorityEpoch + 1,
					"last_sync_at": &now, "last_converged_run_id": run.ID,
					"sync_enabled": false, "next_run_at": nil,
					"visibility_confirmation": run.HandoffVisibilityConfirmation,
					"last_full_audit_at":      nil,
				})
			if result.Error != nil {
				return result.Error
			}
			if result.RowsAffected != 1 {
				return fmt.Errorf("%w: authority epoch changed before handoff switch", ErrConflict)
			}
			if err := setAGSRepositorySyncWriteLockTx(tx, run.BindingID, run.HandoffToSide != SyncSideA); err != nil {
				return err
			}
			return tx.Model(&db.SyncRun{}).Where("id = ?", run.ID).Updates(map[string]any{
				"phase": SyncRunPhaseHandoffSwitch, "status": SyncRunStatusCompleted, "stage": SyncRunStageCompleted,
				"target_inventory_digest": in.TargetInventoryDigest, "counters_json": in.CountersJSON,
				"report_json": in.ReportJSON, "completed_at": &now,
			}).Error
		default:
			return fmt.Errorf("%w: unsupported handoff phase", ErrInvalidState)
		}
	})
	if err != nil {
		return db.SyncRun{}, wrapErr(err)
	}
	return loadRepositorySyncRunWithoutAuthorization(database, runID)
}

func loadRepositorySyncRunItemForUpdate(tx *gorm.DB, itemID uint) (db.SyncRunItem, db.SyncRun, error) {
	var item db.SyncRunItem
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&item, "id = ?", itemID).Error; err != nil {
		return db.SyncRunItem{}, db.SyncRun{}, err
	}
	var run db.SyncRun
	if err := tx.First(&run, "id = ?", item.RunID).Error; err != nil {
		return db.SyncRunItem{}, db.SyncRun{}, err
	}
	if run.Status != SyncRunStatusRunning {
		return db.SyncRunItem{}, db.SyncRun{}, fmt.Errorf("%w: sync run is not running", ErrInvalidState)
	}
	return item, run, nil
}

func repositorySyncRunItemSourceHash(item db.SyncRunItem) string {
	if item.SourceHash != "" {
		return item.SourceHash
	}
	return item.DesiredHash
}

func verifyRepositorySyncRunEpochTx(tx *gorm.DB, run db.SyncRun) error {
	var binding db.SyncBinding
	if err := tx.First(&binding, "id = ?", run.BindingID).Error; err != nil {
		return err
	}
	if binding.Status != SyncBindingStatusActive || binding.AuthorityEpoch != run.AuthorityEpoch || binding.AuthoritySide != run.SourceSide {
		return fmt.Errorf("%w: sync run authority epoch is stale", ErrConflict)
	}
	return nil
}

func verifyRepositorySyncRunLeaseTx(binding db.SyncBinding, run db.SyncRun, owner string, generation uint64, now time.Time) error {
	if run.MapperVersion != canonical.MapperVersion {
		return fmt.Errorf("%w: repository sync mapping version changed", ErrConflict)
	}
	if strings.TrimSpace(owner) == "" || generation == 0 || run.LeaseGeneration != generation ||
		binding.LeaseOwner != owner || binding.LeaseGeneration != generation || binding.LeaseExpiresAt == nil || !binding.LeaseExpiresAt.After(now) {
		return fmt.Errorf("%w: repository sync lease was lost", ErrConflict)
	}
	return nil
}

func requireRepositorySyncItemsCompletedTx(tx *gorm.DB, runID uint, phase string) error {
	return requireRepositorySyncItemsInStatesTx(tx, runID, phase, []string{SyncRunItemStateCompleted})
}

func requireRepositorySyncItemsInStatesTx(tx *gorm.DB, runID uint, phase string, states []string) error {
	var incomplete int64
	if err := tx.Model(&db.SyncRunItem{}).Where("run_id = ? AND phase = ? AND state NOT IN ?", runID, phase, states).Count(&incomplete).Error; err != nil {
		return err
	}
	if incomplete != 0 {
		return fmt.Errorf("%w: sync run has %d incomplete or blocking items", ErrConflict, incomplete)
	}
	return nil
}

func isRepositorySyncTerminalStatus(status string) bool {
	switch status {
	case SyncRunStatusCompleted, SyncRunStatusCompletedWithPendingChanges, SyncRunStatusRequiresAttention,
		SyncRunStatusBlocked, SyncRunStatusFailed, SyncRunStatusCancelled, SyncRunStatusSuperseded:
		return true
	default:
		return false
	}
}

func loadRepositorySyncRunWithoutAuthorization(database *gorm.DB, id uint) (db.SyncRun, error) {
	var run db.SyncRun
	if err := database.First(&run, "id = ?", id).Error; err != nil {
		return db.SyncRun{}, wrapErr(err)
	}
	return run, nil
}