package domain

import (
	"time"

	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"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
	OperationStatus     entity.BizChangefeedOperationStatus
	ChangefeedStatus    entity.ChangefeedKernelStatus
	CheckpointTSO       uint64
	ErrorMessage        string
	ClusterID           uint64
	TenantID            uint64
}

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

func (c *BizChangefeed) GetConnectorStatus() dataflow.Changefeed_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.Changefeed {
	connector := &dataflow.Changefeed{
		ChangefeedId:   c.BizChangefeedID,
		ChangefeedType: TiCDCChangefeedDownStreamType2ConnectorType(c.DownstreamType),
		Name:           c.Name,
		CheckpointTso:  c.CheckpointTSO,
		CreatedAt:      timestamppb.New(c.CreatedAt),
		Status:         c.GetConnectorStatus(),
		Error:          c.ErrorMessage,
		ChangefeedSpec: 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.Changefeeds {
	connectors := &dataflow.Changefeeds{
		Total: r.Total,
		Items: make([]*dataflow.Changefeed, 0, len(r.BizChangefeeds)),
	}

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

type CreateBizChangefeedReq struct {
	// ID lets the caller choose the changefeed's identity instead of learning
	// it from the response. A create whose result was lost can then be resolved
	// with a Get on a key the caller already holds, which is what makes an
	// orchestrated create idempotent. Nil keeps the default: the repo
	// allocates. A non-nil zero is rejected rather than treated as nil, so a
	// caller that meant to choose cannot silently get an allocated id instead.
	ID                    *uint64
	Name                  string
	ClusterID             uint64
	DownstreamType        dataflow.ChangefeedType
	DownstreamParameter   interface{}
	NetworkInfo           *dataflow.NetworkInfo
	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.ChangefeedType
	DownstreamParameter interface{}
	UserID              uint64
	DryRun              bool
}

// ApplyStoppedBizChangefeedConfigReq is used by internal orchestrators to
// update a changefeed while both the business and kernel resources remain
// stopped. Downstream location and credentials are completed by the adaptor
// for the changefeed's own downstream before persistence, as they are for an
// ordinary edit.
//
// There is no downstream type: this edits a changefeed that already exists,
// and its downstream is a property of that changefeed rather than something a
// caller declares. DownstreamParameter must be the shape that downstream's
// adaptor takes.
//
// There is no specification field either: resizing goes through
// ScaleBizChangefeed, which owns the asynchronous TiCDC resource
// reconciliation.
type ApplyStoppedBizChangefeedConfigReq struct {
	BizChangefeedID     uint64
	Name                string
	DownstreamParameter interface{}
	UserID              uint64
}

type ScaleBizChangefeedReq struct {
	BizChangefeedID   uint64
	SpecificationCode string
	UserID            uint64
}

type ScaleBizChangefeedResp struct {
	WorkflowID string
}

type TestBizChangefeedReq struct {
	ClusterID           uint64
	DownstreamType      dataflow.ChangefeedType
	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 {
	ClusterID         uint64
	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 *utils.Cluster
}

type GetBizChangefeedClusterResp struct {
	TiCDCCluster *structpb.Struct
}

type ListTiCDCClusterResp struct {
	TiCDCClusters *structpb.Struct
}

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

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

type DescribeAdminProfileReq struct {
	ClusterID uint64
}

type DescribeAdminProfileResp struct {
	ProviderType string
	// CloudIAMIdentity is the identity the changefeed assumes a customer role
	// from - an AWS role ARN, or a GCP service account.
	CloudIAMIdentity string
	CloudAccountID   string
	AWSExternalID    string
}

func (r *ListBizChangefeedSpecificationsResp) ToDataflowResp() *dataflow.ChangefeedSpecs {
	instanceTypes := &dataflow.ChangefeedSpecs{
		Total: r.Total,
		Items: make([]*dataflow.ChangefeedSpec, 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.ChangefeedSpec {
	return &dataflow.ChangefeedSpec{
		Name:     s.SpecificationCode,
		RcuNum:   s.RCUNumber,
		RpsLimit: s.RPSLimit,
	}
}