package importmachine

import (
	"context"
	"fmt"
	"github.com/jinzhu/gorm"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/db/entity"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/i18n"
	"github.com/pingcap/tep/pkg/operator"
	"github.com/pingcap/tep/pkg/rbac"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	"github.com/pingcap/tep/pkg/utils"
	"github.com/pingcap/tep/pkg/utils/gormutil"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apiserver/pkg/endpoints/request"
	"k8s.io/apiserver/pkg/registry/rest"
	"k8s.io/client-go/util/retry"
	"time"
)

type Storage struct {
	store         *store.Store
	groupResource schema.GroupResource
}

func assembleMachine(machine *apiserver.Machine, node *corev1.Node) {
	machine.Kernel = node.Status.NodeInfo.KernelVersion
	machine.OS = node.Status.NodeInfo.OSImage
	machine.CPUCores = int32(node.Status.Capacity.Cpu().Value())
	machine.Memory = node.Status.Capacity.Memory().Value()

	for _, address := range node.Status.Addresses {
		if address.Type == corev1.NodeInternalIP {
			machine.IP = address.Address
		}
	}

	machine.Status = entity.MachineOffline.String()
	if val, ok := node.Labels[corev1.LabelHostname]; ok {
		machine.Hostname = val
	}
}

func (s *Storage) parseAndImportMachine(importMachine *apiserver.ImportMachine) error {
	db := s.store.DB
	var node *corev1.Node
	var machineEntity *entity.Machine
	var err error
	var errmsg string
	var az *entity.Az
	insertFn := func(tx *gorm.DB) error {
		for rowIndex, machine := range importMachine.Machines {
			if node, az, err = s.parseTableContentAndLabelNode(machine); err != nil {
				errmsg += fmt.Sprintf("row %d: failed to parse machine list, err: %v", rowIndex, err)
				continue
			}

			assembleMachine(&machine, node)

			machineEntity = s.store.Controller().AssembleMachineEntity(&machine)
			machineEntity.AzID = az.ID
			machineEntity.CreatedAt = time.Now()
			if err = tx.Create(machineEntity).Error; err != nil {
				errmsg += fmt.Sprintf("row %d: failed to create machine record, err: %v", rowIndex, err)
				continue
			}
			importMachine.Machines[rowIndex] = machine
		}
		if errmsg != "" {
			return errors.NewBadRequest(errmsg)
		}
		return nil
	}
	return gormutil.DoInTransaction(insertFn, db)
}

func (s *Storage) parseTableContentAndLabelNode(machine apiserver.Machine) (*corev1.Node, *entity.Az, error) {

	var node *corev1.Node
	var purpose string
	var az *entity.Az
	var err error

	if az, err = s.store.Controller().GetAzByName(machine.AZ, s.store.DB); err != nil {
		return nil, nil, err
	}

	purpose = machine.Purpose

	// Nodes update their heartbeat timestamp every ~10 seconds,
	// which makes it likely a get/update will hit a conflict error.
	// See：https://github.com/kubernetes/client-go/issues/414
	err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		var retryErr error

		if node, retryErr = s.store.Controller().GetNode(machine.Name); retryErr != nil {
			return retryErr
		}

		// add TEP label for node
		node.Labels[operator.TEPManagedLabel] = "true"

		if purpose == operator.TEPComputePurpose {
			node.Labels[operator.TEPComputeNodeLabel] = "true"
			if _, ok := node.Labels[operator.TEPStorageNodeLabel]; ok {
				delete(node.Labels, operator.TEPStorageNodeLabel)
			}
		} else if purpose == operator.TEPStoragePurpose {
			node.Labels[operator.TEPStorageNodeLabel] = "true"
			if _, ok := node.Labels[operator.TEPComputeNodeLabel]; ok {
				delete(node.Labels, operator.TEPComputeNodeLabel)
			}
		} else if purpose == operator.TEPBothPurpose {
			node.Labels[operator.TEPComputeNodeLabel] = "true"
			node.Labels[operator.TEPStorageNodeLabel] = "true"
		} else {
			return errors.NewBadRequest(fmt.Sprintf("invalid machine[%s] purpose %s", machine.Name, purpose))
		}

		if node, retryErr = s.store.Controller().UpdateNode(node); retryErr != nil {
			return retryErr
		}
		return nil
	})

	if err != nil {
		return nil, nil, err
	}

	return node, az, nil
}

func NewStorage(store *store.Store, groupResource schema.GroupResource) *Storage {
	return &Storage{
		store:         store,
		groupResource: groupResource,
	}
}

func (s *Storage) Kind() string {
	return "ImportMachine"
}

func (s *Storage) New() runtime.Object {
	return &apiserver.ImportMachine{}
}

func (s *Storage) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
	startTime := time.Now()
	operatorName := utils.GetAPIUsername(ctx)
	if !rbac.CanOperate(s.store.DB, operatorName, rbac.OtherMachineOperation) {
		return nil, errors.NewForbidden(s.groupResource, operatorName, fmt.Errorf("permission denied"))
	}
	var err, msg error
	var userID string
	if user, found := request.UserFrom(ctx); found {
		userID = user.GetUID()
	}

	importMachine := obj.(*apiserver.ImportMachine)

	if err = s.parseAndImportMachine(importMachine); err != nil {
		msg = fmt.Errorf("failed to parse or import machines, err: %s", err.Error())
		log.Error("", zap.Error(msg))
		return nil, apierrors.NewInternalError(err)
	}

	utils.AddEventAndLogError(s.store.DB, &utils.Event{
		UserID:         userID,
		Payload:        importMachine,
		OperationType:  entity.CreateOperation,
		ManagementType: entity.MachineManagement,
		CostTime:       time.Now().Sub(startTime).Nanoseconds(),
	})
	i18n.PlatformMessage.AddPlatformMessageAndLogError(s.store.DB, i18n.OperateMessageType, i18n.ImportMachine)
	return importMachine, nil
}

func (s *Storage) NamespaceScoped() bool {
	return false
}