package service

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strconv"
	"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"
)

type PersistRepositorySyncPlanInput struct {
	Source                []reposync.InventoryObject
	Target                []reposync.InventoryObject
	Links                 []reposync.LinkView
	SourceInventoryDigest string
	TargetInventoryDigest string
	TargetConfirmedEmpty  bool
	CheckpointPlan        bool
	CheckpointSkipped     int
	// FastForwardUnlinkedRefs is populated by the worker only after Git has
	// verified that an existing target branch is an ancestor of the authority
	// branch. It is intentionally not inferred from matching ref names.
	FastForwardUnlinkedRefs map[string]bool
	ConfirmedUnlinkedRefs   map[string]bool
	BatchLimits             reposync.BatchLimits
	LeaseOwner              string
	LeaseGeneration         uint64
}

type PersistRepositorySyncPlanResult struct {
	PlannerMode string
	DryRun      bool
	Items       []reposync.PlannedItem
	Batches     []reposync.PhysicalBatch
}

// repositorySyncPlanPersistenceChunkSize bounds the lifetime of the database
// transaction that persists a logical plan. Planning can contain thousands of
// objects. Keeping all of their links and RunItems in one transaction holds
// the parent Binding row long enough to starve the worker's fenced lease
// heartbeat on TiDB.
//
// This is deliberately independent from the physical apply batch size: it is
// a database correctness boundary, not an external-provider throughput knob.
const repositorySyncPlanPersistenceChunkSize = 16

// PersistRepositorySyncPlan is the worker boundary between provider inventory
// and execution. It materializes object-level RunItems while returning bounded
// physical batches. It intentionally does not require a request user: callers
// are tenant-scoped workers operating on a run authorized at creation.
func (s *Service) PersistRepositorySyncPlan(ctx context.Context, runID uint, in PersistRepositorySyncPlanInput) (PersistRepositorySyncPlanResult, error) {
	var (
		items       []reposync.PlannedItem
		plannerMode string
		err         error
	)
	if in.TargetConfirmedEmpty {
		plannerMode = reposync.PlannerModeBootstrap
		items, err = reposync.PlanBootstrap(in.Source, in.Target)
	} else {
		plannerMode = reposync.PlannerModeCurrentState
		if in.CheckpointPlan {
			plannerMode = reposync.PlannerModeCheckpoint
		}
		items, err = reposync.PlanCurrentStateWithOptions(in.Source, in.Target, in.Links, reposync.CurrentStatePlanOptions{
			FastForwardUnlinkedRefs: in.FastForwardUnlinkedRefs,
			ConfirmedUnlinkedRefs:   in.ConfirmedUnlinkedRefs,
		})
	}
	if err != nil {
		return PersistRepositorySyncPlanResult{}, err
	}
	for i := range items {
		items[i].DependencyPhase = repositorySyncDependencyPhase(items[i])
	}
	_, err = reposync.BuildPhysicalBatches(items, in.BatchLimits)
	if err != nil {
		return PersistRepositorySyncPlanResult{}, err
	}

	counters := map[string]int{"scanned": len(in.Source), "checkpoint_skipped": in.CheckpointSkipped}
	for _, item := range items {
		counters[item.Action]++
	}
	countersJSON, err := json.Marshal(counters)
	if err != nil {
		return PersistRepositorySyncPlanResult{}, err
	}

	database := s.DBForCtx(ctx)
	dryRun := false
	for start := 0; start < len(items); start += repositorySyncPlanPersistenceChunkSize {
		end := start + repositorySyncPlanPersistenceChunkSize
		if end > len(items) {
			end = len(items)
		}
		chunkDryRun, err := s.persistRepositorySyncPlanChunk(ctx, database, runID, items[start:end], in)
		if err != nil {
			if isDuplicateErr(err) {
				return PersistRepositorySyncPlanResult{}, fmt.Errorf("%w: sync plan contains a duplicate logical object or identity", ErrConflict)
			}
			return PersistRepositorySyncPlanResult{}, wrapErr(err)
		}
		dryRun = chunkDryRun
	}

	// A plan is considered durable only after every chunk was stored. This
	// final short transaction is the completion marker used by recovery. If a
	// worker dies before it commits, the next worker re-inventories and replays
	// the plan idempotently rather than applying a partial plan.
	if err := database.Transaction(func(tx *gorm.DB) error {
		var run db.SyncRun
		if _, err := loadRepositorySyncPlanRunAndLeaseTx(tx, runID, in.LeaseOwner, in.LeaseGeneration, &run); err != nil {
			return err
		}
		dryRun = run.Mode == SyncRunModeDryRun || run.Phase == SyncRunPhaseHandoffReverseDry
		var persistedItems int64
		if err := tx.Model(&db.SyncRunItem{}).Where("run_id = ? AND phase = ?", run.ID, run.Phase).Count(&persistedItems).Error; err != nil {
			return err
		}
		if persistedItems != int64(len(items)) {
			// A replayed inventory may have removed an object that was already
			// persisted by a prior, incomplete attempt. Do not silently finalize
			// a mixed plan: no external write has occurred yet, so surfacing a
			// conflict is safer than applying a stale intent.
			return fmt.Errorf("%w: sync run contains stale planned objects", ErrConflict)
		}
		runUpdates := map[string]any{"planner_mode": plannerMode, "counters_json": db.LargeText(countersJSON)}
		if run.Phase != SyncRunPhaseHandoffReverseDry {
			runUpdates["source_inventory_digest"] = in.SourceInventoryDigest
			runUpdates["target_inventory_digest"] = in.TargetInventoryDigest
		}
		// During reverse dry-run these two columns deliberately retain the
		// handoff_final snapshots: source=old authority, target=old standby.
		// A reclaimed reverse worker uses target as its source baseline and must
		// still compare the old authority against the original source digest.
		return tx.Model(&db.SyncRun{}).Where("id = ? AND status = ?", run.ID, SyncRunStatusRunning).Updates(runUpdates).Error
	}); err != nil {
		return PersistRepositorySyncPlanResult{}, wrapErr(err)
	}
	batches, err := reposync.BuildPhysicalBatches(items, in.BatchLimits)
	if err != nil {
		return PersistRepositorySyncPlanResult{}, err
	}
	return PersistRepositorySyncPlanResult{
		PlannerMode: plannerMode, DryRun: dryRun,
		Items: items, Batches: batches,
	}, nil
}

// persistRepositorySyncPlanChunk persists a bounded part of a plan in its own
// transaction. A short transaction is essential here: SyncObjectLink has a
// foreign key to SyncBinding, and a long sequence of link writes can block the
// separate lease heartbeat from updating that Binding.
func (s *Service) persistRepositorySyncPlanChunk(ctx context.Context, database *gorm.DB, runID uint, items []reposync.PlannedItem, in PersistRepositorySyncPlanInput) (bool, error) {
	dryRun := false
	err := database.Transaction(func(tx *gorm.DB) error {
		var run db.SyncRun
		if _, err := loadRepositorySyncPlanRunAndLeaseTx(tx, runID, in.LeaseOwner, in.LeaseGeneration, &run); err != nil {
			return err
		}
		dryRun = run.Mode == SyncRunModeDryRun || run.Phase == SyncRunPhaseHandoffReverseDry
		now := time.Now().UTC()
		for index := range items {
			if err := persistRepositorySyncPlanItemTx(tx, run, &items[index], dryRun, now); err != nil {
				return err
			}
		}
		return nil
	})
	return dryRun, err
}

// loadRepositorySyncPlanRunAndLeaseTx validates the same fenced lease used by
// execution. It is intentionally called once per short plan transaction, not
// once for the entire repository-sized plan.
func loadRepositorySyncPlanRunAndLeaseTx(tx *gorm.DB, runID uint, leaseOwner string, leaseGeneration uint64, run *db.SyncRun) (db.SyncBinding, error) {
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(run, "id = ?", runID).Error; err != nil {
		return db.SyncBinding{}, err
	}
	if run.Status != SyncRunStatusRunning {
		return db.SyncBinding{}, fmt.Errorf("%w: sync run is not running", ErrInvalidState)
	}
	var binding db.SyncBinding
	if err := tx.First(&binding, "id = ?", run.BindingID).Error; err != nil {
		return db.SyncBinding{}, err
	}
	if binding.AuthorityEpoch != run.AuthorityEpoch || binding.AuthoritySide != run.SourceSide {
		return db.SyncBinding{}, fmt.Errorf("%w: sync run authority epoch is stale", ErrConflict)
	}
	if err := verifyRepositorySyncRunLeaseTx(binding, *run, leaseOwner, leaseGeneration, time.Now().UTC()); err != nil {
		return db.SyncBinding{}, err
	}
	return binding, nil
}

func persistRepositorySyncPlanItemTx(tx *gorm.DB, run db.SyncRun, item *reposync.PlannedItem, dryRun bool, now time.Time) error {
	linkID := item.ObjectLinkID
	if linkID != nil {
		var linkedCount int64
		if err := tx.Model(&db.SyncObjectLink{}).Where("id = ? AND binding_id = ?", *linkID, run.BindingID).Count(&linkedCount).Error; err != nil {
			return err
		}
		if linkedCount != 1 {
			return fmt.Errorf("%w: object link does not belong to the sync binding", ErrConflict)
		}
	}
	if linkID == nil && (item.Action == reposync.ActionCreate || item.Representation == "archive_only" ||
		((item.Action == reposync.ActionNoop || item.Action == reposync.ActionUpdate) && item.TargetIdentity != "")) && !dryRun {
		link, err := ensureRepositorySyncObjectLinkTx(tx, run, *item)
		if err != nil {
			return err
		}
		linkID = &link.ID
		item.ObjectLinkID = linkID
	}
	if linkID != nil && item.SourceIdentity != "" && len(item.Payload) != 0 && !dryRun {
		representationColumn := "representation_b"
		numberColumn, urlColumn := "side_b_number", "side_b_url"
		if run.TargetSide == SyncSideA {
			representationColumn = "representation_a"
		}
		if run.SourceSide == SyncSideA {
			numberColumn, urlColumn = "side_a_number", "side_a_url"
		}
		updates := map[string]any{
			"metadata_json": db.LargeText(item.Payload), representationColumn: item.Representation,
		}
		if record, decodeErr := canonical.Decode(item.Payload); decodeErr == nil {
			if number := canonical.IntField(record.Fields, "_number"); number > 0 {
				updates[numberColumn] = number
			}
			if record.SourceURL != "" {
				updates[urlColumn] = record.SourceURL
			}
		}
		if err := tx.Model(&db.SyncObjectLink{}).Where("id = ? AND binding_id = ?", *linkID, run.BindingID).Updates(updates).Error; err != nil {
			return err
		}
	}
	if linkID != nil && item.TargetIdentity != "" && item.Action != reposync.ActionDelete && item.Action != reposync.ActionConflict && !dryRun {
		column := "side_b_id"
		typeColumn := "side_b_object_type"
		if run.TargetSide == SyncSideA {
			column, typeColumn = "side_a_id", "side_a_object_type"
		}
		result := tx.Model(&db.SyncObjectLink{}).Where("id = ? AND binding_id = ? AND "+column+" IS NULL", *linkID, run.BindingID).
			Updates(map[string]any{column: item.TargetIdentity, typeColumn: item.ObjectType, "status": "active"})
		if result.Error != nil {
			return result.Error
		}
	}
	if linkID != nil && !dryRun && (item.Action == reposync.ActionNoop || item.Representation == "archive_only") {
		checkpointUpdates := map[string]any{
			"verified_mapping_version": run.MapperVersion, "last_verified_at": &now,
		}
		if run.SourceSide == SyncSideA {
			checkpointUpdates["side_a_revision"], checkpointUpdates["side_a_hash"] = item.SourceRevision, item.SourceHash
			checkpointUpdates["side_b_revision"], checkpointUpdates["side_b_hash"] = item.ExpectedTargetRevision, item.ExpectedTargetHash
		} else {
			checkpointUpdates["side_b_revision"], checkpointUpdates["side_b_hash"] = item.SourceRevision, item.SourceHash
			checkpointUpdates["side_a_revision"], checkpointUpdates["side_a_hash"] = item.ExpectedTargetRevision, item.ExpectedTargetHash
		}
		if err := tx.Model(&db.SyncObjectLink{}).Where("id = ? AND binding_id = ?", *linkID, run.BindingID).Updates(checkpointUpdates).Error; err != nil {
			return err
		}
	}
	intent, err := json.Marshal(item)
	if err != nil {
		return err
	}
	state := SyncRunItemStatePlanned
	var completedAt *time.Time
	switch {
	case item.Action == reposync.ActionConflict:
		state = SyncRunItemStateBlocked
	case dryRun || item.Action == reposync.ActionNoop || item.Action == reposync.ActionReport:
		state = SyncRunItemStateCompleted
		completedAt = &now
	}
	runItem := db.SyncRunItem{
		RunID: run.ID, ObjectLinkID: linkID, Phase: run.Phase,
		ObjectType: item.ObjectType, ObjectKey: item.ObjectKey, Action: item.Action,
		State: state, DependencyPhase: item.DependencyPhase, PayloadBytes: item.PayloadBytes,
		SourceRevision: item.SourceRevision, SourceHash: item.SourceHash, DesiredHash: item.DesiredHash,
		ExpectedTargetRevision: item.ExpectedTargetRevision, ExpectedTargetHash: item.ExpectedTargetHash,
		IntentJSON: db.LargeText(intent), CompletedAt: completedAt,
	}
	if linkID != nil {
		runItem.IdempotencyKey = repositorySyncLinkIdempotencyKey(run.BindingID, *linkID)
	}
	return ensureRepositorySyncRunItemTx(tx, runItem)
}

// ensureRepositorySyncRunItemTx makes a recovered planning pass idempotent.
// It only accepts an exact copy of a previously stored intent; a changed
// inventory is never silently mixed into the same Run.
func ensureRepositorySyncRunItemTx(tx *gorm.DB, candidate db.SyncRunItem) error {
	var existing db.SyncRunItem
	err := tx.Where("run_id = ? AND phase = ? AND object_type = ? AND object_key = ?", candidate.RunID, candidate.Phase, candidate.ObjectType, candidate.ObjectKey).Take(&existing).Error
	if err == nil {
		if existing.Action != candidate.Action || existing.DependencyPhase != candidate.DependencyPhase || existing.IntentJSON != candidate.IntentJSON {
			return fmt.Errorf("%w: sync run already contains a different planned object", ErrConflict)
		}
		return nil
	}
	if err != gorm.ErrRecordNotFound {
		return err
	}
	return tx.Create(&candidate).Error
}

// LoadRepositorySyncExecutionPlan reconstructs physical batches entirely from
// durable RunItems. This is the crash-recovery path and is also used after a
// worker loses its in-memory batch between apply and verification.
func (s *Service) LoadRepositorySyncExecutionPlan(ctx context.Context, runID uint, phase string, limits reposync.BatchLimits) ([]reposync.PhysicalBatch, map[string]db.SyncRunItem, error) {
	var run db.SyncRun
	if err := s.DBForCtx(ctx).Select("id", "source_side").First(&run, "id = ?", runID).Error; err != nil {
		return nil, nil, wrapErr(err)
	}
	effectiveSourceSide := run.SourceSide
	if phase == SyncRunPhaseHandoffReverseDry {
		effectiveSourceSide = otherSyncSide(effectiveSourceSide)
	}
	var rows []db.SyncRunItem
	if err := s.DBForCtx(ctx).Where("run_id = ? AND phase = ? AND state IN ?", runID, phase,
		[]string{SyncRunItemStatePlanned, SyncRunItemStateApplying, SyncRunItemStateVerifying}).Order("dependency_phase ASC, id ASC").Find(&rows).Error; err != nil {
		return nil, nil, wrapErr(err)
	}
	items := make([]reposync.PlannedItem, 0, len(rows))
	rowByKey := make(map[string]db.SyncRunItem, len(rows))
	for _, row := range rows {
		var item reposync.PlannedItem
		if err := json.Unmarshal([]byte(row.IntentJSON), &item); err != nil {
			return nil, nil, fmt.Errorf("decode sync run item %d intent: %w", row.ID, err)
		}
		item.ObjectLinkID = row.ObjectLinkID
		item.SourceHash = row.SourceHash
		if item.TargetIdentity == "" && row.ObjectLinkID != nil {
			var link db.SyncObjectLink
			if err := s.DBForCtx(ctx).Select("side_a_id", "side_b_id").First(&link, "id = ?", *row.ObjectLinkID).Error; err != nil {
				return nil, nil, wrapErr(err)
			}
			var targetID *string
			if effectiveSourceSide == SyncSideA {
				targetID = link.SideBID
			} else {
				targetID = link.SideAID
			}
			if targetID != nil {
				item.TargetIdentity = *targetID
			}
		}
		item.DependencyPhase = row.DependencyPhase
		item.PayloadBytes = row.PayloadBytes
		items = append(items, item)
		rowByKey[repositorySyncRunItemExecutionKey(item)] = row
	}
	batches, err := reposync.BuildPhysicalBatches(items, limits)
	if err != nil {
		return nil, nil, err
	}
	return batches, rowByKey, nil
}

func repositorySyncRunItemExecutionKey(item reposync.PlannedItem) string {
	return fmt.Sprintf("%d\x00%s\x00%s\x00%s", item.DependencyPhase, item.ObjectType, item.Action, item.ObjectKey)
}

func ensureRepositorySyncObjectLinkTx(tx *gorm.DB, run db.SyncRun, item reposync.PlannedItem) (db.SyncObjectLink, error) {
	var existing db.SyncObjectLink
	err := tx.Where("binding_id = ? AND logical_object_type = ? AND logical_key = ?", run.BindingID, item.ObjectType, item.ObjectKey).Take(&existing).Error
	if err == nil {
		var existingSourceID *string
		if run.SourceSide == SyncSideA {
			existingSourceID = existing.SideAID
		} else {
			existingSourceID = existing.SideBID
		}
		if existingSourceID == nil || *existingSourceID != item.SourceIdentity {
			return db.SyncObjectLink{}, fmt.Errorf("%w: logical object is already linked to another source identity", ErrConflict)
		}
		return existing, nil
	}
	if err != nil && err != gorm.ErrRecordNotFound {
		return db.SyncObjectLink{}, err
	}
	sourceIdentity := item.SourceIdentity
	link := db.SyncObjectLink{
		BindingID: run.BindingID, LogicalObjectType: item.ObjectType, LogicalKey: item.ObjectKey,
		OriginSide: run.SourceSide, IdempotencyKey: repositorySyncPendingIdempotencyKey(run.BindingID, item.ObjectType, item.ObjectKey),
		Status: "active", MetadataJSON: db.LargeText(item.Payload),
	}
	if run.SourceSide == SyncSideA {
		link.SideAObjectType, link.SideAID = item.ObjectType, &sourceIdentity
		link.SideBObjectType = item.ObjectType
		link.RepresentationA, link.RepresentationB = "native", item.Representation
	} else {
		link.SideBObjectType, link.SideBID = item.ObjectType, &sourceIdentity
		link.SideAObjectType = item.ObjectType
		link.RepresentationB, link.RepresentationA = "native", item.Representation
	}
	if record, decodeErr := canonical.Decode(item.Payload); decodeErr == nil {
		number := canonical.IntField(record.Fields, "_number")
		if run.SourceSide == SyncSideA {
			if number > 0 {
				link.SideANumber = &number
			}
			link.SideAURL = record.SourceURL
		} else {
			if number > 0 {
				link.SideBNumber = &number
			}
			link.SideBURL = record.SourceURL
		}
	}
	if err := tx.Create(&link).Error; err != nil {
		return db.SyncObjectLink{}, err
	}
	link.IdempotencyKey = repositorySyncLinkIdempotencyKey(run.BindingID, link.ID)
	if err := tx.Model(&link).Update("idempotency_key", link.IdempotencyKey).Error; err != nil {
		return db.SyncObjectLink{}, err
	}
	return link, nil
}

func repositorySyncLinkIdempotencyKey(bindingID, linkID uint) string {
	sum := sha256.Sum256([]byte(strconv.FormatUint(uint64(bindingID), 10) + "\x00" + strconv.FormatUint(uint64(linkID), 10)))
	return "sync-v1-" + hex.EncodeToString(sum[:])
}

func repositorySyncPendingIdempotencyKey(bindingID uint, objectType, objectKey string) string {
	sum := sha256.Sum256([]byte(strconv.FormatUint(uint64(bindingID), 10) + "\x00pending\x00" + objectType + "\x00" + objectKey))
	return "sync-pending-v1-" + hex.EncodeToString(sum[:])
}

func repositorySyncDependencyPhase(item reposync.PlannedItem) int {
	switch item.ObjectType {
	case "lfs_objects":
		if item.Action == reposync.ActionDelete {
			return 15
		}
		return 5
	case "git_refs", "wiki_refs":
		return 10
	case "labels", "milestones":
		return 20
	case "issues":
		return 30
	case "issue_comments":
		if record, err := canonical.Decode(item.Payload); err == nil && record.ParentType == canonical.TypePullRequest {
			return 55
		}
		return 40
	case "pull_requests":
		return 50
	case "pull_request_reviews", "review_requests":
		return 60
	case "pull_request_review_comments":
		return 70
	case "releases":
		return 80
	case "release_assets":
		return 90
	default:
		return 100
	}
}