package console

import (
	"context"

	data_pipeline "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	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"
	"google.golang.org/protobuf/types/known/emptypb"
)

type DataPipelineController interface {
	dataflow.DataPipelineServiceServer
}

type dataPipelineControllerImpl struct {
	dataflow.UnimplementedDataPipelineServiceServer

	dataPipelineModel data_pipeline.Model
	mgmtSvcCli        grpcli.MgmtSvcGrpcClient
}

func NewDataPipelineController(
	dataPipelineModel data_pipeline.Model,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
) DataPipelineController {
	return &dataPipelineControllerImpl{
		dataPipelineModel: dataPipelineModel,
		mgmtSvcCli:        mgmtSvcCli,
	}
}

func (c *dataPipelineControllerImpl) CreateDataPipeline(
	ctx context.Context,
	req *dataflow.CreateDataPipelineRequest,
) (*dataflow.DataPipelineID, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}

	modelResp, bizErr := c.dataPipelineModel.CreateDataPipeline(
		ctx,
		createDataPipelineRequestDataflow2Domain(
			req,
			bizutils.GetUIDOrZeroFromContext(ctx),
			bizutils.GetUserEmailOrEmptyFromContext(ctx),
		),
	)
	if bizErr != nil {
		return nil, bizErr
	}
	if modelResp == nil {
		return nil, invalidDataPipelineModelResponse("create")
	}

	return &dataflow.DataPipelineID{
		ClusterId:  req.GetClusterId(),
		PipelineId: modelResp.PipelineID,
	}, nil
}

func (c *dataPipelineControllerImpl) GetDataPipeline(
	ctx context.Context,
	req *dataflow.DataPipelineID,
) (*dataflow.DataPipeline, error) {
	ctx = log.WithTraceID(ctx)
	modelResp, bizErr := c.dataPipelineModel.GetDataPipeline(ctx, &domain.GetDataPipelineReq{
		ClusterID:  req.GetClusterId(),
		PipelineID: req.GetPipelineId(),
		UserID:     bizutils.GetUIDOrZeroFromContext(ctx),
	})
	if bizErr != nil {
		return nil, bizErr
	}
	if modelResp == nil || modelResp.DataPipeline == nil || modelResp.DataPipeline.Pipeline == nil {
		return nil, invalidDataPipelineModelResponse("get")
	}

	return dataPipelineDomain2Dataflow(modelResp.DataPipeline), nil
}

func (c *dataPipelineControllerImpl) ListDataPipelines(
	ctx context.Context,
	req *dataflow.ListDataPipelinesRequest,
) (*dataflow.ListDataPipelinesResponse, error) {
	ctx = log.WithTraceID(ctx)
	modelResp, bizErr := c.dataPipelineModel.ListDataPipelines(ctx, &domain.ListDataPipelinesReq{
		ClusterID: req.GetClusterId(),
		UserID:    bizutils.GetUIDOrZeroFromContext(ctx),
		Page:      req.GetPage(),
		PerPage:   req.GetPerPage(),
		Status:    dataPipelineStatusDataflow2Domain(req.GetStatus()),
	})
	if bizErr != nil {
		return nil, bizErr
	}
	if modelResp == nil {
		return nil, invalidDataPipelineModelResponse("list")
	}

	items := make([]*dataflow.DataPipeline, 0, len(modelResp.DataPipelines))
	for _, pipeline := range modelResp.DataPipelines {
		if pipeline == nil || pipeline.Pipeline == nil {
			return nil, invalidDataPipelineModelResponse("list")
		}
		items = append(items, dataPipelineDomain2Dataflow(pipeline))
	}
	return &dataflow.ListDataPipelinesResponse{
		Items: items,
		Total: modelResp.Total,
	}, nil
}

func (c *dataPipelineControllerImpl) PauseDataPipeline(
	ctx context.Context,
	req *dataflow.DataPipelineID,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := c.dataPipelineModel.PauseDataPipeline(ctx, &domain.PauseDataPipelineReq{
		ClusterID:  req.GetClusterId(),
		PipelineID: req.GetPipelineId(),
		UserID:     bizutils.GetUIDOrZeroFromContext(ctx),
	}); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *dataPipelineControllerImpl) EditDataPipeline(
	ctx context.Context,
	req *dataflow.EditDataPipelineRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := c.dataPipelineModel.EditDataPipeline(
		ctx,
		editDataPipelineRequestDataflow2Domain(req, bizutils.GetUIDOrZeroFromContext(ctx)),
	); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *dataPipelineControllerImpl) ScaleDataPipeline(
	ctx context.Context,
	req *dataflow.ScaleDataPipelineRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := c.dataPipelineModel.ScaleDataPipeline(
		ctx,
		scaleDataPipelineRequestDataflow2Domain(req, bizutils.GetUIDOrZeroFromContext(ctx)),
	); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *dataPipelineControllerImpl) ResumeDataPipeline(
	ctx context.Context,
	req *dataflow.DataPipelineID,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := c.dataPipelineModel.ResumeDataPipeline(ctx, &domain.ResumeDataPipelineReq{
		ClusterID:  req.GetClusterId(),
		PipelineID: req.GetPipelineId(),
		UserID:     bizutils.GetUIDOrZeroFromContext(ctx),
	}); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *dataPipelineControllerImpl) ValidateDataPipelineDestination(
	ctx context.Context,
	req *dataflow.ValidateDataPipelineDestinationRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := c.dataPipelineModel.ValidateDataPipelineDestination(
		ctx,
		validateDataPipelineDestinationRequestDataflow2Domain(
			req, bizutils.GetUIDOrZeroFromContext(ctx)),
	); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *dataPipelineControllerImpl) DeleteDataPipeline(
	ctx context.Context,
	req *dataflow.DataPipelineID,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := c.dataPipelineModel.DeleteDataPipeline(ctx, &domain.DeleteDataPipelineReq{
		ClusterID:  req.GetClusterId(),
		PipelineID: req.GetPipelineId(),
		UserID:     bizutils.GetUIDOrZeroFromContext(ctx),
	}); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *dataPipelineControllerImpl) ValidateDataPipelineSourceConnection(
	ctx context.Context,
	req *dataflow.ValidateDataPipelineSourceConnectionRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := c.dataPipelineModel.ValidateDataPipelineSourceConnection(
		ctx,
		&domain.ValidateDataPipelineSourceConnectionReq{
			ClusterID: req.GetClusterId(),
			UserID:    bizutils.GetUIDOrZeroFromContext(ctx),
			Username:  req.GetUsername(),
			Password:  req.GetPassword(),
		},
	); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *dataPipelineControllerImpl) ValidateDataPipelineStaging(
	ctx context.Context,
	req *dataflow.ValidateDataPipelineStagingRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := c.dataPipelineModel.ValidateDataPipelineStaging(
		ctx,
		&domain.ValidateDataPipelineStagingReq{
			ClusterID: req.GetClusterId(),
			UserID:    bizutils.GetUIDOrZeroFromContext(ctx),
			LakeOrgID: req.GetLakeOrgId(),
			Staging:   dataPipelineStagingInputDataflow2Domain(req.GetStaging()),
		},
	); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *dataPipelineControllerImpl) checkDataPipelineCluster(
	ctx context.Context,
	clusterID uint64,
) bizerr.BizError {
	if _, bizErr := c.mgmtSvcCli.GetClusterByID(ctx, clusterID); bizErr != nil {
		return bizErr.Chain("get cluster")
	}
	return nil
}

func invalidDataPipelineModelResponse(operation string) bizerr.BizError {
	return bizerr.New(bizerr.ErrCodeUnknown, "data pipeline model returned an empty %s response", operation)
}

func (c *dataPipelineControllerImpl) GetDataPipelineStagingTrust(
	ctx context.Context,
	req *dataflow.GetDataPipelineStagingTrustRequest,
) (*dataflow.DataPipelineStagingTrust, error) {
	ctx = log.WithTraceID(ctx)
	if bizErr := c.checkDataPipelineCluster(ctx, req.GetClusterId()); bizErr != nil {
		return nil, bizErr
	}
	resp, bizErr := c.dataPipelineModel.GetDataPipelineStagingTrust(
		ctx,
		&domain.GetDataPipelineStagingTrustReq{
			ClusterID: req.GetClusterId(),
			UserID:    bizutils.GetUIDOrZeroFromContext(ctx),
			LakeOrgID: req.GetLakeOrgId(),
		},
	)
	if bizErr != nil {
		return nil, bizErr
	}
	return dataPipelineStagingTrustDomain2Dataflow(resp), nil
}