// TestMultiplePersistConnection tests the second connection to a
// BoltDB fails when one is already open with PersistConnection flag
func TestMultiplePersistConnection(t *testing.T) {
	kv, err := libkv.NewStore(
		store.BOLTDB,
		[]string{"/tmp/not_exist_dir/__boltdbtest"},
		&store.Config{
			Bucket:            "boltDBTest",
			ConnectionTimeout: 1 * time.Second,
			PersistConnection: true},
	)
	assert.NoError(t, err)
	assert.NotNil(t, kv)

	if _, ok := kv.(*BoltDB); !ok {
		t.Fatal("Error registering and initializing boltDB")
	}

	// Must fail if multiple boltdb requests are made with a valid timeout
	kv, err = libkv.NewStore(
		store.BOLTDB,
		[]string{"/tmp/not_exist_dir/__boltdbtest"},
		&store.Config{
			Bucket:            "boltDBTest",
			ConnectionTimeout: 1 * time.Second,
			PersistConnection: true},
	)
	assert.Error(t, err)

	_ = os.Remove("/tmp/not_exist_dir/__boltdbtest")
}

// TestConcurrentConnection tests simultaenous get/put using
// two handles.
func TestConcurrentConnection(t *testing.T) {
	var err error
	kv1, err1 := libkv.NewStore(
		store.BOLTDB,
		[]string{"/tmp/__boltdbtest"},
		&store.Config{
			Bucket:            "boltDBTest",
			ConnectionTimeout: 1 * time.Second},
	)
	assert.NoError(t, err1)
	assert.NotNil(t, kv1)

	kv2, err2 := libkv.NewStore(
		store.BOLTDB,
		[]string{"/tmp/__boltdbtest"},
		&store.Config{Bucket: "boltDBTest",
			ConnectionTimeout: 1 * time.Second},
	)
	assert.NoError(t, err2)
	assert.NotNil(t, kv2)

	key1 := "TestKV1"
	value1 := []byte("TestVal1")
	err = kv1.Put(key1, value1, nil)
	assert.NoError(t, err)

	key2 := "TestKV2"
	value2 := []byte("TestVal2")
	err = kv2.Put(key2, value2, nil)
	assert.NoError(t, err)

	pair1, err1 := kv1.Get(key1)
	assert.NoError(t, err)
	if assert.NotNil(t, pair1) {
		assert.NotNil(t, pair1.Value)
	}
	assert.Equal(t, pair1.Value, value1)

	pair2, err2 := kv2.Get(key2)
	assert.NoError(t, err)
	if assert.NotNil(t, pair2) {
		assert.NotNil(t, pair2.Value)
	}
	assert.Equal(t, pair2.Value, value2)

	// AtomicPut using kv1 and kv2 should succeed
	_, _, err = kv1.AtomicPut(key1, []byte("TestnewVal1"), pair1, nil)
	assert.NoError(t, err)

	_, _, err = kv2.AtomicPut(key2, []byte("TestnewVal2"), pair2, nil)
	assert.NoError(t, err)

	testutils.RunTestCommon(t, kv1)
	testutils.RunTestCommon(t, kv2)

	kv1.Close()
	kv2.Close()

	_ = os.Remove("/tmp/__boltdbtest")
}

func NewStore(machines []string, cmd *ServerCommand, keyspace string) *Storage {
	etcd.Register()

	clt, err := libkv.NewStore(store.ETCD, cmd.Config.BackendMachines, &store.Config{})
	if err != nil {
		panic(err)
	}

	_, err = clt.List(keyspace)
	if err != store.ErrKeyNotFound && err != nil {
		log.WithError(err).Fatal("store: Store backend not reachable")
	}

	cfg := client.Config{
		Endpoints: machines,
		Transport: client.DefaultTransport,
		// set timeout per request to fail fast when the target endpoint is unavailable
		HeaderTimeoutPerRequest: time.Second,
	}
	c, err := client.New(cfg)
	if err != nil {
		log.Fatal(err)
	}
	kapi := client.NewKeysAPI(c)

	return &Storage{Client: clt, Kapi: kapi, command: cmd, keyspace: keyspace}
}

func (s *EtcdSuite) SetUpTest(c *check.C) {
	s.createComposeProject(c, "etcd")
	s.composeProject.Start(c)

	etcd.Register()
	url := s.composeProject.Container(c, "etcd").NetworkSettings.IPAddress + ":2379"
	kv, err := libkv.NewStore(
		store.ETCD,
		[]string{url},
		&store.Config{
			ConnectionTimeout: 10 * time.Second,
		},
	)
	if err != nil {
		c.Fatal("Cannot create store etcd")
	}
	s.kv = kv

	// wait for etcd
	err = utils.Try(60*time.Second, func() error {
		_, err := kv.Exists("test")
		if err != nil {
			return fmt.Errorf("Etcd connection error to %s: %v", url, err)
		}
		return nil
	})
	c.Assert(err, checker.IsNil)
}

func NewStore(backend string, machines []string, a *AgentCommand, keyspace string) *Store {
	store, err := libkv.NewStore(store.Backend(backend), machines, nil)
	if err != nil {
		log.Fatal(err)
	}
	return &Store{Client: store, agent: a, keyspace: keyspace}
}

func getKVStore(addr string, options *kvstore.Config) (kvstore.Store, error) {
	u, err := url.Parse(addr)
	if err != nil {
		return nil, err

	}

	kvType := strings.ToLower(u.Scheme)
	kvHost := u.Host
	var backend kvstore.Backend

	switch kvType {
	case "consul":
		backend = kvstore.CONSUL
	case "etcd":
		backend = kvstore.ETCD
	}

	kv, err := libkv.NewStore(
		backend,
		[]string{kvHost},
		options,
	)

	if err != nil {
		return nil, err

	}

	return kv, nil
}

// Initialize is exported
func (s *Discovery) Initialize(uris string, heartbeat time.Duration, ttl time.Duration) error {
	var (
		parts  = strings.SplitN(uris, "/", 2)
		addrs  = strings.Split(parts[0], ",")
		prefix = ""
		err    error
	)

	// A custom prefix to the path can be optionally used.
	if len(parts) == 2 {
		prefix = parts[1]
	}

	s.heartbeat = heartbeat
	s.ttl = ttl
	s.path = path.Join(prefix, discoveryPath)

	// Creates a new store, will ignore options given
	// if not supported by the chosen store
	s.store, err = libkv.NewStore(
		s.backend,
		addrs,
		&store.Config{
			EphemeralTTL: s.ttl,
		},
	)

	return err
}

func (s *ConsulSuite) setupConsul(c *check.C) {
	s.createComposeProject(c, "consul")
	s.composeProject.Start(c)

	consul.Register()
	kv, err := libkv.NewStore(
		store.CONSUL,
		[]string{s.composeProject.Container(c, "consul").NetworkSettings.IPAddress + ":8500"},
		&store.Config{
			ConnectionTimeout: 10 * time.Second,
		},
	)
	if err != nil {
		c.Fatal("Cannot create store consul")
	}
	s.kv = kv

	// wait for consul
	err = utils.Try(60*time.Second, func() error {
		_, err := kv.Exists("test")
		if err != nil {
			return err
		}
		return nil
	})
	c.Assert(err, checker.IsNil)
}

func NewBgpRouteManager(masterIface string, server net.IP, as string) *BgpRouteManager {
	b := &BgpRouteManager{
		ethIface:     masterIface,
		server:       server,
		neighborlist: map[string]string{},
		asnum:        as,
	}
	consul.Register()
	client := "localhost:8500"
	kv, err := libkv.NewStore(
		store.CONSUL, // or "consul"
		[]string{client},
		&store.Config{
			ConnectionTimeout: 10 * time.Second,
		},
	)
	b.kv = kv
	if err != nil {
		log.Fatal("Cannot create store consul")
	}
	b.neighborkey = "bgpneighbor"
	if exist, _ := kv.Exists(b.neighborkey); exist == false {
		err := b.kv.Put(b.neighborkey, []byte("bgpneighbor"), &store.WriteOptions{IsDir: true})
		if err != nil {
			fmt.Errorf("Something went wrong when initializing key %v", b.neighborkey)
		}
	}
	err = b.kv.Put(b.neighborkey+"/"+server.String(), []byte(as), nil)
	if err != nil {
		log.Errorf("Error trying to put value at key: %v", b.neighborkey)
	}

	return b
}

func NewStore(backend Backend, addrsStr string) (kvstore.Store, error) {

	var kvbackend kvstore.Backend
	switch backend {
	case CONSUL:
		kvbackend = kvstore.CONSUL
	case ETCD:
		kvbackend = kvstore.ETCD
	default:
		return nil, fmt.Errorf("Unknown store backend: %q", backend)
	}

	if addrsStr == "" {
		switch backend {
		case CONSUL:
			addrsStr = DefaultConsulEndpoints
		case ETCD:
			addrsStr = DefaultEtcdEndpoints
		}
	}
	addrs := strings.Split(addrsStr, ",")

	store, err := libkv.NewStore(kvbackend, addrs, &kvstore.Config{ConnectionTimeout: 10 * time.Second})
	if err != nil {
		return nil, err
	}
	return store, nil
}

// newClient used to connect to KV Store
func newClient(kv string, addrs string) (DataStore, error) {
	store, err := libkv.NewStore(store.Backend(kv), []string{addrs}, &store.Config{})
	if err != nil {
		return nil, err
	}
	ds := &datastore{store: store}
	return ds, nil
}

func NewStore(storeUrl string) (store.Store, error) {
	kv, addrs := parseStoreUrl(storeUrl)
	config := &store.Config{}

	st, err := libkv.NewStore(store.Backend(kv), addrs, config)
	if err != nil {
		return nil, err
	}

	return st, nil
}

func (provider *Kv) provide(configurationChan chan<- types.ConfigMessage) error {
	storeConfig := &store.Config{
		ConnectionTimeout: 30 * time.Second,
		Bucket:            "traefik",
	}

	if provider.TLS != nil {
		caPool := x509.NewCertPool()

		if provider.TLS.CA != "" {
			ca, err := ioutil.ReadFile(provider.TLS.CA)

			if err != nil {
				return fmt.Errorf("Failed to read CA. %s", err)
			}

			caPool.AppendCertsFromPEM(ca)
		}

		cert, err := tls.LoadX509KeyPair(provider.TLS.Cert, provider.TLS.Key)

		if err != nil {
			return fmt.Errorf("Failed to load keypair. %s", err)
		}

		storeConfig.TLS = &tls.Config{
			Certificates:       []tls.Certificate{cert},
			RootCAs:            caPool,
			InsecureSkipVerify: provider.TLS.InsecureSkipVerify,
		}
	}

	kv, err := libkv.NewStore(
		provider.storeType,
		strings.Split(provider.Endpoint, ","),
		storeConfig,
	)
	if err != nil {
		return err
	}
	if _, err := kv.List(""); err != nil {
		return err
	}
	provider.kvclient = kv
	if provider.Watch {
		go provider.watchKv(configurationChan, provider.Prefix)
	}
	configuration := provider.loadConfig()
	configurationChan <- types.ConfigMessage{
		ProviderName:  string(provider.storeType),
		Configuration: configuration,
	}
	return nil
}

func TestRegister(t *testing.T) {
	Register()

	kv, err := libkv.NewStore(store.ETCD, []string{client}, nil)
	assert.NoError(t, err)
	assert.NotNil(t, kv)

	if _, ok := kv.(*Etcd); !ok {
		t.Fatal("Error registering and initializing etcd")
	}
}

func TestRegister(t *testing.T) {
	Register()

	kv, err := libkv.NewStore(store.CONSUL, []string{client}, nil)
	assert.NoError(t, err)
	assert.NotNil(t, kv)

	if _, ok := kv.(*Consul); !ok {
		t.Fatal("Error registering and initializing consul")
	}
}

// Initialize is exported
func (s *Discovery) Initialize(uris string, heartbeat time.Duration, ttl time.Duration, clusterOpts map[string]string) error {
	var (
		parts = strings.SplitN(uris, "/", 2)
		addrs = strings.Split(parts[0], ",")
		err   error
	)

	// A custom prefix to the path can be optionally used.
	if len(parts) == 2 {
		s.prefix = parts[1]
	}

	s.heartbeat = heartbeat
	s.ttl = ttl

	// Use a custom path if specified in discovery options
	dpath := defaultDiscoveryPath
	if clusterOpts["kv.path"] != "" {
		dpath = clusterOpts["kv.path"]
	}

	s.path = path.Join(s.prefix, dpath)

	var config *store.Config
	if clusterOpts["kv.cacertfile"] != "" && clusterOpts["kv.certfile"] != "" && clusterOpts["kv.keyfile"] != "" {
		logrus.Info("Initializing discovery with TLS")
		tlsConfig, err := tlsconfig.Client(tlsconfig.Options{
			CAFile:   clusterOpts["kv.cacertfile"],
			CertFile: clusterOpts["kv.certfile"],
			KeyFile:  clusterOpts["kv.keyfile"],
		})
		if err != nil {
			return err
		}
		config = &store.Config{
			// Set ClientTLS to trigger https (bug in libkv/etcd)
			ClientTLS: &store.ClientTLSConfig{
				CACertFile: clusterOpts["kv.cacertfile"],
				CertFile:   clusterOpts["kv.certfile"],
				KeyFile:    clusterOpts["kv.keyfile"],
			},
			// The actual TLS config that will be used
			TLS: tlsConfig,
		}
	} else {
		logrus.Info("Initializing discovery without TLS")
	}

	// Creates a new store, will ignore options given
	// if not supported by the chosen store
	s.store, err = libkv.NewStore(s.backend, addrs, config)
	return err
}

func (kvc *KVClient) prepareClient() error {
	hasCA := kvc.certificateAuthority != ""
	hasCert := kvc.clientCertificate != ""
	hasKey := kvc.clientKey != ""

	config := &store.Config{
		ConnectionTimeout: 5 * time.Second,
	}
	if hasCA || hasCert || hasKey {

		var cacert *x509.CertPool
		if kvc.certificateAuthority != "" {
			capem, err := ioutil.ReadFile(kvc.certificateAuthority)
			if err != nil {
				return err
			}
			cacert = x509.NewCertPool()
			if !cacert.AppendCertsFromPEM(capem) {
				return errors.New("unable to load certificate authority")
			}
		}

		var cert tls.Certificate
		if kvc.clientCertificate != "" && kvc.clientKey != "" {
			c := kvc.clientCertificate
			k := kvc.clientKey
			var err error
			cert, err = tls.LoadX509KeyPair(c, k)
			if err != nil {
				return err
			}
		}

		config.ClientTLS = &store.ClientTLSConfig{
			CertFile:   kvc.clientCertificate,
			KeyFile:    kvc.clientKey,
			CACertFile: kvc.certificateAuthority,
		}
		config.TLS = &tls.Config{
			RootCAs:      cacert,
			Certificates: []tls.Certificate{cert},
		}

	}
	store, err := libkv.NewStore(kvc.backend, kvc.addresses, config)
	if err != nil {
		fmt.Println(err)
		logrus.Error("unable to create kvclient. ", err)
		return err
	}
	kvc.store = store
	return nil
}

func New(config *config.BalancerConfig) (Store, error) {
	u, err := url.Parse(config.StoreAddress)
	if err != nil {
		return nil, errors.Wrap(err, "error paring store address")
	}

	scheme := u.Scheme
	if scheme != "consul" && scheme != "etcd" {
		return nil, ErrUnsupportedStore
	}

	//Validating open connection
	_, err = net.Dial("tcp", u.Host)
	if err != nil {
		return nil, errors.Wrap(err, "Store connection failed. Make sure your store is up and running.")
	}

	kv, err := libkv.NewStore(
		kv.Backend(scheme),
		[]string{u.Host},
		nil,
	)
	if err != nil {
		kv.Close()
		return nil, errors.Wrap(err, "Cannot create store consul")
	}

	svcsChs := []chan []types.Service{}
	dstsChs := []chan []types.Destination{}
	checksChs := []chan []types.CheckSpec{}

	validate := validator.New()
	// Registering custom validations
	validate.RegisterValidation("protocols", validateValues(types.Protocols))
	validate.RegisterValidation("schedulers", validateValues(types.Schedulers))

	fusisStore := &FusisStore{
		kv:                  kv,
		prefix:              config.StorePrefix,
		validate:            validate,
		servicesChannels:    svcsChs,
		destinationChannels: dstsChs,
		checksChannels:      checksChs,
	}

	go fusisStore.WatchServices()
	go fusisStore.WatchDestinations()
	go fusisStore.WatchChecks()

	return fusisStore, nil
}

// New creates a new GDStore
func New() *GDStore {
	//TODO: Make this configurable
	address := "localhost:8500"

	log.WithFields(log.Fields{"type": "consul", "consul.config": address}).Debug("Creating new store")
	s, err := libkv.NewStore(store.CONSUL, []string{address}, &store.Config{ConnectionTimeout: 10 * time.Second})
	if err != nil {
		log.WithField("error", err).Fatal("Failed to create store")
	}

	log.Info("Created new store using Consul")

	return &GDStore{s}
}

func (provider *KvProvider) provide(configurationChan chan<- configMessage) error {
	switch provider.StoreType {
	case store.CONSUL:
		consul.Register()
	case store.ETCD:
		etcd.Register()
	case store.ZK:
		zookeeper.Register()
	case store.BOLTDB:
		boltdb.Register()
	default:
		return errors.New("Invalid kv store: " + string(provider.StoreType))
	}
	kv, err := libkv.NewStore(
		provider.StoreType,
		[]string{provider.Endpoint},
		&store.Config{
			ConnectionTimeout: 30 * time.Second,
			Bucket:            "traefik",
		},
	)
	if err != nil {
		return err
	}
	if _, err := kv.List(""); err != nil {
		return err
	}
	provider.kvclient = kv
	if provider.Watch {
		stopCh := make(chan struct{})
		chanKeys, err := kv.WatchTree(provider.Prefix, stopCh)
		if err != nil {
			return err
		}
		go func() {
			for {
				<-chanKeys
				configuration := provider.loadConfig()
				if configuration != nil {
					configurationChan <- configMessage{string(provider.StoreType), configuration}
				}
				defer close(stopCh)
			}
		}()
	}
	configuration := provider.loadConfig()
	configurationChan <- configMessage{string(provider.StoreType), configuration}
	return nil
}