package nodegroup

import (
	"context"
	"errors"
	"time"

	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/gormutil"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"gorm.io/gorm"
)

type lockRepoImpl struct {
	provider db.Provider
}

// NewDataplaneNodeInstanceGroupLockRepo creates a new DataplaneNodeInstanceGroupLockRepo.
func NewDataplaneNodeInstanceGroupLockRepo(provider db.Provider) repo.DataplaneNodeInstanceGroupLockRepo {
	return &lockRepoImpl{provider: provider}
}

func (r *lockRepoImpl) TryAcquire(ctx context.Context, dataplaneName, nodeSpec, primaryInstance, ownerID string, leaseDuration time.Duration) (*entity.DataplaneNodeInstanceGroupLock, error) {
	if dataplaneName == "" || nodeSpec == "" || primaryInstance == "" || ownerID == "" {
		return nil, status.Errorf(codes.InvalidArgument, "dataplaneName, nodeSpec, primaryInstance and ownerID are required")
	}
	nodeSpec = normalizeNodeSpec(nodeSpec)

	wdb := r.provider.WriteDB(ctx)
	now := time.Now()
	expiresAt := now.Add(leaseDuration)

	// Try to load existing lock
	var existing schema.DataplaneNodeInstanceGroupLock
	err := wdb.Where("dataplane_name = ? AND node_spec = ? AND primary_instance = ?", dataplaneName, nodeSpec, primaryInstance).First(&existing).Error

	if errors.Is(err, gorm.ErrRecordNotFound) {
		// No lock exists, create new one
		newLock := &schema.DataplaneNodeInstanceGroupLock{
			DataplaneName:   dataplaneName,
			NodeSpec:        nodeSpec,
			PrimaryInstance: primaryInstance,
			OwnerID:         ownerID,
			FencingToken:    1, // Start at 1
			ExpiresAt:       expiresAt,
			CreatedAt:       now,
			UpdatedAt:       now,
		}

		if createErr := wdb.Create(newLock).Error; createErr != nil {
			// Handle race condition - another process might have created it
			if errors.Is(createErr, gorm.ErrDuplicatedKey) || gormutil.IsDuplicatedError(createErr) {
				// Retry with existing lock logic
				return r.tryAcquireExisting(ctx, dataplaneName, nodeSpec, primaryInstance, ownerID, leaseDuration)
			}
			return nil, status.Errorf(codes.Internal, "failed to create lock: %v", createErr)
		}

		return &entity.DataplaneNodeInstanceGroupLock{DataplaneNodeInstanceGroupLock: *newLock}, nil
	}

	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to load lock: %v", err)
	}

	// Lock exists, check if we can take it
	return r.tryAcquireExistingLock(ctx, &existing, ownerID, leaseDuration)
}

// tryAcquireExisting is called when we know a lock exists (race condition path)
func (r *lockRepoImpl) tryAcquireExisting(ctx context.Context, dataplaneName, nodeSpec, primaryInstance, ownerID string, leaseDuration time.Duration) (*entity.DataplaneNodeInstanceGroupLock, error) {
	wdb := r.provider.WriteDB(ctx)

	var existing schema.DataplaneNodeInstanceGroupLock
	if err := wdb.Where("dataplane_name = ? AND node_spec = ? AND primary_instance = ?", dataplaneName, nodeSpec, primaryInstance).First(&existing).Error; err != nil {
		return nil, status.Errorf(codes.Internal, "failed to load existing lock: %v", err)
	}

	return r.tryAcquireExistingLock(ctx, &existing, ownerID, leaseDuration)
}

// tryAcquireExistingLock attempts to acquire or renew an existing lock
func (r *lockRepoImpl) tryAcquireExistingLock(ctx context.Context, existing *schema.DataplaneNodeInstanceGroupLock, ownerID string, leaseDuration time.Duration) (*entity.DataplaneNodeInstanceGroupLock, error) {
	wdb := r.provider.WriteDB(ctx)
	now := time.Now()
	expiresAt := now.Add(leaseDuration)

	// Check if we already own the lock
	if existing.OwnerID == ownerID {
		// We own it, renew the lease
		newToken := existing.FencingToken + 1
		result := wdb.Model(&schema.DataplaneNodeInstanceGroupLock{}).
			Where("dataplane_name = ? AND node_spec = ? AND primary_instance = ? AND owner_id = ? AND fencing_token = ?",
				existing.DataplaneName, existing.NodeSpec, existing.PrimaryInstance, ownerID, existing.FencingToken).
			Updates(map[string]any{
				"fencing_token": newToken,
				"expires_at":    expiresAt,
				"updated_at":    now,
			})

		if result.Error != nil {
			return nil, status.Errorf(codes.Internal, "failed to renew lock: %v", result.Error)
		}

		if result.RowsAffected == 0 {
			// Lost the lock in a race
			return nil, status.Errorf(codes.FailedPrecondition, "lock was modified by another process")
		}

		existing.FencingToken = newToken
		existing.ExpiresAt = expiresAt
		existing.UpdatedAt = now
		return &entity.DataplaneNodeInstanceGroupLock{DataplaneNodeInstanceGroupLock: *existing}, nil
	}

	// Someone else owns the lock, check if it's expired
	if now.Before(existing.ExpiresAt) {
		// Lock is still valid, cannot acquire
		return nil, nil //nolint:nilnil // Interface contract: returns nil without error when lock cannot be acquired
	}

	// Lock is expired, try to take over with optimistic locking
	newToken := existing.FencingToken + 1
	result := wdb.Model(&schema.DataplaneNodeInstanceGroupLock{}).
		Where("dataplane_name = ? AND node_spec = ? AND primary_instance = ? AND fencing_token = ? AND expires_at < ?",
			existing.DataplaneName, existing.NodeSpec, existing.PrimaryInstance, existing.FencingToken, now).
		Updates(map[string]any{
			"owner_id":      ownerID,
			"fencing_token": newToken,
			"expires_at":    expiresAt,
			"updated_at":    now,
		})

	if result.Error != nil {
		return nil, status.Errorf(codes.Internal, "failed to acquire expired lock: %v", result.Error)
	}

	if result.RowsAffected == 0 {
		// Lost the race to another process
		return nil, nil //nolint:nilnil // Interface contract: returns nil without error when lock cannot be acquired
	}

	existing.OwnerID = ownerID
	existing.FencingToken = newToken
	existing.ExpiresAt = expiresAt
	existing.UpdatedAt = now
	return &entity.DataplaneNodeInstanceGroupLock{DataplaneNodeInstanceGroupLock: *existing}, nil
}

func (r *lockRepoImpl) Release(ctx context.Context, dataplaneName, nodeSpec, primaryInstance, ownerID string) error {
	if dataplaneName == "" || nodeSpec == "" || primaryInstance == "" || ownerID == "" {
		return status.Errorf(codes.InvalidArgument, "dataplaneName, nodeSpec, primaryInstance and ownerID are required")
	}
	nodeSpec = normalizeNodeSpec(nodeSpec)

	wdb := r.provider.WriteDB(ctx)

	// Hard delete: NodeInstanceGroupLock has no DeletedAt field
	result := wdb.Where("dataplane_name = ? AND node_spec = ? AND primary_instance = ? AND owner_id = ?", dataplaneName, nodeSpec, primaryInstance, ownerID).
		Delete(&schema.DataplaneNodeInstanceGroupLock{})

	if result.Error != nil {
		return status.Errorf(codes.Internal, "failed to release lock: %v", result.Error)
	}

	if result.RowsAffected == 0 {
		return status.Errorf(codes.FailedPrecondition, "lock not held by owner %s", ownerID)
	}

	return nil
}

func (r *lockRepoImpl) Load(ctx context.Context, dataplaneName, nodeSpec, primaryInstance string) (*entity.DataplaneNodeInstanceGroupLock, error) {
	if dataplaneName == "" || nodeSpec == "" || primaryInstance == "" {
		return nil, status.Errorf(codes.InvalidArgument, "dataplaneName, nodeSpec and primaryInstance are required")
	}
	nodeSpec = normalizeNodeSpec(nodeSpec)

	rdb := r.provider.ReadDB(ctx)

	var model schema.DataplaneNodeInstanceGroupLock
	if err := rdb.Where("dataplane_name = ? AND node_spec = ? AND primary_instance = ?", dataplaneName, nodeSpec, primaryInstance).First(&model).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, nil //nolint:nilnil // Interface contract: returns nil without error when not found
		}
		return nil, status.Errorf(codes.Internal, "failed to load lock: %v", err)
	}

	return &entity.DataplaneNodeInstanceGroupLock{DataplaneNodeInstanceGroupLock: model}, nil
}

// Hard delete all locks for this dataplane across every nodeSpec and primaryInstance.
// All callers expect hard-delete semantics; NodeInstanceGroupLock has no DeletedAt field.
func (r *lockRepoImpl) ForceRelease(ctx context.Context, dataplaneName string) error {
	if dataplaneName == "" {
		return status.Errorf(codes.InvalidArgument, "dataplaneName is required")
	}

	wdb := r.provider.WriteDB(ctx)

	// Hard delete all locks for this dataplane across every nodeSpec.
	if err := wdb.Where("dataplane_name = ?", dataplaneName).Delete(&schema.DataplaneNodeInstanceGroupLock{}).Error; err != nil {
		return status.Errorf(codes.Internal, "failed to force release locks: %v", err)
	}

	return nil
}

func (r *lockRepoImpl) IsLockValid(ctx context.Context, dataplaneName, nodeSpec, primaryInstance string, fencingToken int64) (bool, error) {
	if dataplaneName == "" || nodeSpec == "" || primaryInstance == "" {
		return false, status.Errorf(codes.InvalidArgument, "dataplaneName, nodeSpec and primaryInstance are required")
	}
	nodeSpec = normalizeNodeSpec(nodeSpec)

	rdb := r.provider.ReadDB(ctx)

	var model schema.DataplaneNodeInstanceGroupLock
	if err := rdb.Where("dataplane_name = ? AND node_spec = ? AND primary_instance = ?", dataplaneName, nodeSpec, primaryInstance).First(&model).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			// No lock exists, invalid
			return false, nil
		}
		return false, status.Errorf(codes.Internal, "failed to check lock validity: %v", err)
	}

	// Lock is valid only if:
	// 1. The fencing token matches (no new lock was acquired)
	// 2. The lock hasn't expired
	isValid := model.FencingToken == fencingToken && time.Now().Before(model.ExpiresAt)
	return isValid, nil
}