package storage

import (
	"bytes"
	"fmt"
	"io/ioutil"
	"gitea.inhome.blapointe.com/local/storage/minio"
	"gitea.inhome.blapointe.com/local/storage/rclone"
	"gitea.inhome.blapointe.com/local/storage/resolve"
	"log"
	"net"
	"os"
	"path"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/google/uuid"
)

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

func (mock *mock) List(ns []string, limits ...string) ([]string, error) {
	namespace := resolve.Namespace(ns)
	limits = resolve.Limits(limits)
	keys := []string{}
	for k := range mock.m {
		if k >= limits[0] && k <= limits[1] {
			keys = append(keys, strings.TrimPrefix(k, namespace+"/"))
		}
	}
	return keys, nil
}

func (mock *mock) Get(key string, ns ...string) ([]byte, error) {
	namespace := resolve.Namespace(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 := resolve.Namespace(ns)
	mock.m[path.Join(namespace, key)] = value
	return nil
}

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

func TestImplementations(t *testing.T) {
	dir := path.Join(t.TempDir(), "storage_tests")
	if err := os.MkdirAll(dir, os.ModePerm); err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(dir)

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

	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 yamls, err := NewYaml(path.Join(dir, "yaml")); err != nil {
		t.Errorf("cannot make yaml: %v", err)
	} else {
		cases = append(cases, yamls)
	}

	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("MINIO"); ok {
		if minio, err := minio.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")
	}

	rcloneConfigPath := path.Join(t.TempDir(), "rclone.conf."+uuid.New().String())
	if err := ioutil.WriteFile(rcloneConfigPath, []byte(`
[local]
type = local
   `), os.ModePerm); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(path.Join(dir, "rclone"), os.ModePerm); err != nil {
		t.Fatal(err)
	}
	rclone, err := rclone.NewRClone(rcloneConfigPath, "local:"+path.Join(dir, "rclone"))
	if err != nil {
		t.Errorf("cannot make rclone: %v", err)
	} else {
		cases = append(cases, rclone)
	}

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

	for _, db := range cases {
		t.Run(fmt.Sprintf("%T", db), func(t *testing.T) {
			log.Printf("Trying %T", db)
			t.Logf("   %T: list @[ns1, ns2] against empty", db)
			if keys, err := db.List([]string{"ns1", "ns2"}); err != nil || len(keys) > 0 {
				t.Errorf("%T) cannot List() empty: (%T) %+v: %v", db, err, err, keys)
			}
			t.Logf("   %T: set %s @[ns1, ns2]", db, validKey)
			if err := db.Set(validKey, validValue, "ns1", "ns2"); err != nil {
				t.Errorf("%T) cannot set: %v", db, err)
			}
			t.Logf("   %T: db: %+v", db, db)
			t.Logf("   %T: get %s @[ns1, ns2]", db, validKey)
			if v, err := db.Get(validKey, "ns1", "ns2"); err != nil {
				t.Errorf("%T) cannot get: %v", db, err)
				log.Printf("%T) cannot get: %v (%+v)", db, err, db)
				time.Sleep(time.Second * 10)
			} 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 keys[0] != validKey {
				t.Errorf("%T) List()[0] != %s: %s: want %q, got %q", db, validKey, keys[0], 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 keys[0] != validKey {
				t.Errorf("%T) List(prefix)[0] != %s: %s", db, validKey, keys[0])
			}

			if dbstream, ok := db.(DBStream); ok {
				log.Printf("trying %T as DBStream", dbstream)
				raw := "raw"
				if err := dbstream.SetStream("k", strings.NewReader(raw), "ns1", "ns2"); err != nil {
					t.Errorf("%T) cannot setstream: %v", dbstream, err)
				}
				if r, err := dbstream.GetStream("k", "ns1", "ns2"); err != nil {
					t.Errorf("%T) cannot getstream: %v", dbstream, err)
				} else if b, err := ioutil.ReadAll(r); err != nil {
					t.Errorf("%T) cannot readall getstream: %v", dbstream, err)
				} else if string(b) != raw {
					t.Errorf("%T) wrong getstream: %v", dbstream, string(b))
				}
			}

			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: "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()
}