package repo_imp

import (
	"context"
	"errors"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"gorm.io/gorm"
)

type privateLinkConnectionRepoImp struct {
	dbProvider db.TransProvider
}

func NewPrivateLinkConnectionRepoImp(dbProvider db.TransProvider) repo.PrivateLinkConnectionRepo {
	return &privateLinkConnectionRepoImp{dbProvider}
}

func (r *privateLinkConnectionRepoImp) ListByFilter(
	ctx context.Context,
	filter *entity.ListPrivateLinkConnectionFilter,
) ([]*entity.PrivateLinkConnection, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionRepo.ListByFilter")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	rdb = buildPrivateLinkConnectionFilter(filter, rdb)

	var items []*entity.PrivateLinkConnection
	if err := rdb.Find(&items).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list private link connections")
	}
	return items, nil
}

func (r *privateLinkConnectionRepoImp) CountByFilter(
	ctx context.Context,
	filter *entity.ListPrivateLinkConnectionFilter,
) (int64, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionRepo.CountByFilter")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	rdb = buildPrivateLinkConnectionCountFilter(filter, rdb)

	var count int64
	if err := rdb.Model(&entity.PrivateLinkConnection{}).Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "fail to count private link connections")
	}
	return count, nil
}

func (r *privateLinkConnectionRepoImp) Create(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionRepo.Create")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Create(plc).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to create private link connection")
	}
	return plc, nil
}

func (r *privateLinkConnectionRepoImp) GetById(
	ctx context.Context,
	clusterId, id string,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionRepo.GetById")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	plc := &entity.PrivateLinkConnection{}
	if err := rdb.Where("id = ? AND cluster_id = ? AND state != ?", id, clusterId, constant.PrivateLinkConnectionStateDeleted).
		First(plc).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get private link connection")
	}
	return plc, nil
}

func (r *privateLinkConnectionRepoImp) GetByID(
	ctx context.Context,
	id string,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionRepo.GetByID")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	plc := &entity.PrivateLinkConnection{}
	if err := rdb.Where("id = ? AND state != ?", id, constant.PrivateLinkConnectionStateDeleted).
		First(plc).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get private link connection")
	}
	return plc, nil
}

func (r *privateLinkConnectionRepoImp) Save(
	ctx context.Context,
	plc *entity.PrivateLinkConnection,
) (*entity.PrivateLinkConnection, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionRepo.Save")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Save(plc).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to save private link connection")
	}
	return plc, nil
}

func (r *privateLinkConnectionRepoImp) DeleteById(ctx context.Context, clusterId, id string) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionRepo.DeleteById")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	result := wdb.Model(&entity.PrivateLinkConnection{}).
		Where("id = ? AND cluster_id = ? AND state != ?", id, clusterId, constant.PrivateLinkConnectionStateDeleted).
		Update("state", constant.PrivateLinkConnectionStateDeleted)
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to delete private link connection")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "no private link connection found to delete")
	}
	return nil
}

func buildPrivateLinkConnectionFilter(filter *entity.ListPrivateLinkConnectionFilter, db *gorm.DB) *gorm.DB {
	db = buildPrivateLinkConnectionWhereFilter(filter, db)
	if filter.Offset != nil {
		db = db.Offset(int(*filter.Offset))
	}
	if filter.Limit != nil {
		db = db.Limit(int(*filter.Limit))
	}
	if filter.OrderBy != "" {
		db = db.Order(filter.OrderBy)
	}
	return db
}

func buildPrivateLinkConnectionCountFilter(filter *entity.ListPrivateLinkConnectionFilter, db *gorm.DB) *gorm.DB {
	return buildPrivateLinkConnectionWhereFilter(filter, db)
}

func buildPrivateLinkConnectionWhereFilter(filter *entity.ListPrivateLinkConnectionFilter, db *gorm.DB) *gorm.DB {
	db = db.Where("state != ?", constant.PrivateLinkConnectionStateDeleted)
	sharedScopeApplied := false
	if filter.IncludeSharedMSK && filter.ClusterID != "" && filter.TenantID != "" &&
		filter.Region != "" && filter.Kubernetes != "" {
		db = db.Where(
			"(cluster_id = ? OR (type = ? AND tenant_id = ? AND region = ? AND kubernetes = ?))",
			filter.ClusterID,
			constant.PrivateLinkConnectionAmazonMSKProvisioned,
			filter.TenantID,
			filter.Region,
			filter.Kubernetes,
		)
		sharedScopeApplied = true
	} else if filter.ClusterID != "" {
		db = db.Where("cluster_id = ?", filter.ClusterID)
	}
	if !sharedScopeApplied && filter.TenantID != "" {
		db = db.Where("tenant_id = ?", filter.TenantID)
	}
	if !sharedScopeApplied && filter.Region != "" {
		db = db.Where("region = ?", filter.Region)
	}
	if !sharedScopeApplied && filter.Kubernetes != "" {
		db = db.Where("kubernetes = ?", filter.Kubernetes)
	}
	if len(filter.States) > 0 {
		db = db.Where("state in ?", filter.States)
	}
	if len(filter.StatesNotIn) > 0 {
		db = db.Where("state not in ?", filter.StatesNotIn)
	}
	if filter.Type != "" {
		db = db.Where("type = ?", filter.Type)
	}
	if filter.AWSEndpointServiceName != "" {
		db = db.Where("aws_endpoint_service_name = ?", filter.AWSEndpointServiceName)
	}
	if filter.AlicloudEndpointServiceName != "" {
		db = db.Where("alicloud_endpoint_service_name = ?", filter.AlicloudEndpointServiceName)
	}
	if filter.AmazonMSKProvisionedArn != "" {
		db = db.Where("amazon_msk_provisioned_arn = ?", filter.AmazonMSKProvisionedArn)
	}
	return db
}

type privateLinkConnectionLinkRepoImp struct {
	dbProvider db.TransProvider
}

func NewPrivateLinkConnectionLinkRepoImp(dbProvider db.TransProvider) repo.PrivateLinkConnectionLinkRepo {
	return &privateLinkConnectionLinkRepoImp{dbProvider: dbProvider}
}

func (r *privateLinkConnectionLinkRepoImp) Create(
	ctx context.Context,
	link *entity.PrivateLinkConnectionLink,
) (*entity.PrivateLinkConnectionLink, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionLinkRepo.Create")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Create(link).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to create private link connection link")
	}
	return link, nil
}

func (r *privateLinkConnectionLinkRepoImp) ListByPrivateLinkConnectionID(
	ctx context.Context,
	clusterId, privateLinkConnectionID string,
) ([]*entity.PrivateLinkConnectionLink, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionLinkRepo.ListByPrivateLinkConnectionID")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	var items []*entity.PrivateLinkConnectionLink
	rdb = rdb.Where("private_link_connection_id = ?", privateLinkConnectionID)
	if clusterId != "" {
		rdb = rdb.Where("cluster_id = ?", clusterId)
	}
	if err := rdb.Find(&items).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list links by private link connection id")
	}
	return items, nil
}

func (r *privateLinkConnectionLinkRepoImp) Delete(
	ctx context.Context,
	clusterId, privateLinkConnectionID, bizID string,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionLinkRepo.Delete")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	result := wdb.Where(
		"cluster_id = ? AND private_link_connection_id = ? AND biz_id = ?",
		clusterId, privateLinkConnectionID, bizID,
	).Delete(&entity.PrivateLinkConnectionLink{})
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to delete private link connection link")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "no private link connection link found to delete")
	}
	return nil
}

func (r *privateLinkConnectionLinkRepoImp) DeleteByPrivateLinkConnectionID(
	ctx context.Context,
	clusterId, privateLinkConnectionID string,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "privateLinkConnectionLinkRepo.DeleteByPrivateLinkConnectionID")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	wdb = wdb.Where("private_link_connection_id = ?", privateLinkConnectionID)
	if clusterId != "" {
		wdb = wdb.Where("cluster_id = ?", clusterId)
	}
	if err := wdb.Delete(&entity.PrivateLinkConnectionLink{}).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to delete links by private link connection id")
	}
	return nil
}

type plcAttachDomainRepoImp struct {
	dbProvider db.TransProvider
}

func NewPLCAttachDomainRepoImp(dbProvider db.TransProvider) repo.PLCAttachDomainRepo {
	return &plcAttachDomainRepoImp{dbProvider: dbProvider}
}

func (r *plcAttachDomainRepoImp) Create(
	ctx context.Context,
	d *entity.PLCAttachDomain,
) (*entity.PLCAttachDomain, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "plcAttachDomainRepo.Create")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Create(d).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to create plc attach domain")
	}
	return d, nil
}

func (r *plcAttachDomainRepoImp) Save(
	ctx context.Context,
	d *entity.PLCAttachDomain,
) (*entity.PLCAttachDomain, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "plcAttachDomainRepo.Save")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Save(d).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to save plc attach domain")
	}
	return d, nil
}

func (r *plcAttachDomainRepoImp) Get(
	ctx context.Context,
	clusterID, plcID, attachDomainID string,
) (*entity.PLCAttachDomain, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "plcAttachDomainRepo.Get")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	plcd := &entity.PLCAttachDomain{}
	if err := rdb.Where("id = ? AND cluster_id = ? AND private_link_connection_id = ?", attachDomainID, clusterID, plcID).
		First(plcd).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get private link connection attach domain")
	}
	return plcd, nil
}

func (r *plcAttachDomainRepoImp) ListByFilter(
	ctx context.Context,
	filter entity.ListPLCAttachDomainsFilter,
) ([]*entity.PLCAttachDomain, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "plcAttachDomainRepo.ListByPLC")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	rdb = buildPrivateLinkAttachDomainFilter(filter, rdb)
	var items []*entity.PLCAttachDomain
	if err := rdb.Find(&items).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list plc attach domains")
	}
	return items, nil
}

func buildPrivateLinkAttachDomainFilter(filter entity.ListPLCAttachDomainsFilter, db *gorm.DB) *gorm.DB {
	if filter.ClusterID != "" {
		db = db.Where("cluster_id = ?", filter.ClusterID)
	}
	if filter.PrivateLinkConnectionID != "" {
		db = db.Where("private_link_connection_id = ?", filter.PrivateLinkConnectionID)
	}
	if len(filter.PrivateLinkConnectionIDs) > 0 {
		db = db.Where("private_link_connection_id IN ?", filter.PrivateLinkConnectionIDs)
	}
	return db
}

func (r *plcAttachDomainRepoImp) CountByPLC(
	ctx context.Context,
	clusterID, plcID string,
) (int64, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "plcAttachDomainRepo.CountByPLC")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	var count int64
	if err := rdb.Model(&entity.PLCAttachDomain{}).
		Where("cluster_id = ? AND private_link_connection_id = ?", clusterID, plcID).
		Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "fail to count plc attach domains")
	}
	return count, nil
}

func (r *plcAttachDomainRepoImp) DeleteByID(
	ctx context.Context,
	clusterID, plcID, attachDomainID string,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "plcAttachDomainRepo.DeleteByID")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	result := wdb.Where(
		"cluster_id = ? AND private_link_connection_id = ? AND id = ?",
		clusterID, plcID, attachDomainID,
	).Delete(&entity.PLCAttachDomain{})
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to delete plc attach domain")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "no plc attach domain found to delete")
	}
	return nil
}

func (r *plcAttachDomainRepoImp) CheckHostedZone(
	ctx context.Context,
	hostedZone, uniqueID string,
) (bool, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "plcAttachDomainRepo.CheckHostedZone")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	var item entity.PLCAttachDomain
	err := rdb.Where("hosted_zone = ? AND unique_id = ?", hostedZone, uniqueID).First(&item).Error
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return false, nil
		}
		return false, bizerr.WrapGormError(err, "fail to get plc attach domain by hosted zone and unique id")
	}
	return true, nil
}

type plcDomainRecordRepoImp struct {
	dbProvider db.TransProvider
}

func NewPLCDomainRecordRepoImp(dbProvider db.TransProvider) repo.PLCDomainRecordRepo {
	return &plcDomainRecordRepoImp{dbProvider: dbProvider}
}

func (r *plcDomainRecordRepoImp) BatchInsert(
	ctx context.Context,
	records []*entity.PLCDomainRecord,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "plcDomainRecordRepo.BatchInsert")
	defer func() { tracing.EndSpan(span, nil) }()

	if len(records) == 0 {
		return nil
	}
	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.CreateInBatches(records, 100).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to batch insert plc domain records")
	}
	return nil
}

func (r *plcDomainRecordRepoImp) ListByFilter(
	ctx context.Context,
	filter *entity.ListPLCDomainRecordsFilter,
) ([]*entity.PLCDomainRecord, bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "plcDomainRecordRepo.ListByID")
	defer func() { tracing.EndSpan(span, nil) }()

	rdb := r.dbProvider.ReadDB(ctx)
	rdb = buildPLCDomainRecordFilter(filter, rdb)
	var items []*entity.PLCDomainRecord
	if err := rdb.Find(&items).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list plc domain records by attach domain id")
	}
	return items, nil
}

func buildPLCDomainRecordFilter(filter *entity.ListPLCDomainRecordsFilter, db *gorm.DB) *gorm.DB {
	db = db.Where("state != ?", constant.PLCDomainRecordStateDeleted)
	if filter == nil {
		return db
	}
	if filter.ClusterID != "" {
		db = db.Where("cluster_id = ?", filter.ClusterID)
	}
	if filter.PrivateLinkConnectionID != "" {
		db = db.Where("private_link_connection_id = ?", filter.PrivateLinkConnectionID)
	}
	if filter.AttachDomainID != "" {
		db = db.Where("attach_domain_id = ?", filter.AttachDomainID)
	}
	if len(filter.PrivateLinkConnectionIDs) > 0 {
		db = db.Where("private_link_connection_id IN ?", filter.PrivateLinkConnectionIDs)
	}
	if len(filter.States) > 0 {
		db = db.Where("state IN ?", filter.States)
	}
	return db
}

func (r *plcDomainRecordRepoImp) ActiveByAttachDomainID(
	ctx context.Context,
	clusterID, plcID, attachDomainID string,
	awsRoute53CreateChangeID string,
	aliDNSRecordID int64,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "plcDomainRecordRepo.ActiveByAttachDomainID")
	defer func() { tracing.EndSpan(span, nil) }()

	updates := map[string]any{
		"state": constant.PLCDomainRecordStateActive,
	}
	if awsRoute53CreateChangeID != "" {
		updates["aws_route53_create_change_id"] = awsRoute53CreateChangeID
	}
	if aliDNSRecordID != 0 {
		updates["ali_dns_record_id"] = aliDNSRecordID
	}

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Model(&entity.PLCDomainRecord{}).
		Where("cluster_id = ? AND private_link_connection_id = ? AND attach_domain_id = ?",
			clusterID, plcID, attachDomainID,
		).Updates(updates).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to active plc domain records by attach domain id")
	}
	return nil
}

func (r *plcDomainRecordRepoImp) SoftDeleteByPLC(
	ctx context.Context,
	clusterID, plcID string,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "plcDomainRecordRepo.DeleteByPLC")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Model(&entity.PLCDomainRecord{}).
		Where(
			"cluster_id = ? AND private_link_connection_id = ?",
			clusterID, plcID,
		).Update("state", constant.PLCDomainRecordStateDeleted).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to delete plc domain records by plc")
	}
	return nil
}

func (r *plcDomainRecordRepoImp) SoftDeleteByID(
	ctx context.Context,
	clusterID, plcID, attachDomainID string,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "plcDomainRecordRepo.DeleteByID")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Model(&entity.PLCDomainRecord{}).
		Where("cluster_id = ? AND private_link_connection_id = ? AND attach_domain_id = ?",
			clusterID, plcID, attachDomainID,
		).Update("state", constant.PLCDomainRecordStateDeleted).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to delete plc domain records by attach domain id")
	}
	return nil
}

func (r *plcDomainRecordRepoImp) UpdateStateByAttachDomainID(
	ctx context.Context,
	clusterID, plcID, attachDomainID string,
	state constant.PLCDomainRecordState,
	message string,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "plcDomainRecordRepo.UpdateStateByAttachDomainID")
	defer func() { tracing.EndSpan(span, nil) }()

	updates := map[string]any{
		"state": state,
	}
	if message != "" {
		updates["message"] = message
	}

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Model(&entity.PLCDomainRecord{}).
		Where("cluster_id = ? AND private_link_connection_id = ? AND attach_domain_id = ?",
			clusterID, plcID, attachDomainID,
		).Updates(updates).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to update plc domain records state by attach domain id")
	}
	return nil
}

func (r *plcDomainRecordRepoImp) UpdateByID(
	ctx context.Context,
	clusterID, plcID, attachDomainID string,
	awsRoute53DeleteChangeID string,
) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "plcDomainRecordRepo.DeleteByID")
	defer func() { tracing.EndSpan(span, nil) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Model(&entity.PLCDomainRecord{}).
		Where("cluster_id = ? AND private_link_connection_id = ? AND attach_domain_id = ?",
			clusterID, plcID, attachDomainID,
		).Update("aws_rout53_delete_change_id", awsRoute53DeleteChangeID).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to update plc domain records by attach domain id")
	}
	return nil
}