package domain

import (
	"time"

	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"google.golang.org/protobuf/types/known/structpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type GetBizChangefeedReq struct {
	BizChangefeedID uint64
}

type GetBizChangefeedResp struct {
	BizChangefeed *BizChangefeed
}

type BizChangefeed struct {
	BizChangefeedID     uint64
	Name                string
	CreatedAt           time.Time
	SpecificationCode   string
	DownstreamType      entity.TiCDCChangefeedDownStreamType
	DownstreamParameter string
	NetworkType         entity.ChangefeedDownstreamNetworkType
	NetworkInfo         string
	DeployType          entity.TiCDCClusterDeployType
	OperationStatus     entity.BizChangefeedOperationStatus
	ChangefeedStatus    entity.ChangefeedKernelStatus
	CheckpointTSO       uint64
	ErrorMessage        string
	ClusterID           uint64
	ProjectID           uint64
	TenantID            uint64
	TiCDCClusterID      uint64
}

func (c *BizChangefeed) GetCheckpointUTCTime() time.Time {
	return utils.GetTimeFromTS(c.CheckpointTSO)
}

func (c *BizChangefeed) GetConnectorStatus() dataflow.Connector_Status {
	if c.ChangefeedStatus == entity.ChangefeedKernelStatusFailed ||
		c.ChangefeedStatus == entity.ChangefeedKernelStatusError {
		return ChangefeedKernelStatus2ConnectorStatus(c.ChangefeedStatus)
	}
	if c.OperationStatus == entity.BizChangefeedOperationStatusNormal {
		return ChangefeedKernelStatus2ConnectorStatus(c.ChangefeedStatus)
	} else {
		return BizChangefeedOperationStatus2ConnectorStatus(c.OperationStatus)
	}
}

func (c *BizChangefeed) ToDataflowConnector() *dataflow.Connector {
	connector := &dataflow.Connector{
		ConnectorId:            c.BizChangefeedID,
		ConnectorType:          TiCDCChangefeedDownStreamType2ConnectorType(c.DownstreamType),
		Name:                   c.Name,
		CheckpointTso:          c.CheckpointTSO,
		CreatedAt:              timestamppb.New(c.CreatedAt),
		Status:                 c.GetConnectorStatus(),
		Error:                  c.ErrorMessage,
		ChangefeedInstanceType: c.SpecificationCode,
	}
	if connector.CheckpointTso > 0 {
		connector.CheckpointUtc = timestamppb.New(c.GetCheckpointUTCTime())
	}
	return connector
}

type ListBizChangefeedsReq struct {
	Page           uint32
	PerPage        uint32
	ClusterID      uint64
	DownstreamType *entity.TiCDCChangefeedDownStreamType
}

type ListBizChangefeedsResp struct {
	BizChangefeeds []*BizChangefeed
	Total          uint32
}

func (r *ListBizChangefeedsResp) ToDataflowConnectors() *dataflow.Connectors {
	connectors := &dataflow.Connectors{
		Total: r.Total,
		Items: make([]*dataflow.Connector, 0, len(r.BizChangefeeds)),
	}

	for _, bizChangefeed := range r.BizChangefeeds {
		connectors.Items = append(connectors.Items, bizChangefeed.ToDataflowConnector())
	}
	return connectors
}

type CreateBizChangefeedReq struct {
	Name                  string
	ClusterID             uint64
	DownstreamType        dataflow.ConnectorType
	DownstreamParameter   interface{}
	NetworkInfo           *dataflow.NetworkInfo
	DeployType            entity.TiCDCClusterDeployType
	SpecificationCode     string
	UserID                uint64
	PrivateLinkEndpointID entity.ID

	DryRun bool
}

type CreateBizChangefeedResp struct {
	BizChangefeedID uint64
	WorkflowID      string
}

type DeleteBizChangefeedReq struct {
	BizChangefeedID uint64
	UserID          uint64
}

type DeleteBizChangefeedResp struct {
	WorkflowID string
}

type PauseBizChangefeedReq struct {
	BizChangefeedID uint64
	UserID          uint64
}

type ResumeBizChangefeedReq struct {
	BizChangefeedID uint64
	UserID          uint64
}

type EditBizChangefeedReq struct {
	BizChangefeedID     uint64
	Name                string
	DownstreamType      dataflow.ConnectorType
	DownstreamParameter interface{}
	UserID              uint64
	DryRun              bool
}

type ScaleBizChangefeedReq struct {
	BizChangefeedID   uint64
	SpecificationCode string
	UserID            uint64
}

type ScaleBizChangefeedResp struct {
	WorkflowID string
}

type TestBizChangefeedReq struct {
	ClusterID           uint64
	DownstreamType      dataflow.ConnectorType
	DownstreamParameter interface{}
}

type GetClusterBizChangefeedRCUReq struct {
	ClusterID uint64
}

type GetClusterBizChangefeedRCUResp struct {
	RCUNumber uint32
}

func (r *GetClusterBizChangefeedRCUResp) ToDataflowResp() *dataflow.GetClusterTCUResp {
	return &dataflow.GetClusterTCUResp{
		TcuNumber: uint64(r.RCUNumber),
	}
}

type CalBizChangefeedFeeReq struct {
	TenantID          uint64
	ProjectID         uint64
	RegionName        string
	CloudProvider     CloudProviderType
	SpecificationCode string
	Duration          int64
}

type CalBizChangefeedFeeResp struct {
	Fee    float64
	RawFee float64
}

func (r *CalBizChangefeedFeeResp) ToDataflowResp() *dataflow.CalChangefeedFeeResp {
	return &dataflow.CalChangefeedFeeResp{
		Fee:    r.Fee,
		RawFee: r.RawFee,
	}
}

type ListBizChangefeedSpecificationsReq struct {
	ClusterID uint64
}

type ListBizChangefeedSpecificationsResp struct {
	Specifications []*BizChangefeedSpecification
	Total          uint32
}

type GetBizChangefeedDetailReq struct {
	BizChangefeedID uint64
}

type GetBizChangefeedDetailResp struct {
	Changefeed *structpb.Struct
}

type GetBizChangefeedClusterReq struct {
	BizChangefeedID uint64
}

type ListTiCDCClusterReq struct {
	Cluster *core.Cluster
}

type GetBizChangefeedClusterResp struct {
	TiCDCCluster *structpb.Struct
}

type ListTiCDCClusterResp struct {
	TiCDCClusters *structpb.Struct
}

type DescribeAdminProfileReq struct {
	TenantID  uint64
	ProjectID uint64
	ClusterID uint64
}

type DescribeAdminProfileResp struct {
	InfraProviderType InfraProviderType
	AWSAccountID      string
	AWSExternalID     string
	GCPServiceAccount string
}

type DescribeSchemaTableReq struct {
	ClusterID     uint64
	Filters       []string
	CaseSensitive bool
}

type DescribeSchemaTableResp struct {
	ReplicateAllowSchemaTable []string
	ReplicateDenySchemaTable  []string
	GcSafePoint               uint64
}

func (r *ListBizChangefeedSpecificationsResp) ToDataflowResp() *dataflow.ChangefeedInstanceTypes {
	instanceTypes := &dataflow.ChangefeedInstanceTypes{
		Total: r.Total,
		Items: make([]*dataflow.ChangefeedInstanceType, 0, len(r.Specifications)),
	}
	for _, spec := range r.Specifications {
		instanceTypes.Items = append(instanceTypes.Items, spec.ToDataflowInstanceType())
	}
	return instanceTypes
}

type BizChangefeedSpecification struct {
	SpecificationCode string
	RCUNumber         uint32
	Replicas          uint32
	RPSLimit          uint32
}

func (s *BizChangefeedSpecification) ToDataflowInstanceType() *dataflow.ChangefeedInstanceType {
	return &dataflow.ChangefeedInstanceType{
		Name:     s.SpecificationCode,
		RcuNum:   s.RCUNumber,
		Replicas: s.Replicas,
		RpsLimit: s.RPSLimit,
	}
}