package console

import (
	"context"

	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/tidbcloud/dataflow-service-ng/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/sink_endpoint"
	"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"
	"github.com/tidbcloud/dataflow-service-ng/internal/bizutils"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/structpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type ChangefeedController interface {
	dataflow.ChangefeedServiceServer
}

type bizChangefeedControllerImpl struct {
	dataflow.UnimplementedChangefeedServiceServer
	*BaseController

	changefeedModel   biz_changefeed.BizChangefeedModel
	downstreamFactory *bizChangefeedDownstreamAdaptorFactory
	featureGate       feature_gate.FeatureGate

	mgmtSvcCli grpcli.MgmtSvcGrpcClient
}

func NewBiZChangefeedController(
	trans db.TransProvider,
	changefeedModel biz_changefeed.BizChangefeedModel,
	sinkEndpointModel sink_endpoint.SinkEndpointModel,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
	featureGate feature_gate.FeatureGate,
) ChangefeedController {
	awsSession := session.Must(session.NewSession())
	downstreamFactory := newBizChangefeedDownstreamAdaptorFactory(awsSession, sinkEndpointModel)
	return &bizChangefeedControllerImpl{
		BaseController:    NewBaseController(trans),
		changefeedModel:   changefeedModel,
		downstreamFactory: downstreamFactory,
		mgmtSvcCli:        mgmtSvcCli,
		featureGate:       featureGate,
	}
}

func (c *bizChangefeedControllerImpl) ListChangefeeds(
	ctx context.Context, req *dataflow.ChangefeedListReq,
) (*dataflow.Changefeeds, error) {
	ctx = log.WithTraceID(ctx)
	listReq := &domain.ListBizChangefeedsReq{
		ClusterID: req.ClusterId,
		Page:      req.Page,
		PerPage:   req.PerPage,
	}
	if req.ChangefeedType != dataflow.ChangefeedType_TOTAL {
		downstreamType := domain.ConnectorType2TiCDCChangefeedDownStreamType(req.ChangefeedType)
		listReq.DownstreamType = &downstreamType
	}

	listResp, bizErr := c.changefeedModel.ListBizChangefeeds(ctx, listReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "list biz_changefeeds error")
		return nil, bizErr
	}

	connectors := listResp.ToDataflowConnectors()
	for i := range connectors.Items {
		if bizErr = c.fillCreateReq2Connector(connectors.Items[i], listResp.BizChangefeeds[i]); bizErr != nil {
			logger := log.WithRedactedRequest(log.GetLogger(ctx), req).With(
				zap.Uint64("changefeed_id", listResp.BizChangefeeds[i].BizChangefeedID))
			c.logError(logger, bizErr, "assemble create request error")
			return nil, bizErr
		}
	}

	return connectors, nil
}

func (c *bizChangefeedControllerImpl) DeleteChangefeed(
	ctx context.Context, req *dataflow.ChangefeedID,
) (_ *emptypb.Empty, rErr error) {
	ctx = log.WithTraceID(ctx)
	var cluster *mgmt.Cluster
	auditCtx := c.loadChangefeedAuditContext(ctx, req.GetChangefeedId())
	defer func() {
		c.recordAuditLogForDeleteChangefeed(ctx, cluster, req, auditCtx, rErr)
	}()

	cluster, bizErr := c.mgmtSvcCli.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		rErr = bizErr
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}

	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	deleteReq := &domain.DeleteBizChangefeedReq{
		BizChangefeedID: req.ChangefeedId,
		UserID:          userID,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	deleteResp, bizErr := c.changefeedModel.DeleteBizChangefeed(ctx, deleteReq)
	if bizErr != nil {
		rErr = bizErr
		c.logError(logger, bizErr, "delete biz_changefeeds error")
		return nil, bizErr
	}

	logger.Info("delete biz_changefeed successfully", zap.String("workflow_id", deleteResp.WorkflowID))
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) ScaleChangefeed(
	ctx context.Context, req *dataflow.ChangefeedScaleReq,
) (_ *emptypb.Empty, rErr error) {
	ctx = log.WithTraceID(ctx)
	var cluster *mgmt.Cluster
	auditCtx := c.loadChangefeedAuditContext(ctx, req.GetChangefeedId())
	defer func() {
		c.recordAuditLogForScaleChangefeed(ctx, cluster, req, auditCtx, rErr)
	}()

	cluster, bizErr := c.mgmtSvcCli.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		rErr = bizErr
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}

	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	scaleReq := &domain.ScaleBizChangefeedReq{
		BizChangefeedID:   req.ChangefeedId,
		SpecificationCode: req.ChangefeedSpec,
		UserID:            userID,
	}
	scaleResp, bizErr := c.changefeedModel.ScaleBizChangefeed(ctx, scaleReq)
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	if bizErr != nil {
		rErr = bizErr
		c.logError(logger, bizErr, "scale biz_changefeeds error")
		return nil, bizErr
	}

	logger.Info("scale biz_changefeed successfully", zap.String("workflow_id", scaleResp.WorkflowID))
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) CalChangefeedFee(
	ctx context.Context, req *dataflow.CalChangefeedFeeReq,
) (*dataflow.CalChangefeedFeeResp, error) {
	ctx = log.WithTraceID(ctx)
	feeReq := &domain.CalBizChangefeedFeeReq{
		ClusterID:         req.ClusterId,
		SpecificationCode: req.ChangefeedSpec,
		Duration:          req.Duration,
	}
	feeResp, bizErr := c.changefeedModel.CalBizChangefeedFee(ctx, feeReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "calculate biz_changefeed fee error")
		return nil, bizErr
	}
	return feeResp.ToDataflowResp(), nil
}

func (c *bizChangefeedControllerImpl) ResumeChangefeed(
	ctx context.Context, req *dataflow.ChangefeedID,
) (*emptypb.Empty, error) {
	return c.runChangefeedStateChange(
		ctx,
		req,
		c.recordAuditLogForResumeChangefeed,
		func(ctx context.Context, changefeedID, userID uint64) bizerr.BizError {
			return c.changefeedModel.ResumeBizChangefeed(ctx, &domain.ResumeBizChangefeedReq{
				BizChangefeedID: changefeedID,
				UserID:          userID,
			})
		},
		"resume biz_changefeeds",
		"resume biz_changefeed",
	)
}

func (c *bizChangefeedControllerImpl) PauseChangefeed(
	ctx context.Context, req *dataflow.ChangefeedID,
) (*emptypb.Empty, error) {
	return c.runChangefeedStateChange(
		ctx,
		req,
		c.recordAuditLogForPauseChangefeed,
		func(ctx context.Context, changefeedID, userID uint64) bizerr.BizError {
			return c.changefeedModel.PauseBizChangefeed(ctx, &domain.PauseBizChangefeedReq{
				BizChangefeedID: changefeedID,
				UserID:          userID,
			})
		},
		"pause biz_changefeeds",
		"pause biz_changefeed",
	)
}

func (c *bizChangefeedControllerImpl) runChangefeedStateChange(
	ctx context.Context,
	req *dataflow.ChangefeedID,
	recordAudit func(context.Context, *mgmt.Cluster, *dataflow.ChangefeedID, changefeedAuditContext, error),
	changeState func(context.Context, uint64, uint64) bizerr.BizError,
	errLogPrefix, successLog string,
) (_ *emptypb.Empty, rErr error) {
	ctx = log.WithTraceID(ctx)
	var cluster *mgmt.Cluster
	auditCtx := c.loadChangefeedAuditContext(ctx, req.GetChangefeedId())
	defer func() {
		recordAudit(ctx, cluster, req, auditCtx, rErr)
	}()

	cluster, bizErr := c.mgmtSvcCli.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		rErr = bizErr
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}

	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	if bizErr := changeState(ctx, req.ChangefeedId, userID); bizErr != nil {
		rErr = bizErr
		c.logError(logger, bizErr, errLogPrefix+" error")
		return nil, bizErr
	}

	logger.Info(successLog + " successfully")
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) GetChangefeed(
	ctx context.Context, req *dataflow.ChangefeedID,
) (*dataflow.Changefeed, error) {
	ctx = log.WithTraceID(ctx)
	getReq := &domain.GetBizChangefeedReq{
		BizChangefeedID: req.ChangefeedId,
	}
	getResp, bizErr := c.changefeedModel.GetBizChangefeed(ctx, getReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "get biz_changefeed error")
		return nil, bizErr
	}

	connector := getResp.BizChangefeed.ToDataflowConnector()
	if bizErr = c.fillCreateReq2Connector(connector, getResp.BizChangefeed); bizErr != nil {
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req).With(
			zap.Uint64("changefeed_id", req.GetChangefeedId()))
		c.logError(logger, bizErr, "assemble create request error")
		return nil, bizErr
	}

	return connector, nil
}

func (c *bizChangefeedControllerImpl) DescribeSchemaTable(
	ctx context.Context, req *dataflow.DescribeSchemaTableReq,
) (*dataflow.DescribeSchemaTableResp, error) {
	ctx = log.WithTraceID(ctx)
	pauseReq := &domain.DescribeSchemaTableReq{
		ClusterID:     req.ClusterId,
		Filters:       req.FilterRule,
		CaseSensitive: req.CaseSensitive,
		Password:      req.Password,
		Username:      req.Username,
	}
	logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
	mresp, bizErr := c.changefeedModel.DescribeSchemaTable(ctx, pauseReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "describe schema error")
		return nil, bizErr
	}
	logger.Info("describe schema successfully")
	return &dataflow.DescribeSchemaTableResp{
		ClusterId:                 req.ClusterId,
		ReplicateDenySchemaTable:  mresp.ReplicateDenySchemaTable,
		ReplicateAllowSchemaTable: mresp.ReplicateAllowSchemaTable,
		GcSafePoint:               timestamppb.New(utils.GetTimeFromTS(mresp.GcSafePoint)),
	}, nil
}

func (c *bizChangefeedControllerImpl) GetAdminProfile(
	ctx context.Context, req *dataflow.ClusterIdReq,
) (*dataflow.AdminProfile, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	profileReq := &domain.DescribeAdminProfileReq{
		ClusterID: req.ClusterId,
	}
	profileResp, err := c.changefeedModel.DescribeAdminProfile(ctx, profileReq)
	if err != nil {
		c.logError(logger, err, "get admin profile error")
		return nil, err
	}
	logger.Info("get admin profile successfully")
	return &dataflow.AdminProfile{
		Provider:       profileResp.ProviderType,
		CloudAccountId: profileResp.CloudAccountID,
		AwsExternalId:  profileResp.AWSExternalID,
	}, nil
}

func (c *bizChangefeedControllerImpl) fillCreateReq2Connector(
	connector *dataflow.Changefeed, bizChangefeed *domain.BizChangefeed,
) bizerr.BizError {
	downstreamAdaptor, ok := c.downstreamFactory.Get(connector.ChangefeedType)
	if !ok {
		return bizerr.New(bizerr.ErrCodeUnknown, "invalid connector type: %d", connector.ChangefeedType)
	}
	createReq, bizErr := downstreamAdaptor.AssembleGetResponse(bizChangefeed)
	if bizErr != nil {
		return bizErr
	}
	connector.Request = createReq
	return nil
}

func (c *bizChangefeedControllerImpl) GetClusterTCU(
	ctx context.Context, req *dataflow.GetClusterTCUReq,
) (*dataflow.GetClusterTCUResp, error) {
	ctx = log.WithTraceID(ctx)
	rcuReq := &domain.GetClusterBizChangefeedRCUReq{
		ClusterID: req.ClusterId,
	}
	rcuResp, bizErr := c.changefeedModel.GetClusterBizChangefeedRCU(ctx, rcuReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "get cluster biz_changefeed rcu error")
		return nil, bizErr
	}
	return rcuResp.ToDataflowResp(), nil
}

func (c *bizChangefeedControllerImpl) CreateChangefeed(
	ctx context.Context, req *dataflow.ChangefeedCreateReq,
) (resp *dataflow.ChangefeedID, rErr error) {
	ctx = log.WithTraceID(ctx)
	var cluster *mgmt.Cluster
	var changefeedID uint64
	if !req.GetDryRun() {
		defer func() {
			c.recordAuditLogForCreateChangefeed(ctx, cluster, req, changefeedID, rErr)
		}()
	}

	downstreamAdaptor, ok := c.downstreamFactory.Get(req.ChangefeedType)
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid changefeed type: %d", req.ChangefeedType)
	}

	cluster, bizErr := c.mgmtSvcCli.GetClusterByID(ctx, req.ClusterId)
	if bizErr != nil {
		rErr = bizErr
		logger := log.GetLogger(ctx).With(zap.Uint64("cluster_id", req.ClusterId))
		c.logError(logger, bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}
	downstreamParams, bizErr := downstreamAdaptor.ParseCreateParams(ctx, req, utils.NewCluster(cluster))
	if bizErr != nil {
		rErr = bizErr
		logger := log.GetLogger(ctx).With(zap.Uint64("cluster_id", req.ClusterId))
		c.logError(logger, bizErr, "parse param error")
		return nil, bizErr.Chain("get cluster")
	}
	if bizErr := downstreamAdaptor.ValidateCreateParams(downstreamParams, utils.NewCluster(cluster)); bizErr != nil {
		rErr = bizErr
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "validate downstream params error")
		return nil, bizErr
	}
	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	createReq := &domain.CreateBizChangefeedReq{
		Name:                req.Name,
		ClusterID:           req.ClusterId,
		DownstreamType:      req.ChangefeedType,
		NetworkInfo:         req.NetworkInfo,
		DownstreamParameter: downstreamParams,
		SpecificationCode:   req.ChangefeedSpec,
		UserID:              userID,
		DryRun:              req.DryRun,
	}
	if req.NetworkInfo != nil && req.NetworkInfo.SinkEndpointId != nil {
		createReq.PrivateLinkEndpointID = entity.ID(*req.NetworkInfo.SinkEndpointId)
	}
	createResp, bizErr := c.changefeedModel.CreateBizChangefeed(ctx, createReq)
	if bizErr != nil {
		rErr = bizErr
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "create biz_changefeed error")
		return nil, bizErr
	}
	changefeedID = createResp.BizChangefeedID

	return &dataflow.ChangefeedID{
		ChangefeedId: createResp.BizChangefeedID,
		ClusterId:    req.ClusterId,
	}, nil
}

func (c *bizChangefeedControllerImpl) EditChangefeed(
	ctx context.Context, req *dataflow.ChangefeedEditReq,
) (_ *emptypb.Empty, rErr error) {
	ctx = log.WithTraceID(ctx)
	var cluster *mgmt.Cluster
	if !req.GetDryRun() {
		oldName := c.loadChangefeedAuditContext(ctx, req.GetChangefeedId()).name
		defer func() {
			c.recordAuditLogForEditChangefeed(ctx, cluster, req, oldName, rErr)
		}()
	}

	downstreamAdaptor, ok := c.downstreamFactory.Get(req.ChangefeedType)
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid connector type: %d", req.ChangefeedType)
	}

	cluster, bizErr := c.mgmtSvcCli.GetClusterByID(ctx, req.ClusterId)
	if bizErr != nil {
		rErr = bizErr
		logger := log.GetLogger(ctx).With(zap.Uint64("cluster_id", req.ClusterId))
		c.logError(logger, bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}
	downstreamParams := downstreamAdaptor.ParseEditParams(req)
	if bizErr := downstreamAdaptor.ValidateEditParams(downstreamParams,
		utils.NewCluster(cluster).GetVersion()); bizErr != nil {
		rErr = bizErr
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "validate downstream params error")
		return nil, bizErr
	}

	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	editReq := &domain.EditBizChangefeedReq{
		BizChangefeedID:     req.ChangefeedId,
		Name:                req.Name,
		DownstreamType:      req.ChangefeedType,
		DownstreamParameter: downstreamParams,
		UserID:              userID,
		DryRun:              req.DryRun,
	}
	if bizErr := c.changefeedModel.EditBizChangefeed(ctx, editReq); bizErr != nil {
		rErr = bizErr
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "edit biz_changefeed error")
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) TestChangefeed(
	ctx context.Context, req *dataflow.ChangefeedCreateReq,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	downstreamAdaptor, ok := c.downstreamFactory.Get(req.ChangefeedType)
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid connector type: %d", req.ChangefeedType)
	}

	cluster, bizErr := c.mgmtSvcCli.GetClusterByID(ctx, req.ClusterId)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Uint64("cluster_id", req.ClusterId))
		c.logError(logger, bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}
	downstreamParams, bizErr := downstreamAdaptor.ParseCreateParams(ctx, req, utils.NewCluster(cluster))
	if bizErr != nil {
		return nil, bizErr
	}
	testReq := &domain.TestBizChangefeedReq{
		ClusterID:           req.ClusterId,
		DownstreamType:      req.ChangefeedType,
		DownstreamParameter: downstreamParams,
	}
	if bizErr := c.changefeedModel.TestBizChangefeed(ctx, testReq); bizErr != nil {
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "test biz_changefeed error")
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) GetClusterTCUForAdmin(
	ctx context.Context, req *dataflow.GetClusterTCUReq,
) (*dataflow.GetClusterTCUResp, error) {
	// admin permission is check in func canAccessAdminEndpoint
	return c.GetClusterTCU(ctx, req)
}

func (c *bizChangefeedControllerImpl) GetChangefeedDetail(ctx context.Context,
	req *dataflow.ChangefeedID,
) (*structpb.Struct, error) {
	ctx = log.WithTraceID(ctx)
	detailReq := &domain.GetBizChangefeedDetailReq{
		BizChangefeedID: req.ChangefeedId,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	resp, bizErr := c.changefeedModel.GetChangefeedDetail(ctx, detailReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "get biz_changefeeds configuration detail error")
		return nil, bizErr
	}

	logger.Info("get biz_changefeeds configuration detail successfully")
	return resp.Changefeed, nil
}

func (c *bizChangefeedControllerImpl) GetTiCDCCluster(ctx context.Context,
	req *dataflow.ChangefeedID,
) (*structpb.Struct, error) {
	ctx = log.WithTraceID(ctx)
	clusterReq := &domain.GetBizChangefeedClusterReq{
		BizChangefeedID: req.ChangefeedId,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	resp, bizErr := c.changefeedModel.GetTiCDCCluster(ctx, clusterReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "get biz_changefeeds cluster detail error")
		return nil, bizErr
	}

	logger.Info("get biz_changefeeds cluster detail successfully")
	return resp.TiCDCCluster, nil
}

func (c *bizChangefeedControllerImpl) ListTiCDCClusters(ctx context.Context,
	req *dataflow.ClusterIdReq,
) (*structpb.Struct, error) {
	ctx = log.WithTraceID(ctx)
	cluster, bizErr := c.mgmtSvcCli.GetClusterByID(ctx, req.ClusterId)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Uint64("cluster_id", req.ClusterId))
		c.logError(logger, bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}
	clusterReq := &domain.ListTiCDCClusterReq{
		Cluster: utils.NewCluster(cluster),
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	resp, bizErr := c.changefeedModel.ListTiCDCClusters(ctx, clusterReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "get ticdc clusters failed")
		return nil, bizErr
	}
	logger.Info("get ticdc clusters detail successfully")
	return resp.TiCDCClusters, nil
}