package pimport

import (
	"context"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/internal/model/cluster"
	pimport "github.com/tidbcloud/dataflow-service/internal/model/import"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/dataflow-service/pkg/account/accountdb"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
)

type importInnerService struct {
	dataflow.UnimplementedImportInnerServiceServer
	importModel   pimport.Model
	clustersModel cluster.Model
	importSvc     *importService
}

// NewInternalService constructs a importInnerService
func NewInternalService(model pimport.Model, clustersModel cluster.Model) dataflow.ImportInnerServiceServer {
	importService := &importService{importModel: model, clustersModel: clustersModel}
	return &importInnerService{importModel: model, clustersModel: clustersModel, importSvc: importService}
}

func (s *importInnerService) List(ctx context.Context, req *dataflow.ListImportReq) (*dataflow.ListImportResp, error) {
	var err error
	logger := log.WithContextL(ctx)
	logger.Info("inner list import", zap.Reflect("req", req))

	cluster, err := grpcli.GetClusterByIDAndExcludeStatuses(ctx, req.ClusterId, nil)
	if err != nil {
		return nil, err
	}
	if cluster.TenantId != req.OrgId || cluster.ProjectId != req.ProjectId {
		return nil, errcode.NewForbiddenErr(errors.New("you do not have permission to access this resource"))
	}

	pager := utils.AdjustedPager(&req.Page, &req.PageSize)
	logger.Info("adjusted pager", zap.Reflect("pager", pager))

	statuses := pbStatusList2ImportStatusList(req.GetStatuses())
	clusterID := entity.ID(req.GetClusterId())
	totalCount, importEntities, err := pimport.ListImport(ctx, clusterID, pager, statuses, req.GetCreatedByDm())
	if err != nil {
		return nil, errors.Annotate(err, "list import entities failed")
	}
	return importEntitiesToListRespPB(ctx, clusterID, totalCount, importEntities, logger)
}

func userCtxWithAuth(ctx context.Context) (context.Context, error) {
	clusterID, _ := ctx.Value(utils.CtxClusterIDKey).(uint64)
	projectID, _ := ctx.Value(utils.CtxProjectIDKey).(uint64)
	orgID, _ := ctx.Value(utils.CtxOrgIDKey).(uint64)
	cluster, err := grpcli.GetClusterByIDAndExcludeStatuses(ctx, clusterID, nil)
	if err != nil {
		return ctx, err
	}
	if cluster.TenantId != orgID || cluster.ProjectId != projectID {
		return ctx, errors.New("you do not have permission to access this resource")
	}
	u, _ := accountdb.GetUserByID(cluster.CreatedBy)
	if u == nil {
		return ctx, errors.New("failed to get user")
	}
	user, flag, er := accountdb.GetUserInOrg(u.Email, orgID, true)
	if !flag || er != nil {
		return ctx, errors.New("unknown user")
	}
	return context.WithValue(ctx, utils.CtxInnerCallerKey, user), nil
}

func (s *importInnerService) Create(ctx context.Context, req *dataflow.ImportCreateReq) (*dataflow.ClusterImportId, error) {
	logger := log.WithContextS(ctx)
	ctx = context.WithValue(ctx, utils.CtxClusterIDKey, req.ClusterId)
	ctx = context.WithValue(ctx, utils.CtxProjectIDKey, req.ProjectId)
	ctx = context.WithValue(ctx, utils.CtxOrgIDKey, req.OrgId)
	var err error
	if ctx, err = userCtxWithAuth(ctx); err != nil {
		logger.Error("build import error", zap.Error(err))
		return nil, wrapNonErrorCodeAsInternalError(err, "inner import caller auth failed.")
	}
	return s.importSvc.Create(ctx, req)
}

func (s *importInnerService) Validate(ctx context.Context, req *dataflow.ImportValidateReq) (*emptypb.Empty, error) {
	logger := log.WithContextS(ctx)
	ctx = context.WithValue(ctx, utils.CtxClusterIDKey, req.ClusterId)
	ctx = context.WithValue(ctx, utils.CtxProjectIDKey, req.ProjectId)
	ctx = context.WithValue(ctx, utils.CtxOrgIDKey, req.OrgId)
	var err error
	if ctx, err = userCtxWithAuth(ctx); err != nil {
		logger.Error("build import error", zap.Error(err))
		return nil, wrapNonErrorCodeAsInternalError(err, "inner import caller auth failed.")
	}
	return s.importSvc.Validate(ctx, req)
}

func (s *importInnerService) Get(ctx context.Context, req *dataflow.ClusterImportId) (*dataflow.ImportGetResp, error) {
	logger := log.WithContextS(ctx)
	ctx = context.WithValue(ctx, utils.CtxClusterIDKey, req.ClusterId)
	ctx = context.WithValue(ctx, utils.CtxProjectIDKey, req.ProjectId)
	ctx = context.WithValue(ctx, utils.CtxOrgIDKey, req.OrgId)
	var err error
	if ctx, err = userCtxWithAuth(ctx); err != nil {
		logger.Error("build import error", zap.Error(err))
		return nil, wrapNonErrorCodeAsInternalError(err, "inner import caller auth failed.")
	}
	return s.importSvc.Get(ctx, req)
}

func (s *importInnerService) Cancel(ctx context.Context, req *dataflow.ClusterImportId) (*emptypb.Empty, error) {
	logger := log.WithContextS(ctx)
	ctx = context.WithValue(ctx, utils.CtxClusterIDKey, req.ClusterId)
	ctx = context.WithValue(ctx, utils.CtxProjectIDKey, req.ProjectId)
	ctx = context.WithValue(ctx, utils.CtxOrgIDKey, req.OrgId)
	var err error
	if ctx, err = userCtxWithAuth(ctx); err != nil {
		logger.Error("build import error", zap.Error(err))
		return nil, wrapNonErrorCodeAsInternalError(err, "inner import caller auth failed.")
	}
	return s.importSvc.Cancel(ctx, req)
}

func (s *importInnerService) GetS3BucketRegion(ctx context.Context, req *dataflow.GetS3BucketRegionReq) (*dataflow.GetS3BucketRegionResp, error) {
	ctx = context.WithValue(ctx, utils.CtxInnerCallerKey, new(accountdb.User))
	return s.importSvc.GetS3BucketRegion(ctx, req)
}

func (s *importInnerService) Exit(ctx context.Context, req *dataflow.ClusterImportId) (*emptypb.Empty, error) {
	logger := log.WithContextS(ctx)
	ctx = context.WithValue(ctx, utils.CtxClusterIDKey, req.ClusterId)
	ctx = context.WithValue(ctx, utils.CtxProjectIDKey, req.ProjectId)
	ctx = context.WithValue(ctx, utils.CtxOrgIDKey, req.OrgId)
	var err error
	if ctx, err = userCtxWithAuth(ctx); err != nil {
		logger.Error("build import error", zap.Error(err))
		return nil, wrapNonErrorCodeAsInternalError(err, "inner import caller auth failed.")
	}
	return s.importSvc.Exit(ctx, req)
}