package storage

import (
	"bytes"
	"io/ioutil"
	"log"
	"net"
	"os"
	"path"
	"strings"
	"sync"
	"testing"
)

type mock struct {
	m map[string][]byte
}

func (mock *mock) List(ns []string, limits ...string) ([]string, error) {
	limits = resolveLimits(limits)
	keys := []string{}
	for k := range mock.m {
		if k >= limits[0] && k <= limits[1] {
			keys = append(keys, k)
		}
	}
	return keys, nil
}

func (mock *mock) Get(key string, ns ...string) ([]byte, error) {
	namespace := resolveNamespace(ns)
	v, ok := mock.m[path.Join(namespace, key)]
	if ok {
		return v, nil
	}
	return nil, ErrNotFound
}

func (mock *mock) Set(key string, value []byte, ns ...string) error {
	namespace := resolveNamespace(ns)
	mock.m[path.Join(namespace, key)] = value
	return nil
}

func (mock *mock) Close() error {
	return nil
}

func TestImplementations(t *testing.T) {
	dir, err := ioutil.TempDir("", "storage_tests_")
	if err != nil {
		t.Fatalf("cannot create temp dir: %v", err)
		return
	}
	defer os.RemoveAll(dir)

	cases := make([]DB, 0)
	cases = append(cases, &mock{m: make(map[string][]byte)})
	cases = append(cases, NewMap())

	if cacheMem, err := NewCache(); err != nil {
		t.Errorf("cannot make cache/mem: %v", err)
	} else {
		cases = append(cases, cacheMem)
	}

	if cacheFile, err := NewCache(path.Join(dir, "cache")); err != nil {
		t.Errorf("cannot make cache/file: %v", err)
	} else {
		cases = append(cases, cacheFile)
	}

	if _, ok := os.LookupEnv("DYNOMITE"); ok {
		if dynomite, err := NewDynomite("localhost:8102", "", ""); err != nil {
			t.Logf("cannot make dynomite: %v", err)
		} else {
			cases = append(cases, dynomite)
		}
	} else {
		t.Log("$DYNOMITE not set. Skipping")
	}

	if _, ok := os.LookupEnv("REDIS"); ok {
		if redis, err := NewRedis("localhost:8103", "", ""); err != nil {
			t.Logf("cannot make redis: %v", err)
		} else {
			cases = append(cases, redis)
		}
	} else {
		t.Log("$REDIS not set. Skipping")
	}

	if bolt, err := NewBolt(path.Join(dir, "bolt")); err != nil {
		t.Errorf("cannot make bolt: %v", err)
	} else {
		cases = append(cases, bolt)
	}

	if files, err := NewFiles(path.Join(dir, "files")); err != nil {
		t.Errorf("cannot make files: %v", err)
	} else {
		cases = append(cases, files)
	}

	if leveldb, err := NewLevelDB(path.Join(dir, "leveldb")); err != nil {
		t.Errorf("cannot make leveldb: %v", err)
	} else {
		cases = append(cases, leveldb)
	}

	if _, ok := os.LookupEnv("MONGO"); ok {
		mongoLN, err := net.Listen("tcp", "localhost:27017")
		if err == nil {
			defer mongoLN.Close()
			go func() {
				for {
					conn, err := mongoLN.Accept()
					if err == nil {
						conn.Close()
					}
				}
			}()
		}
		if mongo1, err := NewMongo("localhost:27017"); err == nil {
			cases = append(cases, mongo1)
		} else if mongo2, err := NewMongo("localhost:27017", "root", "pass"); err == nil {
			cases = append(cases, mongo2)
		} else {
			t.Logf("cannot make mongo: %v", err)
		}
	} else {
		t.Log("$MONGO not set. Skipping")
	}

	if _, ok := os.LookupEnv("MEMCACHED"); ok {
		if memcache, err := NewMemcache("localhost:11211"); err != nil {
			t.Logf("cannot make memcache: %v", err)
		} else {
			cases = append(cases, memcache)
		}
	} else {
		t.Log("$MEMCACHED not set. Skipping")
	}

	if _, ok := os.LookupEnv("MEMCACHEDCLUSTER"); ok {
		if memcacheCluster, err := NewMemcacheCluster("localhost:11211"); err != nil {
			t.Logf("cannot make memcacheCluster: %v", err)
		} else {
			cases = append(cases, memcacheCluster)
		}
	} else {
		t.Log("$MEMCACHEDCLUSTER not set. Skipping")
	}

	if _, ok := os.LookupEnv("MINIO"); ok {
		if minio, err := NewMinio("localhost:9000", "accesskey", "secretkey"); err != nil {
			t.Logf("cannot make minio: %v", err)
		} else {
			cases = append(cases, minio)
		}
	} else {
		t.Log("$MINIO not set. Skipping")
	}

	f, err := ioutil.TempFile(os.TempDir(), "rclone.conf.*")
	if err != nil {
		t.Fatal(err)
	}
	defer os.Remove(f.Name())
	f.Write([]byte(`
[local]
type = local
   `))
	f.Close()
	rclone, err := NewRClone(f.Name(), "local:/tmp")
	if err != nil {
		t.Errorf("cannot make rclone: %v", err)
	} else {
		cases = append(cases, rclone)
	}

	validKey := "key"
	validValue := []byte("value")

	for _, db := range cases {
		log.Printf("Trying %T", db)
		t.Logf("   %T: set", db)
		if err := db.Set(validKey, validValue, "ns1", "ns2"); err != nil {
			t.Errorf("%T) cannot set: %v", db, err)
		}
		t.Logf("   %T: get", db)
		if v, err := db.Get(validKey, "ns1", "ns2"); err != nil {
			t.Errorf("%T) cannot get: %v", db, err)
		} else if !bytes.Equal(v, validValue) {
			t.Errorf("%T) wrong get: %q vs %q", db, v, validValue)
		}
		t.Logf("   %T: list1", db)
		if keys, err := db.List([]string{"ns1", "ns2"}); err != nil || len(keys) < 1 {
			t.Errorf("%T) cannot List(): %v", db, err)
		} else if !strings.Contains(keys[0], validKey) {
			t.Errorf("%T) List()[0] != %s: %s", db, validKey, keys[0])
		}
		t.Logf("   %T: list2", db)
		if keys, err := db.List([]string{"ns1", "ns2"}, validKey[:1]); err != nil || len(keys) < 1 {
			t.Errorf("%T) cannot List(prefix): %v", db, err)
		} else if !strings.Contains(keys[0], validKey) {
			t.Errorf("%T) List(prefix)[0] != %s: %s", db, validKey, keys[0])
		}
		t.Logf("   %T: close", db)
		if err := db.Close(); err != nil {
			t.Errorf("cannot close %T: %v", db, err)
		}
	}
}

func TestToFromString(t *testing.T) {
	cases := []struct {
		key string
		t   Type
	}{
		{
			key: "map",
			t:   MAP,
		},
		{
			key: "redis",
			t:   REDIS,
		},
		{
			key: "minio",
			t:   MINIO,
		},
		{
			key: "bolt",
			t:   BOLT,
		},
		{
			key: "cache",
			t:   CACHE,
		},
		{
			key: "leveldb",
			t:   LEVELDB,
		},
		{
			key: "memcache",
			t:   MEMCACHE,
		},
		{
			key: "memcachecluster",
			t:   MEMCACHECLUSTER,
		},
		{
			key: "mongo",
			t:   MONGO,
		},
	}

	for _, c := range cases {
		if TypeFromString(c.key) != c.t {
			t.Errorf("wrong type for %v: got %v", c.key, TypeFromString(c.key))
		}
		if c.key != c.t.String() {
			t.Errorf("wrong string for %v (%v): got %v", int(c.t), c.key, c.t.String())
		}
	}
}

func TestNew(t *testing.T) {
	dir, err := ioutil.TempDir("", "storage_tests_")
	if err != nil {
		t.Fatalf("cannot create temp dir: %v", err)
		return
	}
	defer os.RemoveAll(dir)

	if b, err := New(BOLT); err == nil {
		t.Errorf("can create bolt without path")
		b.Close()
	}

	if b, err := New(BOLT, path.Join(dir, "bolt")); err != nil {
		t.Errorf("cannot create bolt with path")
	} else {
		b.Close()
	}
}

func recoverDeferred(c Type, t *testing.T, wg *sync.WaitGroup) {
	if err := recover(); err != nil {
		log.Printf("recover deferre fail: %s", c)
		t.Errorf("[%s] panic: %v", c, err)
	} else {
		log.Printf("recover deferre ok: %s", c)
	}
	defer wg.Done()
}

func TestResolveLimitsStart(t *testing.T) {
	cases := map[string]struct {
		in  string
		out string
	}{
		"explicit": {in: "no", out: "no"},
	}

	for name, c := range cases {
		input := strings.Split(c.in, ",")
		out := resolveLimitsStart(input)
		if out != c.out {
			t.Errorf("%v: got %v, want %v from %v", name, out, c.out, c.in)
		}
	}
}

func TestResolveLimitsStop(t *testing.T) {
	cases := map[string]struct {
		in  string
		out string
	}{
		"short arr": {in: "", out: "}}}}}}"},
		"explicit":  {in: ",no", out: "no"},
	}

	for name, c := range cases {
		input := strings.Split(c.in, ",")
		out := resolveLimitsStop(input)
		if out != c.out {
			t.Errorf("%v: got %v, want %v from %v", name, out, c.out, c.in)
		}
	}
}

func TestResolveLimitsLimit(t *testing.T) {
	cases := map[string]struct {
		in  string
		out string
	}{
		"15":        {in: ",,15", out: "15"},
		"0 set":     {in: ",,0", out: "0"},
		"0 default": {in: ",,", out: "0"},
	}

	for name, c := range cases {
		input := strings.Split(c.in, ",")
		out := resolveLimitsLimit(input)
		if out != c.out {
			t.Errorf("%v: got %v, want %v from %v", name, out, c.out, c.in)
		}
	}
}

func TestResolveLimitsAscending(t *testing.T) {
	cases := map[string]struct {
		in  string
		out string
	}{
		"desc":              {in: "a,b,c,-", out: "false"},
		"asc":               {in: "a,b,c,+", out: "true"},
		"default asc prev":  {in: "a,b,c,", out: "true"},
		"default asc empty": {in: ",,,", out: "true"},
	}

	for name, c := range cases {
		input := strings.Split(c.in, ",")
		out := resolveLimitsAscending(input)
		if out != c.out {
			t.Errorf("%v: got %v, want %v from %v", name, out, c.out, c.in)
		}
	}
}