package storage

import (
	"fmt"
)

// scope is execTask id and id is actionTask id
type DagStatusStg interface {
	SetDone(scope, id string) error
	IsDone(scope, id string) (bool, error)
	SetError(scope string, id string, actionErr error) error
}

type KvBasedDagStatusStg struct {
	kvs KvStg
}

func NewKvBasedDagStatusStg(kvs KvStg) *KvBasedDagStatusStg {
	return &KvBasedDagStatusStg{
		kvs: kvs,
	}
}

func (stg *KvBasedDagStatusStg) SetDone(scope, id string) error {
	return stg.kvs.Set(stg.key(scope, id), Bool2Val(true))
}

func (stg *KvBasedDagStatusStg) IsDone(scope, id string) (bool, error) {
	val, existed, err := stg.kvs.Get(stg.key(scope, id))
	if !existed {
		return false, nil
	}
	return Val2Bool(val), err
}

func (stg *KvBasedDagStatusStg) SetError(scope, id string, actionErr error) error {
	return nil
}

func (stg *KvBasedDagStatusStg) key(scope, key string) []string {
	return []string{string(DagExecKeyTypeDone), scope, key}
}

type DagExecKeyType string

const (
	DagExecKeyTypeDone DagExecKeyType = "done"
)

type CachedDagStatusStg struct {
	origin DagStatusStg
	cache  map[string]bool
}

func NewCachedDagStatusStg(origin DagStatusStg) *CachedDagStatusStg {
	return &CachedDagStatusStg{
		origin: origin,
		cache:  make(map[string]bool),
	}
}

func (stg *CachedDagStatusStg) SetDone(scope, id string) error {
	err := stg.origin.SetDone(scope, id)
	if err != nil {
		return err
	}
	stg.cache[stg.key(scope, id)] = true
	return nil
}

func (stg *CachedDagStatusStg) IsDone(scope, id string) (done bool, err error) {
	done, existed := stg.cache[stg.key(scope, id)]
	if existed {
		return done, nil
	}
	done, err = stg.origin.IsDone(scope, id)
	if err == nil {
		stg.cache[stg.key(scope, id)] = done
	}
	return done, err
}

func (stg *CachedDagStatusStg) SetError(scope, id string, actionErr error) error {
	return stg.origin.SetError(scope, id, actionErr)
}

func (stg *CachedDagStatusStg) key(scope, id string) string {
	return fmt.Sprintf("scope:%s;id:%s", scope, id)
}