package ffs

import (
	"context"
	"os"
	"strings"
	"time"

	"github.com/container-storage-interface/spec/lib/go/csi"
	"github.com/pingcap/log"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"k8s.io/kubernetes/pkg/volume"
	"k8s.io/utils/mount"
)

var _ = csi.NodeServer(&NodeServer{})

type NodeServer struct {
	Driver  *Driver
	mounter mount.Interface
}

func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {
	log.S().Infof("NodeStageVolume VolumeId %s StagingTargetPath %s", req.GetVolumeId(), req.GetStagingTargetPath())

	// Check arguments
	if req.GetVolumeCapability() == nil {
		return nil, status.Error(codes.InvalidArgument, "Volume capability missing in request")
	}
	if req.GetVolumeId() == "" {
		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
	}
	if req.GetStagingTargetPath() == "" {
		return nil, status.Error(codes.InvalidArgument, "Staging target path missing in request")
	}
	if !isValidVolumeCapabilities(ns.Driver.vcap, []*csi.VolumeCapability{req.GetVolumeCapability()}) {
		return nil, status.Error(codes.InvalidArgument, "Volume capability not supported")
	}

	volumeID := req.GetVolumeId()
	stagingTargetPath := req.GetStagingTargetPath()

	_, _, _, name, ok := ParseVolumeID(volumeID)
	if !ok {
		log.S().Errorf("Invalid volume ID in request %s", volumeID)
		return nil, status.Errorf(codes.InvalidArgument, "Invalid volume ID in request %s", volumeID)
	}

	ns.Driver.mutex.Lock()
	defer ns.Driver.mutex.Unlock()

	if _, err := os.Lstat(stagingTargetPath); err != nil && os.IsNotExist(err) {
		log.S().Infof("path %s does not exist", stagingTargetPath)
		err := makeDir(stagingTargetPath)
		if err != nil {
			return nil, status.Error(codes.Internal, err.Error())
		}
	}

	// check whether it can be mounted
	notMnt, err := checkMount(stagingTargetPath)
	if err != nil {
		log.S().Errorf("Check Mount volume %s to %s err %s", name, stagingTargetPath, err.Error())
		return nil, status.Error(codes.Internal, err.Error())
	}
	if !notMnt {
		return &csi.NodeStageVolumeResponse{}, nil
	}

	volContext := req.GetVolumeContext()
	v := newVolume(name, stagingTargetPath, volContext)
	mounter, err := ns.Driver.newMounter(v)
	if err != nil {
		return nil, err
	}
	if err := mounter.Mount(); err != nil {
		if os.IsPermission(err) {
			return nil, status.Error(codes.PermissionDenied, err.Error())
		}
		if strings.Contains(err.Error(), "invalid argument") {
			return nil, status.Error(codes.InvalidArgument, err.Error())
		}
		return nil, status.Error(codes.Internal, err.Error())
	}
	ns.Driver.stageVolume(v)
	log.S().Infof("volume %s has been staged.", name)
	return &csi.NodeStageVolumeResponse{}, nil
}
func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) {
	log.S().Infof("NodeUnstageVolume VolumeId %s StagingTargetPath %s", req.GetVolumeId(), req.GetStagingTargetPath())

	// Check arguments
	if req.GetVolumeId() == "" {
		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
	}
	if req.GetStagingTargetPath() == "" {
		return nil, status.Error(codes.InvalidArgument, "Staging target path missing in request")
	}

	volumeID := req.GetVolumeId()
	stagingTargetPath := req.GetStagingTargetPath()

	_, _, _, name, ok := ParseVolumeID(volumeID)
	if !ok {
		log.S().Errorf("Invalid volume ID in request %s", volumeID)
		return nil, status.Errorf(codes.InvalidArgument, "Invalid volume ID in request %s", volumeID)
	}

	ns.Driver.mutex.Lock()
	defer ns.Driver.mutex.Unlock()

	volume, err := ns.Driver.getVolume(name)
	if err != nil {
		return nil, err
	}
	if !volume.Published.Empty() {
		return nil, status.Errorf(codes.Internal, "volume %q is still published at %v", volume.Name, volume.Published)
	}

	if err := fuseUnmount(stagingTargetPath); err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	if err := os.Remove(stagingTargetPath); err != nil && !os.IsNotExist(err) {
		return nil, status.Error(codes.Internal, err.Error())
	}
	ns.Driver.unstageVolume(name)
	log.S().Infof("volume %s has been unstaged.", name)
	return &csi.NodeUnstageVolumeResponse{}, nil
}
func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
	log.S().Infof("NodePublishVolume VolumeId %s TargetPath %s", req.GetVolumeId(), req.GetTargetPath())

	// Check arguments
	if req.GetVolumeCapability() == nil {
		return nil, status.Error(codes.InvalidArgument, "Volume capability missing in request")
	}
	if !isValidVolumeCapabilities(ns.Driver.vcap, []*csi.VolumeCapability{req.GetVolumeCapability()}) {
		return nil, status.Error(codes.InvalidArgument, "Volume capability not supported")
	}
	if req.GetVolumeId() == "" {
		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
	}
	if req.GetStagingTargetPath() == "" {
		return nil, status.Error(codes.InvalidArgument, "Staging target path missing in request")
	}
	if req.GetTargetPath() == "" {
		return nil, status.Error(codes.InvalidArgument, "Target path missing in request")
	}

	volumeID := req.GetVolumeId()
	stagingTargetPath := req.GetStagingTargetPath()
	targetPath := req.GetTargetPath()

	_, _, _, name, ok := ParseVolumeID(volumeID)
	if !ok {
		log.S().Errorf("Invalid volume ID in request %s", volumeID)
		return nil, status.Errorf(codes.InvalidArgument, "Invalid volume ID in request %s", volumeID)
	}

	ns.Driver.mutex.Lock()
	defer ns.Driver.mutex.Unlock()

	notMnt, err := mount.IsNotMountPoint(mount.New(""), targetPath)
	if err != nil {
		if os.IsNotExist(err) {
			if err = os.Mkdir(targetPath, 0777); err != nil {
				return nil, status.Errorf(codes.Internal, "create target path: %v", err)
			}
			notMnt = true
		} else {
			return nil, status.Errorf(codes.Internal, "check target path: %v", err)
		}
	}
	if !notMnt {
		return &csi.NodePublishVolumeResponse{}, nil
	}
	options := []string{"bind"}
	mounter := mount.New("")
	if err := mounter.Mount(stagingTargetPath, targetPath, "", options); err != nil {
		return nil, status.Errorf(codes.Internal, "failed to mount %s at %s: %s", stagingTargetPath, targetPath, err.Error())
	}
	if err := ns.Driver.publishVolume(name, targetPath); err != nil {
		return nil, err
	}
	log.S().Infof("volume %s has been published.", name)
	return &csi.NodePublishVolumeResponse{}, nil
}
func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
	log.S().Infof("NodeUnpublishVolume VolumeId %s TargetPath %s", req.GetVolumeId(), req.GetTargetPath())
	if req.GetVolumeId() == "" {
		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
	}
	if req.GetTargetPath() == "" {
		return nil, status.Error(codes.InvalidArgument, "Target path missing in request")
	}

	volumeID := req.GetVolumeId()
	targetPath := req.GetTargetPath()

	_, _, _, name, ok := ParseVolumeID(volumeID)
	if !ok {
		log.S().Errorf("Invalid volume ID in request %s", volumeID)
		return nil, status.Errorf(codes.InvalidArgument, "Invalid volume ID in request %s", volumeID)
	}

	ns.Driver.mutex.Lock()
	defer ns.Driver.mutex.Unlock()

	// Unmount only if the target path is really a mount point.
	if !IsNotMountPoint(targetPath) {
		if err := waitForUnmount(targetPath, time.Second*10); err != nil {
			return nil, status.Errorf(codes.Internal, "wait for unmount: %s", err)
		}
		log.S().Infof("volume %s unmount %s", name, targetPath)
	}
	// Delete the mount point.
	if err := os.RemoveAll(targetPath); err != nil && !os.IsNotExist(err) {
		return nil, status.Errorf(codes.Internal, "remove target path: %s", err)
	}
	if err := ns.Driver.unpublishVolume(name, targetPath); err != nil {
		return nil, err
	}
	log.S().Infof("volume %s has been unpublished.", name)
	return &csi.NodeUnpublishVolumeResponse{}, nil
}
func (ns *NodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
	log.S().Infof("NodeGetVolumeStats VolumeId %s VolumePath %s", req.GetVolumeId(), req.GetVolumePath())

	if len(req.VolumeId) == 0 {
		return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume ID was empty")
	}
	if len(req.VolumePath) == 0 {
		return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume path was empty")
	}

	ns.Driver.mutex.Lock()
	defer ns.Driver.mutex.Unlock()

	if _, err := os.Lstat(req.VolumePath); err != nil {
		if os.IsNotExist(err) {
			return nil, status.Errorf(codes.NotFound, "path %s does not exist", req.VolumePath)
		}
		if ok, err := ns.mounter.IsLikelyNotMountPoint(req.VolumePath); ok || IsCorruptedMnt(err) {
			return nil, status.Errorf(codes.Internal, "is corrupted mnt %s: %v", req.VolumePath, err)
		}
		return nil, status.Errorf(codes.Internal, "failed to stat file %s: %v", req.VolumePath, err)
	}
	if ok, err := ns.mounter.IsLikelyNotMountPoint(req.VolumePath); ok || IsCorruptedMnt(err) {
		return nil, status.Errorf(codes.Internal, "is likely not mount point %s: %v", req.VolumePath, err)
	}

	volumeMetrics, err := volume.NewMetricsStatFS(req.VolumePath).GetMetrics()
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get metrics: %v", err)
	}

	available, ok := volumeMetrics.Available.AsInt64()
	if !ok {
		return nil, status.Errorf(codes.Internal, "failed to transform volume available size(%v)", volumeMetrics.Available)
	}
	capacity, ok := volumeMetrics.Capacity.AsInt64()
	if !ok {
		return nil, status.Errorf(codes.Internal, "failed to transform volume capacity size(%v)", volumeMetrics.Capacity)
	}
	used, ok := volumeMetrics.Used.AsInt64()
	if !ok {
		return nil, status.Errorf(codes.Internal, "failed to transform volume used size(%v)", volumeMetrics.Used)
	}

	inodesFree, ok := volumeMetrics.InodesFree.AsInt64()
	if !ok {
		return nil, status.Errorf(codes.Internal, "failed to transform disk inodes free(%v)", volumeMetrics.InodesFree)
	}
	inodes, ok := volumeMetrics.Inodes.AsInt64()
	if !ok {
		return nil, status.Errorf(codes.Internal, "failed to transform disk inodes(%v)", volumeMetrics.Inodes)
	}
	inodesUsed, ok := volumeMetrics.InodesUsed.AsInt64()
	if !ok {
		return nil, status.Errorf(codes.Internal, "failed to transform disk inodes used(%v)", volumeMetrics.InodesUsed)
	}

	return &csi.NodeGetVolumeStatsResponse{
		Usage: []*csi.VolumeUsage{
			{
				Unit:      csi.VolumeUsage_BYTES,
				Available: available,
				Total:     capacity,
				Used:      used,
			},
			{
				Unit:      csi.VolumeUsage_INODES,
				Available: inodesFree,
				Total:     inodes,
				Used:      inodesUsed,
			},
		},
	}, nil
}
func (*NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
	log.S().Infof("NodeExpandVolume VolumeId %s StagingTargetPath %s VolumePath %s", req.GetVolumeId(), req.GetStagingTargetPath(), req.GetVolumePath())

	return nil, status.Errorf(codes.Unimplemented, "method NodeExpandVolume not implemented")
}
func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) {
	log.S().Info("NodeGetCapabilities")

	log.S().Infof("Using default NodeGetCapabilities")

	return &csi.NodeGetCapabilitiesResponse{
		Capabilities: ns.Driver.nscap,
	}, nil
}
func (ns *NodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
	log.S().Info("NodeGetInfo")

	log.S().Infof("Using default NodeGetInfo: nodeID %s", ns.Driver.nodeID)

	return &csi.NodeGetInfoResponse{
		NodeId: ns.Driver.nodeID,
	}, nil
}

func checkMount(targetPath string) (bool, error) {
	m := mount.New("")
	notMnt, err := m.IsLikelyNotMountPoint(targetPath)
	if err != nil {
		if os.IsNotExist(err) {
			if err = os.MkdirAll(targetPath, 0777); err != nil {
				return false, err
			}
			notMnt = true
		} else {
			return false, err
		}
	}
	return notMnt, nil
}

func makeDir(pathname string) error {
	err := os.MkdirAll(pathname, os.FileMode(0777))
	if err != nil {
		if !os.IsExist(err) {
			return err
		}
	}
	return nil
}