package clusterrecovery

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"github.com/pingcap-inc/clinic/pkg/datasource/view"
	"github.com/pingcap-inc/clinic/pkg/db"
)

var (
	ErrMutationConflict  = errors.New("recovery mutation conflict")
	ErrRowDrift          = errors.New("Clinic row changed after recovery planning")
	ErrParentUnavailable = errors.New("required resource-pool parent is unavailable")
)

type MutationResult struct {
	Outcome        MutationOutcome
	BeforeChecksum string
	AfterChecksum  string
}

type MutationStore interface {
	ApplyCreate(context.Context, ObjectKind, *db.Cluster, string) (*MutationResult, error)
	ApplyUpdate(context.Context, ObjectKind, *db.Cluster, string, string, []string) (*MutationResult, error)
}

type RecoveryStore interface {
	ReadStore
	MutationStore
}

func (s *GormStore) ApplyCreate(
	ctx context.Context,
	kind ObjectKind,
	desired *db.Cluster,
	expectedDesiredChecksum string,
) (*MutationResult, error) {
	if err := s.validateMutationInput(kind, desired, expectedDesiredChecksum); err != nil {
		return nil, err
	}
	var result *MutationResult
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := verifyOrganizationForMutation(tx, desired); err != nil {
			return err
		}
		if err := verifyParentForMutation(tx, desired); err != nil {
			return err
		}
		existing, err := readClusterForMutation(tx, desired.ClusterID)
		switch {
		case err == nil:
			if err := verifyExistingIdentity(existing, desired); err != nil {
				return err
			}
			changes, err := diffOwnedFields(kind, existing, desired)
			if err != nil {
				return err
			}
			if len(changes) != 0 {
				return ErrRowDrift
			}
			checksum, err := clusterRecordChecksum(existing)
			if err != nil {
				return err
			}
			result = &MutationResult{Outcome: MutationOutcomeNoop, BeforeChecksum: checksum, AfterChecksum: checksum}
			return nil
		case !errors.Is(err, gorm.ErrRecordNotFound):
			return err
		}

		create := *desired
		create.ID = 0
		create.UpdatedAt = time.Time{}
		create.Organization = db.Organization{}
		if err := tx.Unscoped().Omit(clause.Associations).Create(&create).Error; err != nil {
			return err
		}
		stored, err := readClusterForMutation(tx, desired.ClusterID)
		if err != nil {
			return err
		}
		if err := verifyExistingIdentity(stored, desired); err != nil {
			return err
		}
		changes, err := diffOwnedFields(kind, stored, desired)
		if err != nil {
			return err
		}
		if len(changes) != 0 {
			return ErrRowDrift
		}
		afterChecksum, err := clusterRecordChecksum(stored)
		if err != nil {
			return err
		}
		result = &MutationResult{Outcome: MutationOutcomeApplied, AfterChecksum: afterChecksum}
		return nil
	})
	if err != nil {
		return nil, fmt.Errorf("guarded create: %w", err)
	}
	return result, nil
}

func (s *GormStore) ApplyUpdate(
	ctx context.Context,
	kind ObjectKind,
	desired *db.Cluster,
	expectedDesiredChecksum string,
	expectedBeforeChecksum string,
	fields []string,
) (*MutationResult, error) {
	if err := s.validateMutationInput(kind, desired, expectedDesiredChecksum); err != nil {
		return nil, err
	}
	if !isSHA256(expectedBeforeChecksum) {
		return nil, errors.New("expected before checksum is invalid")
	}
	if err := validateMutationFields(kind, fields); err != nil {
		return nil, err
	}
	var result *MutationResult
	err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := verifyOrganizationForMutation(tx, desired); err != nil {
			return err
		}
		if err := verifyParentForMutation(tx, desired); err != nil {
			return err
		}
		existing, err := readClusterForMutation(tx, desired.ClusterID)
		if err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return ErrRowDrift
			}
			return err
		}
		if err := verifyExistingIdentity(existing, desired); err != nil {
			return err
		}
		beforeChecksum, err := clusterRecordChecksum(existing)
		if err != nil {
			return err
		}
		changes, err := diffOwnedFields(kind, existing, desired)
		if err != nil {
			return err
		}
		if len(changes) == 0 {
			result = &MutationResult{Outcome: MutationOutcomeNoop, BeforeChecksum: beforeChecksum, AfterChecksum: beforeChecksum}
			return nil
		}
		if beforeChecksum != expectedBeforeChecksum {
			return ErrRowDrift
		}
		actualFields := changeFieldNames(changes)
		if !slices.Equal(actualFields, fields) {
			return ErrRowDrift
		}
		updates, err := mutationUpdateMap(desired, fields)
		if err != nil {
			return err
		}
		operation := tx.Unscoped().Model(&db.Cluster{}).Where("cluster_id = ?", desired.ClusterID).Updates(updates)
		if operation.Error != nil {
			return operation.Error
		}
		if operation.RowsAffected != 1 {
			return ErrRowDrift
		}
		stored, err := readClusterForMutation(tx, desired.ClusterID)
		if err != nil {
			return err
		}
		remaining, err := diffOwnedFields(kind, stored, desired)
		if err != nil {
			return err
		}
		if len(remaining) != 0 {
			return ErrRowDrift
		}
		afterChecksum, err := clusterRecordChecksum(stored)
		if err != nil {
			return err
		}
		result = &MutationResult{
			Outcome:        MutationOutcomeApplied,
			BeforeChecksum: beforeChecksum,
			AfterChecksum:  afterChecksum,
		}
		return nil
	})
	if err != nil {
		return nil, fmt.Errorf("guarded update: %w", err)
	}
	return result, nil
}

func (s *GormStore) validateMutationInput(kind ObjectKind, desired *db.Cluster, expectedDesiredChecksum string) error {
	if s == nil || s.db == nil {
		return errors.New("gorm database is required")
	}
	if desired == nil || desired.OrgID == 0 || (desired.TenantID == 0 && kind != ObjectKindResourcePool) || desired.ClusterType != "cloud" ||
		desired.CreatedAt.IsZero() || desired.Status == "" {
		return ErrMutationConflict
	}
	if err := validateDesiredLifecycle(desired); err != nil {
		return ErrMutationConflict
	}
	if _, err := parsePositiveID(desired.ClusterID); err != nil {
		return ErrMutationConflict
	}
	if desired.ParentID != "" {
		if _, err := parsePositiveID(desired.ParentID); err != nil {
			return ErrMutationConflict
		}
	}
	checksum, err := desiredClusterChecksum(kind, desired)
	if err != nil {
		return err
	}
	if !isSHA256(expectedDesiredChecksum) || checksum != expectedDesiredChecksum {
		return ErrRowDrift
	}
	return nil
}

func verifyOrganizationForMutation(tx *gorm.DB, desired *db.Cluster) error {
	var organization db.Organization
	err := tx.Unscoped().
		Clauses(clause.Locking{Strength: "SHARE"}).
		Where("id = ? AND dbaas_org_id = ? AND org_type = ? AND deleted_at IS NULL", desired.OrgID, desired.TenantID, string(view.CloudDataSource)).
		Take(&organization).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return ErrMutationConflict
	}
	return err
}

func verifyParentForMutation(tx *gorm.DB, desired *db.Cluster) error {
	if desired.ParentID == "" || terminalStatus(desired.Status) {
		return nil
	}
	var parent db.Cluster
	err := tx.Unscoped().
		Clauses(clause.Locking{Strength: "SHARE"}).
		Where("cluster_id = ?", desired.ParentID).
		Take(&parent).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return ErrParentUnavailable
	}
	if err != nil {
		return err
	}
	if parent.TenantID != desired.TenantID || parent.OrgID != desired.OrgID || parent.DeletedAt.Valid || terminalStatus(parent.Status) {
		return ErrParentUnavailable
	}
	return nil
}

func readClusterForMutation(tx *gorm.DB, clusterID string) (*db.Cluster, error) {
	var cluster db.Cluster
	err := tx.Unscoped().
		Clauses(clause.Locking{Strength: "UPDATE"}).
		Where("cluster_id = ?", clusterID).
		Take(&cluster).Error
	if err != nil {
		return nil, err
	}
	return &cluster, nil
}

func verifyExistingIdentity(existing, desired *db.Cluster) error {
	if existing == nil || desired == nil || existing.ClusterID != desired.ClusterID ||
		existing.TenantID != desired.TenantID || existing.OrgID != desired.OrgID ||
		existing.ClusterType != desired.ClusterType ||
		(hasTerminalMarker(existing) && !terminalStatus(desired.Status)) ||
		!immutableSourceFieldsMatch(existing, desired) {
		return ErrMutationConflict
	}
	return nil
}

func validateMutationFields(kind ObjectKind, fields []string) error {
	if len(fields) == 0 || !sortedUniqueStrings(fields) {
		return errors.New("mutation fields must be sorted and unique")
	}
	allowed, err := ownedFields(kind)
	if err != nil {
		return err
	}
	for _, field := range fields {
		if !slices.Contains(allowed, field) {
			return fmt.Errorf("field %q is not owned by recovery", field)
		}
	}
	return nil
}

func mutationUpdateMap(desired *db.Cluster, fields []string) (map[string]any, error) {
	updates := make(map[string]any, len(fields))
	for _, field := range fields {
		if field == "deleted_at" {
			updates[field] = desired.DeletedAt
			continue
		}
		value, err := clusterFieldValue(desired, field)
		if err != nil {
			return nil, err
		}
		updates[field] = value
	}
	return updates, nil
}