package impl

import (
	"context"

	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

type bucketRepoImpl struct {
	dbProvider db.Provider
}

var _ repo.BucketRepo = (*bucketRepoImpl)(nil)

func NewBucketRepo(dbProvider db.Provider) repo.BucketRepo {
	return &bucketRepoImpl{dbProvider: dbProvider}
}

func (c *bucketRepoImpl) Load(ctx context.Context, id uint64) (*entity.Bucket, bizerr.BizError) {
	rdb := c.dbProvider.ReadDB(ctx)
	bucket := &entity.Bucket{}
	err := entity.NewBucketQuerySet(rdb).
		IDEq(entity.ID(id)).
		One(bucket)
	if err != nil {
		return nil, bizerr.WrapGormError(err, config.ErrMsgBucketNotFound, id)
	}
	return bucket, nil
}

func (c *bucketRepoImpl) Save(ctx context.Context, bucket *entity.Bucket) bizerr.BizError {
	wdb := c.dbProvider.WriteDB(ctx)
	if err := entity.SaveWithVersioning(wdb, bucket); err != nil {
		return err.Chain("save bucket, id=%d", bucket.ID)
	}
	return nil
}

func (c *bucketRepoImpl) ListByFilter(ctx context.Context, filter repo.BucketFilter) (
	[]*entity.Bucket, bizerr.BizError,
) {
	rdb := c.dbProvider.ReadDB(ctx)
	tx := buildTxBucketFilter(rdb, filter)
	var buckets []*entity.Bucket
	if err := tx.Model(&entity.Bucket{}).Find(&buckets).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "list bucket by filter is %v", filter)
	}
	return buckets, nil
}

func (c *bucketRepoImpl) DeleteByFilter(ctx context.Context, filter repo.BucketFilter) (int64, bizerr.BizError) {
	tx := buildTxBucketFilter(c.dbProvider.WriteDB(ctx), filter)
	if err := tx.Delete(&entity.Bucket{}).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "delete bucket by filter is %v", filter)
	}
	return tx.RowsAffected, nil
}

func buildTxBucketFilter(tx *gorm.DB, filter repo.BucketFilter) *gorm.DB {
	if filter.ID != nil {
		tx = tx.Where("id = ?", filter.ID)
	}
	if filter.ProjectID != nil {
		tx = tx.Where("project_id = ?", filter.ProjectID)
	}
	if filter.TenantID != nil {
		tx = tx.Where("tenant_id = ?", filter.TenantID)
	}
	if filter.ClusterID != nil {
		tx = tx.Where("cluster_id = ?", filter.ClusterID)
	}
	// note there is no index for region,but it's ok because there are limited regions
	if filter.Region != nil {
		tx = tx.Where("regions = ?", filter.Region)
	}
	if filter.RegionPrefix != nil {
		tx = tx.Where("regions like ?", *filter.RegionPrefix+"%")
	}
	if filter.Provider != nil {
		tx = tx.Where("provider = ?", filter.Provider)
	}
	if filter.Scope != nil {
		tx = tx.Where("scope = ?", filter.Scope)
	}
	if filter.Extension != nil {
		tx = tx.Where("extension like ?", "%"+*filter.Extension+"%")
	}
	if len(filter.Statuses) > 0 {
		tx = tx.Where("status in (?)", filter.Statuses)
	}
	if len(filter.StatusesNotIn) > 0 {
		tx = tx.Where("status not in (?)", filter.StatusesNotIn)
	}
	if filter.Name != nil {
		tx = tx.Where("name = ?", filter.Name)
	}
	if filter.DeletedTime != nil {
		tx = tx.Where("deleted_time = ?", filter.DeletedTime)
	}
	if filter.CloudAccountID != nil {
		tx = tx.Where("cloud_account_id = ?", filter.CloudAccountID)
	}
	return tx
}

type awsBucketConfigurationRepoImpl struct {
	dbProvider db.Provider
}

var _ repo.AWSBucketConfigurationRepo = (*awsBucketConfigurationRepoImpl)(nil)

func NewAWSBucketConfigurationRepo(dbProvider db.Provider) repo.AWSBucketConfigurationRepo {
	return &awsBucketConfigurationRepoImpl{dbProvider: dbProvider}
}

func (c *awsBucketConfigurationRepoImpl) Lock(
	ctx context.Context, id uint64,
) bizerr.BizError {
	tx := c.dbProvider.WriteDB(ctx)
	configuration := &entity.AWSBucketConfiguration{}
	if err := tx.Where("id = ?", id).
		Clauses(clause.Locking{Strength: "UPDATE"}).
		First(configuration).Error; err != nil {
		return bizerr.WrapGormError(err, "lock AWSBucketConfiguration(%d) occur error", id)
	}
	return nil
}

func (c *awsBucketConfigurationRepoImpl) Load(
	ctx context.Context, id uint64,
) (*entity.AWSBucketConfiguration, bizerr.BizError) {
	rdb := c.dbProvider.ReadDB(ctx)
	configuration := &entity.AWSBucketConfiguration{}
	err := entity.NewAWSBucketConfigurationQuerySet(rdb).
		IDEq(entity.ID(id)).
		One(configuration)
	if err != nil {
		return nil, bizerr.WrapGormError(err, config.ErrMsgAWSBucketConfigurationNotFound, id)
	}
	return configuration, nil
}

func (c *awsBucketConfigurationRepoImpl) Save(
	ctx context.Context, configuration *entity.AWSBucketConfiguration,
) bizerr.BizError {
	wdb := c.dbProvider.WriteDB(ctx)
	if err := entity.SaveWithVersioning(wdb, configuration); err != nil {
		return err.Chain("save configuration, id=%d", configuration.ID)
	}
	return nil
}

func (c *awsBucketConfigurationRepoImpl) List(
	ctx context.Context, ops ...repo.AWSBucketConfigurationRepoOp,
) ([]entity.AWSBucketConfiguration, bizerr.BizError) {
	rdb := c.dbProvider.ReadDB(ctx)

	qs := entity.NewAWSBucketConfigurationQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}
	var replications []entity.AWSBucketConfiguration
	if err := qs.All(&replications); err != nil {
		return nil, bizerr.WrapGormError(err, "list AWSBucketConfiguration occur error")
	}
	return replications, nil
}

func (c *awsBucketConfigurationRepoImpl) Get(
	ctx context.Context, ops ...repo.AWSBucketConfigurationRepoOp,
) (*entity.AWSBucketConfiguration, bizerr.BizError) {
	rdb := c.dbProvider.ReadDB(ctx)

	qs := entity.NewAWSBucketConfigurationQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}
	configuration := &entity.AWSBucketConfiguration{}
	if err := qs.One(configuration); err != nil {
		return nil, bizerr.WrapGormError(err, "get AWSBucketConfiguration occur error")
	}
	return configuration, nil
}

type awsBucketReplicationRepoImpl struct {
	dbProvider db.Provider
}

var _ repo.AWSBucketReplicationRepo = (*awsBucketReplicationRepoImpl)(nil)

func NewAWSBucketReplicationRepo(dbProvider db.Provider) repo.AWSBucketReplicationRepo {
	return &awsBucketReplicationRepoImpl{dbProvider: dbProvider}
}

func (c *awsBucketReplicationRepoImpl) Load(
	ctx context.Context, id uint64,
) (*entity.AWSBucketReplication, bizerr.BizError) {
	rdb := c.dbProvider.ReadDB(ctx)
	replication := &entity.AWSBucketReplication{}
	err := entity.NewAWSBucketReplicationQuerySet(rdb).
		IDEq(entity.ID(id)).
		One(replication)
	if err != nil {
		return nil, bizerr.WrapGormError(err, config.ErrMsgAWSBucketReplicationNotFound, id)
	}
	return replication, nil
}

func (c *awsBucketReplicationRepoImpl) Save(
	ctx context.Context, replication *entity.AWSBucketReplication,
) bizerr.BizError {
	wdb := c.dbProvider.WriteDB(ctx)
	if err := entity.SaveWithVersioning(wdb, replication); err != nil {
		return err.Chain("save replication, id=%d", replication.ID)
	}
	return nil
}

func (c *awsBucketReplicationRepoImpl) List(
	ctx context.Context, ops ...repo.AWSBucketReplicationRepoOp,
) ([]entity.AWSBucketReplication, bizerr.BizError) {
	rdb := c.dbProvider.ReadDB(ctx)

	qs := entity.NewAWSBucketReplicationQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}
	var replications []entity.AWSBucketReplication
	if err := qs.All(&replications); err != nil {
		return nil, bizerr.WrapGormError(err, "list replications occur error")
	}
	return replications, nil
}

func (c *awsBucketReplicationRepoImpl) Get(
	ctx context.Context, ops ...repo.AWSBucketReplicationRepoOp,
) (*entity.AWSBucketReplication, bizerr.BizError) {
	rdb := c.dbProvider.ReadDB(ctx)

	qs := entity.NewAWSBucketReplicationQuerySet(rdb)
	for _, op := range ops {
		qs = op(qs)
	}
	replication := &entity.AWSBucketReplication{}
	if err := qs.One(replication); err != nil {
		return nil, bizerr.WrapGormError(err, "get replication occur error")
	}
	return replication, nil
}

type gcpBucketConfigurationRepoImpl struct {
	dbProvider db.Provider
}

var _ repo.GCPBucketConfigurationRepo = (*gcpBucketConfigurationRepoImpl)(nil)

func NewGCPBucketConfigurationRepo(dbProvider db.Provider) repo.GCPBucketConfigurationRepo {
	return &gcpBucketConfigurationRepoImpl{dbProvider: dbProvider}
}

func (c *gcpBucketConfigurationRepoImpl) Lock(
	ctx context.Context, id uint64,
) bizerr.BizError {
	tx := c.dbProvider.WriteDB(ctx)
	configuration := &entity.GCPBucketConfiguration{}
	if err := tx.Where("id = ?", id).
		Clauses(clause.Locking{Strength: "UPDATE"}).
		First(configuration).Error; err != nil {
		return bizerr.WrapGormError(err, "lock GCPBucketConfiguration(%d) occur error", id)
	}
	return nil
}

func (m *gcpBucketConfigurationRepoImpl) Load(
	ctx context.Context, id uint64,
) (*entity.GCPBucketConfiguration, bizerr.BizError) {
	rdb := m.dbProvider.ReadDB(ctx)
	bucketConf := new(entity.GCPBucketConfiguration)
	err := entity.NewGCPBucketConfigurationQuerySet(rdb).IDEq(entity.ID(id)).One(bucketConf)
	if err != nil {
		if err == gorm.ErrRecordNotFound {
			return nil, bizerr.WrapGormError(err, config.ErrMsgBucketNotFound, id)
		}
		return nil, bizerr.WrapGormError(err, "load gcp bucket configuration error")
	}
	return bucketConf, nil
}

func (m *gcpBucketConfigurationRepoImpl) Save(
	ctx context.Context, configuration *entity.GCPBucketConfiguration,
) bizerr.BizError {
	wdb := m.dbProvider.WriteDB(ctx)
	if err := entity.SaveWithVersioning(wdb, configuration); err != nil {
		return err.Chain("save gcp bucket configuration %d error", configuration.ID)
	}
	return nil
}

func (m *gcpBucketConfigurationRepoImpl) List(
	ctx context.Context, filter *repo.GCPBucketConfigurationFilter,
) ([]*entity.GCPBucketConfiguration, bizerr.BizError) {
	rdb := m.dbProvider.ReadDB(ctx)
	rdb = buildTxByGCPBucketConfigurationFilter(rdb, filter)

	var bucketConfs []*entity.GCPBucketConfiguration
	if result := rdb.Find(&bucketConfs); result.Error != nil {
		return nil, bizerr.WrapGormError(result.Error, "list gcp bucket configurations error")
	}
	return bucketConfs, nil
}

func (m *gcpBucketConfigurationRepoImpl) Get(
	ctx context.Context, filter *repo.GCPBucketConfigurationFilter,
) (*entity.GCPBucketConfiguration, bizerr.BizError) {
	rdb := m.dbProvider.ReadDB(ctx)
	rdb = buildTxByGCPBucketConfigurationFilter(rdb, filter)

	bucketConf := new(entity.GCPBucketConfiguration)
	if result := rdb.First(bucketConf); result.Error != nil {
		return nil, bizerr.WrapGormError(result.Error, "get gcp bucket configuration error")
	}
	return bucketConf, nil
}

func buildTxByGCPBucketConfigurationFilter(tx *gorm.DB, filter *repo.GCPBucketConfigurationFilter) *gorm.DB {
	if filter.BucketID != nil {
		tx = tx.Where("bucket_id = ?", *filter.BucketID)
	}
	return tx
}