package datasynchronization

import (
	"context"
	"errors"

	"github.com/samber/lo"
	"go.opentelemetry.io/otel"
	"google.golang.org/protobuf/types/known/emptypb"
	"gorm.io/gorm"

	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/datasynchronization"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/dedicated/cluster"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/apimiddleware/bizerror"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

var tracer = otel.Tracer("App.InternalApi.DataSynchronization")

func NewServer(
	dataSyncSvc datasynchronization.Service,
	dedicatedClusterSvc cluster.Service,
) tidbmgmtservicev1.DataSynchronizationServiceServer {
	return &serverImpl{
		dataSyncSvc:         dataSyncSvc,
		dedicatedClusterSvc: dedicatedClusterSvc,
	}
}

type serverImpl struct {
	dataSyncSvc         datasynchronization.Service
	dedicatedClusterSvc cluster.Service
}

func (s *serverImpl) CreateDataSynchronization(
	ctx context.Context,
	req *tidbmgmtservicev1.CreateDataSynchronizationRequest,
) (*tidbmgmtservicev1.DataSynchronization, error) {
	ctx, span := tracer.Start(ctx, "CreateDataSynchronization")
	defer span.End()

	dsReq := req.GetDataSynchronization()
	if dsReq == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "data_synchronization is required")
	}

	// Validate required fields
	if dsReq.GetSourceClusterId() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "source_cluster_id is required")
	}
	if dsReq.GetTargetClusterId() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "target_cluster_id is required")
	}
	if dsReq.GetName() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "name is required")
	}
	if req.GetDataSynchronization().GetTargetClusterUsername() != "" && req.GetDataSynchronization().GetTargetClusterPassword() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "target_cluster_password is required when target_cluster_username is set")
	}
	if req.GetDataSynchronization().GetTargetClusterUsername() == "" && req.GetDataSynchronization().GetTargetClusterPassword() != "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "target_cluster_username is required when target_cluster_password is set")
	}

	sourceCluster, err := s.dedicatedClusterSvc.GetCluster(ctx, dsReq.GetSourceClusterId(), entity.ClusterViewClusterViewLocal, false, false)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, bizerror.NewNotFound(err, "dedicated cluster not found")
		}
		return nil, bizerror.NewInternal(err, "failed to get dedicated cluster")
	}

	// gen request meta and append to outgoing meta
	ctx = metadata.AppendToOutgoingMeta(ctx, &metadata.RequestMeta{
		OrgID:             sourceCluster.OrganizationId,
		DefaultProjectID:  sourceCluster.ProjectId,
		ValidProjectList:  []uint64{sourceCluster.ProjectId},
		UserID:            "0",
		IsCheckPermission: lo.ToPtr(true),
	})

	// Convert proto message to entity
	ds := &entity.DataSynchronization{
		DataSynchronization: lo.FromPtr(dsReq),
		CdcUsername:         dsReq.GetTargetClusterUsername(),
		CdcPassword:         dsReq.GetTargetClusterPassword(),
	}

	// Call domain service to create
	result, err := s.dataSyncSvc.Create(ctx, ds, req.GetValidateOnly())
	if err != nil {
		return nil, err
	}

	return &result.DataSynchronization, nil
}

func (s *serverImpl) GetDataSynchronization(
	ctx context.Context,
	req *tidbmgmtservicev1.GetDataSynchronizationRequest,
) (*tidbmgmtservicev1.DataSynchronization, error) {
	ctx, span := tracer.Start(ctx, "GetDataSynchronization")
	defer span.End()

	if req.GetId() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "id is required")
	}

	result, err := s.dataSyncSvc.Get(ctx, req.GetId())
	if err != nil {
		return nil, err
	}

	return &result.DataSynchronization, nil
}

func (s *serverImpl) DeleteDataSynchronization(
	ctx context.Context,
	req *tidbmgmtservicev1.DeleteDataSynchronizationRequest,
) (*emptypb.Empty, error) {
	ctx, span := tracer.Start(ctx, "DeleteDataSynchronization")
	defer span.End()

	if req.GetId() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "id is required")
	}

	ds, err := s.dataSyncSvc.Get(ctx, req.GetId())
	if err != nil {
		return nil, err
	}

	sourceCluster, err := s.dedicatedClusterSvc.GetCluster(ctx, ds.GetSourceClusterId(), entity.ClusterViewClusterViewLocal, false, false)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, bizerror.NewNotFound(err, "dedicated cluster not found")
		}
		return nil, bizerror.NewInternal(err, "failed to get dedicated cluster")
	}

	// gen request meta and append to outgoing meta
	ctx = metadata.AppendToOutgoingMeta(ctx, &metadata.RequestMeta{
		OrgID:             sourceCluster.OrganizationId,
		DefaultProjectID:  sourceCluster.ProjectId,
		ValidProjectList:  []uint64{sourceCluster.ProjectId},
		UserID:            "0",
		IsCheckPermission: lo.ToPtr(true),
	})

	if err := s.dataSyncSvc.Delete(ctx, req.GetId()); err != nil {
		return nil, err
	}

	return &emptypb.Empty{}, nil
}