// Copyright 2017 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package volume

import (
	"encoding/json"
	"fmt"
	"sort"

	yaml "gopkg.in/yaml.v2"

	"github.com/juju/errors"
	"github.com/ngaut/log"

	apiv1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
)

// ErrVolumeConflict may be returned when there are conflicting configurations
// matching the same node.
var ErrVolumeConflict = errors.New("conflicting volume configuration")

// Pool is an interface for getting tidb-volumes
type Pool interface {
	// Volumes returns tidb-volumes
	Volumes() (*TiDBVolume, error)
}

// Configurator will return the config map data.
type Configurator interface {
	Config() (data map[string]string, err error)
	Name() string
}

type volumePool struct {
	cfg Configurator
}

// NewVolumePool returns a volume Pool
func NewVolumePool(cfg Configurator) Pool {
	return &volumePool{cfg}
}

func (vp *volumePool) Volumes() (*TiDBVolume, error) {
	data, err := vp.cfg.Config()
	if err != nil {
		log.Errorf("ConfigMap[%s] get failed: %v", vp.cfg.Name(), err)
		return nil, errors.Trace(err)
	}

	tv := &TiDBVolume{}
	cmData, ok := data["volume"]
	if !ok {
		err := fmt.Errorf("ConfigMap[%s] has no volume", vp.cfg.Name())
		log.Error(err)
		return nil, err
	}

	if err := yaml.Unmarshal([]byte(cmData), tv); err != nil {
		log.Warnf("failed to unmarshal %s from yaml data %s: %v. will retry with json unmarshal", vp.cfg.Name(), cmData, err)
		err = json.Unmarshal([]byte(cmData), tv)
		if err != nil {
			log.Errorf("failed to unmarshal %s from json data %s: %v", vp.cfg.Name(), data, err)
			return nil, errors.Trace(err)
		}
	}

	return tv, nil
}

// TiDBVolume represents tidb cluster's volumes among all nodes
type TiDBVolume struct {
	// Nodes is parsed from tidb-volume ConfigMap
	Nodes []Node `json:"nodes,omitempty" yaml:"nodes,omitempty"`

	// Volumes is flatten as Volume come from Nodes
	Volumes []Volume `json:"volumes,omitempty" yaml:"volumes,omitempty"`
}

// Node represents a single volumes on a node
type Node struct {
	Name string `json:"name,omitempty" yaml:"name,omitempty"`

	// NodeSelector for matching on node labels instead of name.
	NodeSelector map[string]string `json:"nodeSelector,omitempty" yaml:"nodeSelector,omitempty"`

	Directories []Directory `json:"directories,omitempty" yaml:"directories,omitempty"`
}

// Directory represents a nodes's directory
type Directory struct {
	Path string `json:"path,omitempty" yaml:"path,omitempty"`
	Size string `json:"size,omitempty" yaml:"size,omitempty"`
}

// Volume represents a tidb-volume
type Volume struct {
	// NodeName is kubernetes's node name
	NodeName string

	// Path is the hostPath on the node
	Path string

	// Size is the capacity size of this volume
	Size resource.Quantity

	// Priority is used by tidb-scheduler for selecting a suitable volume
	Priority int32
}

// From intersects Volumes with available Nodes.
func (tv *TiDBVolume) From(allNodes []apiv1.Node) (*TiDBVolume, error) {
	// Node name matching have precedence over other matching methods.
	sort.Sort(byNodeNamePredecence(tv.Nodes))
	nameMatchSet := map[string]struct{}{}
	labelMatchSet := map[string]struct{}{}
	volumes := []Volume{}

Volumes:
	for _, tvNode := range tv.Nodes {
		nodeSelector := tvNode.NodeSelector != nil
		for _, node := range allNodes {
			name := node.GetName()
			switch {
			case nodeSelector:
				// A label match is one to many relationship. Expand into new volume.
				if labelMatches(tvNode, node) {
					// Skip nodes that already was matched by name to avoid duplicates.
					if _, ok := nameMatchSet[name]; ok {
						continue
					}
					// Conflict if multiple node selects matches the same node.
					if _, ok := labelMatchSet[name]; ok {
						return nil, ErrVolumeConflict
					}
					labelMatchSet[name] = struct{}{}
					volumes = appendDir(name, volumes, tvNode.Directories...)
				}
			default:
				// A name match is one to one relationship. Add volume as is.
				if nameMatches(tvNode, node) {
					// Conflict if multiple entries use the same node name.
					if _, ok := nameMatchSet[name]; ok {
						return nil, ErrVolumeConflict
					}
					nameMatchSet[name] = struct{}{}
					volumes = appendDir(name, volumes, tvNode.Directories...)
					// Only one node name can match.
					continue Volumes
				}
			}
		}
	}

	tv.Volumes = volumes
	log.Infof("After From volumes: %v", tv.Volumes)

	return tv, nil
}

func appendDir(nodeName string, volumes []Volume, dirs ...Directory) []Volume {
	for _, dir := range dirs {
		size, err := resource.ParseQuantity(dir.Size)
		if err != nil {
			log.Errorf(
				"failed to parse directory: %s:%s, size: [%s] %v, skipping",
				nodeName, dir.Path, dir.Size, err,
			)
			continue
		}
		volumes = append(volumes, Volume{
			NodeName: nodeName,
			Path:     dir.Path,
			Size:     size,
			Priority: 0,
		})
	}
	return volumes
}

// byNodeNamePredecence sorts node configurations that use node selector labels
// last.
type byNodeNamePredecence []Node

func (a byNodeNamePredecence) Len() int           { return len(a) }
func (a byNodeNamePredecence) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a byNodeNamePredecence) Less(i, j int) bool { return a[i].NodeSelector == nil }

func nameMatches(tvNode Node, node apiv1.Node) bool {
	return tvNode.Name == node.GetName()
}

func labelMatches(tvNode Node, node apiv1.Node) bool {
	if tvNode.NodeSelector == nil || len(tvNode.NodeSelector) == 0 {
		return false
	}
	for k, v := range tvNode.NodeSelector {
		if node.Labels[k] != v {
			return false
		}
	}
	return true
}

// Exclude remove alreadyProvVolumes from Volumes
func (tv *TiDBVolume) Exclude(vols []Volume) *TiDBVolume {
	idxs := make([]int, 0)

	for idx, volOut := range tv.Volumes {
		for _, volIn := range vols {
			if volOut.NodeName == volIn.NodeName && volOut.Path == volIn.Path {
				idxs = append(idxs, idx)
				break
			}
		}
	}
	tv.DeleteWithIdxs(idxs)

	log.Infof("After Exclude volumes: %v", tv.Volumes)

	return tv
}

// WithQuantity filters Volumes with pvcQuantity
func (tv *TiDBVolume) WithQuantity(pvcQuantity resource.Quantity) *TiDBVolume {
	idxs := make([]int, 0)

	for idx, vol := range tv.Volumes {
		if vol.Size.Cmp(pvcQuantity) == -1 {
			idxs = append(idxs, idx)
		}
	}
	tv.DeleteWithIdxs(idxs)

	log.Infof("After WithQuantity %s volumes: %v", pvcQuantity.String(), tv.Volumes)

	return tv
}

// DeleteWithIdxs delete delemets from TiDBVolume.Volumes
func (tv *TiDBVolume) DeleteWithIdxs(idxs []int) {
	if len(idxs) == 0 {
		return
	}

	newVols := make([]Volume, 0)

out:
	for idx, vol := range tv.Volumes {
		for _, i := range idxs {
			if idx == i {
				continue out
			}

		}
		newVols = append(newVols, vol)
	}

	tv.Volumes = newVols
}

// HighestPriorityVolume returns the highest priority volume
func (tv *TiDBVolume) HighestPriorityVolume() *Volume {
	if len(tv.Volumes) == 0 {
		return nil
	}

	sort.Slice(tv.Volumes, func(i, j int) bool {
		return tv.Volumes[i].Priority < tv.Volumes[j].Priority
	})

	return &tv.Volumes[len(tv.Volumes)-1]
}