package datasynchronization

import (
	"context"
	"fmt"
	"slices"

	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/pkgv2/db"
	"go.opentelemetry.io/otel"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	dataflowcli "github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/dataflow"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/dedicatedglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/remoteencryptor"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/workflow/dag"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	utilworkflow "github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	defaultRcu = 256 // default rcu 256
)

var tracer = otel.Tracer("Domain.DataSynchronization")

type Service interface {
	Create(ctx context.Context, ds *entity.DataSynchronization, validateOnly bool) (*entity.DataSynchronization, error)
	Get(ctx context.Context, id uint64) (*entity.DataSynchronization, error)
	Delete(ctx context.Context, id uint64) error
}

type service struct {
	dsRepo       repository.DataSynchronizationRepo
	clusterRepo  repository.ClusterRepository
	dedicatedCli dedicatedglobal.DedicatedGlobal
	dataflowCli  dataflowcli.DataFlow
	encryptor    remoteencryptor.Client
	transaction  db.TransProvider
	workflow     utilworkflow.Workflow
}

func NewService(
	dsRepo repository.DataSynchronizationRepo,
	clusterRepo repository.ClusterRepository,
	dedicatedCli dedicatedglobal.DedicatedGlobal,
	dataflowCli dataflowcli.DataFlow,
	encryptor remoteencryptor.Client,
	transaction db.TransProvider,
	workflow utilworkflow.Workflow,
) Service {
	return &service{
		dsRepo:       dsRepo,
		clusterRepo:  clusterRepo,
		dedicatedCli: dedicatedCli,
		dataflowCli:  dataflowCli,
		encryptor:    encryptor,
		transaction:  transaction,
		workflow:     workflow,
	}
}

func (s *service) Create(ctx context.Context, ds *entity.DataSynchronization, validateOnly bool) (*entity.DataSynchronization, error) {
	spanName := "DataSynchronizationService.Create"
	ctx, span := tracer.Start(ctx, spanName)
	defer span.End()

	if ds == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "data synchronization is nil")
	}
	if ds.CdcUsername != "" && ds.CdcPassword == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cdc_password is required when cdc_username is set")
	}
	if ds.CdcUsername == "" && ds.CdcPassword != "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cdc_username is required when cdc_password is set")
	}
	if !validateOnly && ds.CdcUsername != "" {
		encrypted, err := s.encryptor.EncryptBase64(ctx, ds.CdcPassword)
		if err != nil {
			return nil, bizerr.WrapError(err, "encrypt cdc password failed")
		}
		ds.CdcPassword = encrypted
	}
	// Never return password in response.
	ds.TargetClusterPassword = ""

	// Validate input and fetch cluster info.
	sourceCluster, err := s.clusterRepo.GetCluster(ctx, ds.GetSourceClusterId(), false)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get source cluster")
	}
	if sourceCluster.DeployType != entity.DeployTypeDedicated {
		return nil, bizerr.New(bizerr.ErrCodeParams, "source cluster must be a dedicated cluster")
	}

	targetCluster, err := s.clusterRepo.GetCluster(ctx, ds.GetTargetClusterId(), false)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get target cluster")
	}
	if targetCluster.DeployType != entity.DeployTypeDedicated {
		return nil, bizerr.New(bizerr.ErrCodeParams, "target cluster must be a dedicated cluster")
	}
	crossProject := sourceCluster.ProjectId != targetCluster.ProjectId

	// Set initial state.
	ds.State = tidbmgmtservicev1.DataSynchronization_PENDING
	if ds.GetRcu() == 0 {
		ds.Rcu = defaultRcu
	}

	// get available zone ids from source cluster and target cluster
	sourceAvailableZones, err := s.dataflowCli.ListSinkAvailabilityZones(ctx, &dataflow.ListSinkAvailabilityZonesRequest{
		OrgId:     sourceCluster.OrganizationId,
		ProjectId: sourceCluster.ProjectId,
		Provider:  sourceCluster.CloudProvider.String(),
		Region:    sourceCluster.Region,
	})
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get source available zones")
	}
	targetAvailableZones, err := s.dataflowCli.ListSinkAvailabilityZones(ctx, &dataflow.ListSinkAvailabilityZonesRequest{
		OrgId:     targetCluster.OrganizationId,
		ProjectId: targetCluster.ProjectId,
		Provider:  targetCluster.CloudProvider.String(),
		Region:    targetCluster.Region,
	})
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get target available zones")
	}

	availableZoneIDs := []string{}
	for _, zone := range sourceAvailableZones.GetAvailabilityZones() {
		if slices.Contains(targetAvailableZones.GetAvailabilityZones(), zone) {
			availableZoneIDs = append(availableZoneIDs, zone)
		}
	}
	if len(availableZoneIDs) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "no compatible zones found for source and target clusters")
	}
	ds.AvailableZoneIDs = availableZoneIDs

	// Fast return and prevent repo saving if validate only.
	if validateOnly {
		return ds, nil
	}

	// Persist record then start workflow.
	if err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		if err := s.dsRepo.Create(ctx, ds); err != nil {
			return err
		}

		if err := dag.RunCreateDataSynchronizationDag(ctx, s.workflow, &dag.CreateDataSynchronizationDagTasks{
			TenantId:              sourceCluster.OrganizationId,
			ProjectId:             sourceCluster.ProjectId,
			DataSynchronizationId: ds.Id,
			CrossProject:          crossProject,
			SpanName:              spanName,
		}); err != nil {
			return err
		}
		return nil
	}); err != nil {
		return nil, bizerr.WrapError(err, "failed to start create data synchronization workflow")
	}

	return ds, nil
}

func (s *service) Get(ctx context.Context, id uint64) (*entity.DataSynchronization, error) {
	ctx, span := tracer.Start(ctx, "DataSynchronizationService.Get")
	defer span.End()

	ds, err := s.dsRepo.Get(ctx, id)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get data synchronization")
	}

	return ds, nil
}

func (s *service) Delete(ctx context.Context, id uint64) error {
	ctx, span := tracer.Start(ctx, "DataSynchronizationService.Delete")
	defer span.End()

	ds, err := s.dsRepo.Get(ctx, id)
	if err != nil {
		return bizerr.WrapGormError(err, "failed to get data synchronization for deletion")
	}

	// Get source cluster info for IDs
	sourceCluster, err := s.clusterRepo.GetCluster(ctx, ds.GetSourceClusterId(), true)
	if err != nil {
		return bizerr.WrapGormError(err, "failed to get source cluster for deletion")
	}

	// delete all dag instances related to this data synchronization
	if err := s.workflow.DeleteDagInstancesWithDagId(ctx, fmt.Sprintf("%d", id), constant.DagPrefixDataSynchronizationCreate); err != nil {
		return bizerr.WrapError(err, "failed to delete data synchronization dag instances")
	}

	// Delete Changefeed in dataflow-service
	if ds.ChangefeedConnectorID != 0 {
		deleteChangefeedReq := &dataflow.ConnectorID{
			ConnectorId: ds.ChangefeedConnectorID,
			ClusterId:   ds.GetSourceClusterId(),
			ProjectId:   sourceCluster.ProjectId,
			OrgId:       sourceCluster.OrganizationId,
		}
		if err := s.dataflowCli.DeleteConnector(ctx, deleteChangefeedReq); err != nil {
			// Ignore error if connector is already deleted
			grpcStatus, ok := status.FromError(err)
			if !ok || grpcStatus.Code() != codes.NotFound {
				return bizerr.WrapGrpcError(err, "failed to delete changefeed in dataflow service")
			}
			// Connector already deleted, continue
		}
	}

	// Delete Sink Endpoint if it exists
	if ds.SinkEndpointID != 0 {
		deleteSinkReq := &dataflow.DeleteSinkEndpointRequest{
			SinkEndpointId: ds.SinkEndpointID,
			ProjectId:      sourceCluster.ProjectId,
			OrgId:          sourceCluster.OrganizationId,
		}
		if err := s.dataflowCli.DeleteSinkEndpoint(ctx, deleteSinkReq); err != nil {
			// Ignore error if sink endpoint is already deleted
			grpcStatus, ok := status.FromError(err)
			if !ok || grpcStatus.Code() != codes.NotFound {
				return bizerr.WrapGrpcError(err, "failed to delete sink endpoint in dataflow service")
			}
			// Sink endpoint already deleted, continue
		}
	}

	// Delete DataSynchronization entity from persistence
	if err := s.dsRepo.Delete(ctx, id); err != nil {
		return bizerr.WrapGormError(err, "failed to delete data synchronization from persistence")
	}

	return nil
}