package sharedpool

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

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

type sharedPoolRepoImpl struct {
	provider            db.Provider
	dataplaneRepo       repo.DataplaneRepo
	backupGcSettingRepo repo.BackupGcSettingRepo
	dataplaneManager    dataplane.Manager
	bucketClient        dplanemgrv1.BucketServiceClient
}

func NewSharedPoolRepo(provider db.Provider, dataplaneRepo repo.DataplaneRepo, backupGcSettingRepo repo.BackupGcSettingRepo, dataplaneManager dataplane.Manager, bucketClient dplanemgrv1.BucketServiceClient) repo.SharedPoolRepo {
	return &sharedPoolRepoImpl{
		provider:            provider,
		dataplaneRepo:       dataplaneRepo,
		backupGcSettingRepo: backupGcSettingRepo,
		dataplaneManager:    dataplaneManager,
		bucketClient:        bucketClient,
	}
}

func (r *sharedPoolRepoImpl) Save(ctx context.Context, sharedpool *entity.SharedPool) error {
	if sharedpool == nil {
		return status.Errorf(codes.InvalidArgument, "SharedPool is nil")
	}

	if sharedpool.GetProvider() == nil {
		return status.Errorf(codes.InvalidArgument, "provider is nil")
	}

	sharedpoolModel, objectStores, err := r.cvtSharedPoolToModel(sharedpool)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to convert SharedPool to model: %v", err)
	}

	wdb := r.provider.WriteDB(ctx)

	// not found, then set the max template version
	// if found, then set the existing template version
	// Check if the SharedPool already exists
	var existingSharedPool schema.SharedPool
	if err := wdb.Where("id = ?", sharedpoolModel.ID).First(&existingSharedPool).Error; err != nil {
		if !errors.Is(err, gorm.ErrRecordNotFound) {
			return status.Errorf(codes.Internal, "failed to check existing SharedPool: %v", err)
		}

		// If not found, create a new SharedPool
		// get the max template version with the all shared pools
		var maxTemplateVersion sql.NullInt64
		if err := wdb.Model(&schema.SharedPool{}).Select("MAX(template_version)").Scan(&maxTemplateVersion).Error; err != nil {
			return status.Errorf(codes.Internal, "failed to get max template version: %v", err)
		}
		if maxTemplateVersion.Valid {
			sharedpoolModel.TemplateVersion = maxTemplateVersion.Int64
		} else {
			// If no existing shared pool, set to 1
			sharedpoolModel.TemplateVersion = 1
		}
	} else {
		// If found, update the existing SharedPool
		sharedpoolModel.TemplateVersion = existingSharedPool.TemplateVersion
		// Preserve the original created_at value to avoid zero value issues
		sharedpoolModel.CreatedAt = existingSharedPool.CreatedAt
	}

	// Pause/resume fields are exclusively owned by UpdatePauseResumeState. A stale
	// entity saved by an unrelated workflow must never overwrite them.
	if err := wdb.Model(&schema.SharedPool{}).
		Omit("pause_resume_state", "suspend_compute_desired").
		Where("id = ?", sharedpoolModel.ID).Save(sharedpoolModel).Error; err != nil {
		return status.Errorf(codes.Internal, "failed to save SharedPool: %v", err)
	}

	for _, store := range objectStores {
		// if already exists, skip
		var existingStore schema.SharedPoolObjectStore
		if err := wdb.Where("shared_pool_id = ? AND store_type = ?", sharedpoolModel.ID, store.StoreType).First(&existingStore).Error; err != nil {
			if !errors.Is(err, gorm.ErrRecordNotFound) {
				return status.Errorf(codes.Internal, "failed to check existing ObjectStore: %v", err)
			}
			// If not found, create a new ObjectStore
			if err := wdb.Create(store).Error; err != nil {
				return status.Errorf(codes.Internal, "failed to create SharedPool ObjectStore: %v", err)
			}
		}
	}

	return nil
}

func (r *sharedPoolRepoImpl) UpdatePauseResumeState(ctx context.Context, id string, state entity.PauseResumeState, suspendComputeDesired bool) error {
	var dbState any
	if state != entity.PauseResumeStateNone {
		dbState = string(state)
	}
	wdb := r.provider.WriteDB(ctx)
	result := wdb.Model(&schema.SharedPool{}).Where("id = ?", id).Updates(map[string]any{
		"pause_resume_state":      dbState,
		"suspend_compute_desired": suspendComputeDesired,
	})
	if result.Error != nil {
		return fmt.Errorf("update pause/resume state for shared pool %s: %w", id, result.Error)
	}
	if result.RowsAffected == 0 {
		// MySQL reports zero changed rows when an idempotent retry writes the
		// values already stored. Distinguish that from a missing SharedPool.
		var count int64
		if err := wdb.Model(&schema.SharedPool{}).Where("id = ?", id).Count(&count).Error; err != nil {
			return fmt.Errorf("check shared pool %s after pause/resume state update: %w", id, err)
		}
		if count > 0 {
			return nil
		}
		return status.Errorf(codes.NotFound, "SharedPool with id %s not found", id)
	}
	return nil
}

func (r *sharedPoolRepoImpl) Load(ctx context.Context, id string, deleted bool) (*entity.SharedPool, error) {
	if id == "" {
		return nil, status.Errorf(codes.InvalidArgument, "shared pool id cannot be empty")
	}

	rdb := r.provider.ReadDB(ctx)

	var sharedpoolModel schema.SharedPool
	if deleted {
		rdb = rdb.Unscoped() // Load deleted records
		// When loading deleted records, also need to preload deleted ObjectStores
		rdb = rdb.Preload("ObjectStores", func(db *gorm.DB) *gorm.DB {
			return db.Unscoped()
		})
	} else {
		rdb = rdb.Preload("ObjectStores")
	}

	if err := rdb.Where("id = ?", id).First(&sharedpoolModel).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, status.Errorf(codes.NotFound, "SharedPool with id %s not found", id)
		}
		return nil, status.Errorf(codes.Internal, "failed to load SharedPool: %v", err)
	}

	en, err := r.cvtSharedPoolToEn(ctx, &sharedpoolModel)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to convert SharedPool to entity: %v", err)
	}

	return en, nil
}

func (r *sharedPoolRepoImpl) LoadDataplaneName(ctx context.Context, id string, deleted bool) (string, error) {
	if id == "" {
		return "", status.Errorf(codes.InvalidArgument, "shared pool id cannot be empty")
	}

	rdb := r.provider.ReadDB(ctx)
	if deleted {
		rdb = rdb.Unscoped()
	}

	var sharedpoolModel schema.SharedPool
	if err := rdb.Select("dataplane_name").Where("id = ?", id).First(&sharedpoolModel).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return "", status.Errorf(codes.NotFound, "SharedPool with id %s not found", id)
		}
		return "", status.Errorf(codes.Internal, "failed to load SharedPool dataplane name: %v", err)
	}

	return sharedpoolModel.DataplaneName, nil
}

func (r *sharedPoolRepoImpl) Delete(ctx context.Context, id string) error {
	if id == "" {
		return status.Errorf(codes.InvalidArgument, "SharedPool id cannot be empty")
	}

	wdb := r.provider.WriteDB(ctx)

	if err := wdb.Transaction(func(tx *gorm.DB) error {
		if err := tx.Where("id = ?", id).Delete(&schema.SharedPool{}).Error; err != nil {
			if !errors.Is(err, gorm.ErrRecordNotFound) {
				return status.Errorf(codes.Internal, "failed to delete SharedPool: %v", err)
			}
		}
		return nil
	}); err != nil {
		return err
	}

	return nil
}

func (r *sharedPoolRepoImpl) List(ctx context.Context, filter *repo.SharedPoolFilter) ([]*entity.SharedPool, error) {
	rdb := r.provider.ReadDB(ctx)

	var sharedPools []*schema.SharedPool

	// Handle nil filter as empty filter (return all)
	if filter == nil {
		filter = &repo.SharedPoolFilter{}
	}

	if filter.Deleted {
		rdb = rdb.Unscoped() // Load deleted records
	}

	if len(filter.Ids) > 0 {
		rdb = rdb.Where("id IN ?", filter.Ids)
	}

	if len(filter.DataplaneNames) > 0 {
		rdb = rdb.Where("dataplane_name IN ?", filter.DataplaneNames)
	}

	// When loading deleted records, also need to preload deleted ObjectStores
	if filter.Deleted {
		rdb = rdb.Preload("ObjectStores", func(db *gorm.DB) *gorm.DB {
			return db.Unscoped()
		})
	} else {
		rdb = rdb.Preload("ObjectStores")
	}

	if err := rdb.Find(&sharedPools).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, status.Errorf(codes.NotFound, "no SharedPools found")
		}
		return nil, status.Errorf(codes.Internal, "failed to list SharedPools: %v", err)
	}

	var result []*entity.SharedPool
	for _, spModel := range sharedPools {
		en, err := r.cvtSharedPoolToEn(ctx, spModel)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "failed to convert SharedPool to entity: %v", err)
		}

		result = append(result, en)
	}

	return result, nil
}