package storage

import (
	"fmt"
	"sync"
)

type KvStg interface {
	Set(keyPath []string, val string) error
	Get(keyPath []string) (val string, existed bool, err error)
	Delete(keyPath []string) error

	ListNextLevel(keyPath []string) (keys []string, err error)
	GetChild(keyPath []string) (KvStg, error)

	GetChunk(keyPath []string) (KvStgChunk, error)
}

type ErrKvStgGetSetOnNotLeaf struct {
	KeyPath []string
}

func (err *ErrKvStgGetSetOnNotLeaf) Error() string {
	return fmt.Sprintf("key path `%v` is not leaf node, can't Get/Set value", err.KeyPath)
}

type KvStgChunkItem struct {
	RelativeKeyPath []string
	Value           string
}

type KvStgChunk []KvStgChunkItem

type KvStgChunkObject map[string]string

func (chunk *KvStgChunk) ToNamedObjects(props []string) (map[string]KvStgChunkObject, error) {
	propSet := make(map[string]bool)
	for _, prop := range props {
		propSet[prop] = true
	}
	objects := make(map[string]KvStgChunkObject)
	for _, it := range *chunk {
		if len(it.RelativeKeyPath) != 2 {
			return nil, fmt.Errorf("KvStgChunk.ToNameObjects: chunk item `%v` key path is too long, should be 2 segments",
				it.RelativeKeyPath)
		}
		name := it.RelativeKeyPath[0]
		prop := it.RelativeKeyPath[1]
		if !propSet[prop] {
			return nil, fmt.Errorf("KvStgChunk.ToNameObjects: chunk item `%v` doesn't match schema `%v`",
				it.RelativeKeyPath, props)
		}
		object, ok := objects[name]
		if !ok {
			object = make(KvStgChunkObject)
		}
		if _, ok := object[prop]; ok {
			return nil, fmt.Errorf("KvStgChunk.ToNameObjects: chunk item `%v` prop `%v` duplicated with other item(s)",
				it.RelativeKeyPath, prop)
		}
		object[prop] = it.Value
	}
	return objects, nil
}

type MemKvStgNode struct {
	value    string
	children map[string]*MemKvStgNode
}

func (node *MemKvStgNode) iterate(parentPath []string, fn func(keyPath []string, val string)) {
	if node.children == nil {
		fn(parentPath, node.value)
	} else {
		for key, child := range node.children {
			child.iterate(append(parentPath, key), fn)
		}
	}
}

type MemKvStg struct {
	*MemKvStgNode
	locker *sync.Mutex
}

func NewMemKvStg() *MemKvStg {
	return &MemKvStg{
		MemKvStgNode: &MemKvStgNode{},
		locker:       &sync.Mutex{},
	}
}

func (mkv *MemKvStg) Set(keyPath []string, val string) error {
	mkv.locker.Lock()
	defer mkv.locker.Unlock()
	node := mkv.get(keyPath, true)
	if node == nil {
		return fmt.Errorf("MemKvStg.Set: `%v` create failed", keyPath)
	}
	if node.children != nil {
		return &ErrKvStgGetSetOnNotLeaf{KeyPath: keyPath}
	}
	node.value = val
	return nil
}

func (mkv *MemKvStg) Get(keyPath []string) (val string, existed bool, err error) {
	mkv.locker.Lock()
	defer mkv.locker.Unlock()
	node := mkv.get(keyPath, true)
	if node == nil {
		return "", false, fmt.Errorf("MemKvStg.Get: `%v` not found", keyPath)
	}
	if node.children != nil {
		return "", true, &ErrKvStgGetSetOnNotLeaf{KeyPath: keyPath}
	}
	return node.value, true, nil
}

func (mkv *MemKvStg) Delete(keyPath []string) error {
	mkv.locker.Lock()
	defer mkv.locker.Unlock()
	node := mkv.MemKvStgNode
	if len(keyPath) > 1 {
		node = mkv.get(keyPath[0:len(keyPath)-1], false)
	}
	if node != nil {
		delete(node.children, keyPath[len(keyPath)-1])
	}
	return nil
}

func (mkv *MemKvStg) ListNextLevel(keyPath []string) (keys []string, err error) {
	mkv.locker.Lock()
	defer mkv.locker.Unlock()
	node := mkv.get(keyPath, true)
	for key := range node.children {
		keys = append(keys, key)
	}
	return keys, nil
}

func (mkv *MemKvStg) GetChild(keyPath []string) (stg KvStg, err error) {
	mkv.locker.Lock()
	defer mkv.locker.Unlock()
	node := mkv.get(keyPath, false)
	if node == nil {
		return nil, fmt.Errorf("MemKvStg.GetChild: `%v` not found", keyPath)
	}
	stg = &MemKvStg{
		MemKvStgNode: node,
		locker:       mkv.locker,
	}
	return stg, nil
}

func (mkv *MemKvStg) GetChunk(keyPath []string) (chunk KvStgChunk, err error) {
	mkv.locker.Lock()
	defer mkv.locker.Unlock()
	node := mkv.get(keyPath, false)
	if node == nil {
		// return nil, fmt.Errorf("MemKvStg.GetChunk: `%v` not found", keyPath)
		return nil, nil
	}
	node.iterate([]string{}, func(keyPath []string, val string) {
		chunk = append(chunk, KvStgChunkItem{
			RelativeKeyPath: keyPath,
			Value:           val,
		})
	})
	return chunk, nil
}

func (mkv *MemKvStg) get(keyPath []string, allowCreate bool) *MemKvStgNode {
	var node *MemKvStgNode
	node = mkv.MemKvStgNode
	for _, key := range keyPath {
		if node.children == nil {
			if allowCreate {
				node.children = make(map[string]*MemKvStgNode)
			} else {
				return nil
			}
		}
		child, found := node.children[key]
		if !found {
			if allowCreate {
				child = &MemKvStgNode{}
				node.children[key] = child
			} else {
				return nil
			}
		}
		node = child
	}
	return node
}

type PrefixWrapKvStg struct {
	prefixKeyPath []string
	origin        KvStg
}

func NewPrefixWrapKvStg(prefixKeyPath []string, origin KvStg) *PrefixWrapKvStg {
	return &PrefixWrapKvStg{
		prefixKeyPath: prefixKeyPath,
		origin:        origin,
	}
}

func (stg *PrefixWrapKvStg) Set(keyPath []string, val string) error {
	return stg.origin.Set(stg.key(keyPath), val)
}

func (stg *PrefixWrapKvStg) Get(keyPath []string) (val string, existed bool, err error) {
	return stg.origin.Get(stg.key(keyPath))
}

func (stg *PrefixWrapKvStg) Delete(keyPath []string) error {
	return stg.origin.Delete(stg.key(keyPath))
}

func (stg *PrefixWrapKvStg) ListNextLevel(keyPath []string) (keys []string, err error) {
	return stg.origin.ListNextLevel(stg.key(keyPath))
}

func (stg *PrefixWrapKvStg) GetChild(keyPath []string) (KvStg, error) {
	return stg.origin.GetChild(stg.key(keyPath))
}

func (stg *PrefixWrapKvStg) GetChunk(keyPath []string) (KvStgChunk, error) {
	return stg.origin.GetChunk(stg.key(keyPath))
}

func (stg *PrefixWrapKvStg) key(keyPath []string) []string {
	return append(stg.prefixKeyPath, keyPath...)
}