// Login on Etcd
func EtcdLogin(utente client.User) string {

	c := GetEtcdAuthClient(utente)
	kapi := client.NewKeysAPI(c)

	// Options for loop Node of a cluster

	_, err := kapi.Get(context.Background(), "/gru/", &client.GetOptions{
		Recursive: true,
		Sort:      true,
		Quorum:    false,
	})

	if err != nil {
		log.Printf("Errore: %v", err)
		if strings.Contains(err.Error(), ERRORE_SERVER_DOWN) {
			return ERRORE_SERVER_DOWN
		}
		if strings.Contains(err.Error(), ERRORE_LOGIN) {
			return ERRORE_CREDENZIALI
		}
		if !strings.Contains(err.Error(), KEY_NOT_FOUND_LOGIN) {
			return err.Error()
		}
	}

	return ""
}

/*
Monitoring changes in etcd server.
It designed for run in separate goroutine.
*/
func etcdMon(etcdRootPath string, config client.Config, bus chan fileChangeEvent, startIndex uint64) {
	c, err := client.New(config)
	if err != nil {
		panic(err)
	}
	kapi := client.NewKeysAPI(c)
	var nextEvent uint64 = startIndex
	for {
		response, err := kapi.Watcher(etcdRootPath, &client.WatcherOptions{AfterIndex: nextEvent, Recursive: true}).Next(context.Background())
		if err != nil {
			log.Println(err)
			time.Sleep(time.Second)
			continue
		}
		nextEvent = response.Index
		if response.Action == "delete" {
			bus <- fileChangeEvent{Path: response.Node.Key, IsRemoved: true, IsDir: response.Node.Dir}
			continue
		}
		if response.Node.Dir {
			bus <- fileChangeEvent{Path: response.Node.Key, IsDir: response.Node.Dir}
			continue
		}
		bus <- fileChangeEvent{Path: response.Node.Key, Content: []byte(response.Node.Value)}
	}
}

func initEtcd() {
	key := "/" + etcdKey

	c, err := etcdConnect()
	if err != nil {
		log.Fatalf("Can't connect to etcd: %v", err)
	}

	keysAPI := client.NewKeysAPI(c)

	gopts := client.GetOptions{Recursive: false, Sort: false, Quorum: true}
	r, err := keysAPI.Get(context.Background(), key, &gopts)
	if err != nil {
		switch err := err.(type) {
		case client.Error:
			if err.Code == 100 {
				// Not found - create it
				sopts := client.SetOptions{Dir: true}
				_, err := keysAPI.Set(context.Background(), key, "", &sopts)
				if err != nil {
					log.Fatalf("Error creating etcdKey dir: %v", err)
				}
				return
			}
		default:
			log.Fatalf("etcd error: %v", err)
		}
	}
	if !r.Node.Dir {
		log.Fatalf("Error: etcdKey %q is not a directory", key)
	}

}

func NewEtcdBackend(address string) (Backend, error) {
	if address == "" {
		address = "http://127.0.0.1:2379/vault"
	}
	url, err := url.Parse(address)
	if err != nil {
		return nil, maskAny(err)
	}
	path := url.Path

	// Ensure path is prefixed.
	if !strings.HasPrefix(path, "/") {
		path = "/" + path
	}

	url.Path = ""
	endpoint := url.String()

	c, err := client.New(client.Config{
		Endpoints: []string{endpoint},
	})
	if err != nil {
		return nil, err
	}
	kAPI := client.NewKeysAPI(c)

	return &etcdBackend{
		path: path,
		kAPI: kAPI,
	}, nil
}

// Add adds a given frontend record with given ID to the list of frontends.
// If the given ID already exists, a DuplicateIDError is returned.
func (eb *etcdBackend) Add(id string, record api.FrontendRecord) error {
	if err := validateID(id); err != nil {
		return maskAny(err)
	}
	if err := record.Validate(); err != nil {
		return maskAny(err)
	}
	etcdPath := path.Join(eb.prefix, frontEndPrefix, id)
	kAPI := client.NewKeysAPI(eb.client)
	options := &client.SetOptions{
		PrevExist: client.PrevNoExist,
	}
	rawJSON, err := json.Marshal(record)
	if err != nil {
		return maskAny(err)
	}
	if _, err := kAPI.Set(context.Background(), etcdPath, string(rawJSON), options); isEtcdError(err, client.ErrorCodeNodeExist) {
		return maskAny(errgo.WithCausef(nil, api.DuplicateIDError, "Duplicate ID '%s'", id))
	} else if err != nil {
		eb.Logger.Warningf("ETCD error in Add: %#v", err)
		return maskAny(err)
	}

	return nil
}

func stress(mb int) error {
	time.Sleep(5 * time.Second)

	cfg := client.Config{
		Endpoints: []string{"http://localhost:12379", "http://localhost:22379", "http://localhost:32379"},
		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 {
		return err
	}
	kapi := client.NewKeysAPI(c)

	for i := 0; i < mb*2; i++ {
		fmt.Println("stressing", i)
		k := make([]byte, 100)
		binary.PutVarint(k, int64(rand.Intn(putSize)))
		_, err = kapi.Set(context.Background(), string(k), "", nil)
		if err != nil {
			if i < 2 {
				return err
			}
		}
		time.Sleep(500 * time.Millisecond)
	}

	return nil
}

func TestLeaderLeaseSwapWhileWaiting(t *testing.T) {
	testutil.RequireEtcd(t)
	defer testutil.DumpEtcdOnFailure(t)
	c, err := testutil.MakeNewEtcdClient()
	if err != nil {
		t.Fatal(err)
	}
	client := etcdclient.NewKeysAPI(c)
	key := "/random/key"

	if _, err := client.Set(context.Background(), key, "holder", &etcdclient.SetOptions{TTL: 10 * time.Second, PrevExist: etcdclient.PrevNoExist}); err != nil {
		t.Fatal(err)
	}

	go func() {
		time.Sleep(time.Second)
		if _, err := client.Set(context.Background(), key, "other", &etcdclient.SetOptions{TTL: 10 * time.Second}); err != nil {
			t.Fatal(err)
		}
		glog.Infof("Changed key ownership")
	}()

	lease := leaderlease.NewEtcd(c, key, "other", 10)
	ch := make(chan error, 1)
	go lease.AcquireAndHold(ch)

	<-ch
	glog.Infof("Lease acquired")
	lease.Release()
	if err, ok := <-ch; err == nil || !ok || !strings.Contains(err.Error(), "the lease has been lost") {
		t.Errorf("Expected error and open channel when lease was swapped: %v %t", err, ok)
	}
	<-ch
	glog.Infof("Lease gone")
}

func main() {
	logger := log.New(os.Stderr, "server", log.LstdFlags)
	if err := serverFlags.Parse(os.Args[1:]); err != nil {
		logger.Fatalf("%v", err)
	}

	client, err := etcd.New(etcd.Config{Endpoints: *etcdEndpoints})
	if err != nil {
		logger.Fatalf("Failed setting up etcd %v", err)
	}
	w, err := watcher.New(serverFlags, etcd.NewKeysAPI(client), *etcdFlagzPath, logger)
	if err != nil {
		logger.Fatalf("Failed setting up watcher %v", err)
	}
	err = w.Initialize()
	if err != nil {
		logger.Fatalf("Failed initializing watcher %v", err)
	}
	w.Start()
	logger.Printf("etcd flag value watching initialized")

	flagzEndpoint := flagz.NewStatusEndpoint(serverFlags)
	http.HandleFunc("/debug/flagz", flagzEndpoint.ListFlags)
	http.HandleFunc("/", handleDefaultPage)

	addr := fmt.Sprintf("%s:%d", *listenHost, *listenPort)
	logger.Printf("Serving at: %v", addr)
	if err := http.ListenAndServe(addr, http.DefaultServeMux); err != nil {
		logger.Fatalf("Failed serving: %v", err)
	}
	logger.Printf("Done, bye.")
}

// NewEtcdConfig creates a new service discovery backend for etcd
func NewEtcdConfig(config map[string]interface{}) Etcd {
	etcd := Etcd{
		Prefix: "/containerbuddy",
	}
	etcdConfig := client.Config{}
	switch endpoints := config["endpoints"].(type) {
	case string:
		etcdConfig.Endpoints = []string{endpoints}
	case []string:
		etcdConfig.Endpoints = endpoints
	default:
		log.Fatal("Must provide etcd endpoints")
	}

	prefix, ok := config["prefix"].(string)
	if ok {
		etcd.Prefix = prefix
	}

	etcdClient, err := client.New(etcdConfig)
	if err != nil {
		log.Fatal(err)
	}
	etcd.Client = etcdClient
	etcd.API = client.NewKeysAPI(etcdClient)
	return etcd
}

// GetV2 gets the value to the key using V2 API.
func (c *Cluster) GetV2(w io.Writer, key []byte) error {
	endpoints := []string{}
	for _, nd := range c.NameToNode {
		for v := range nd.Flags.ListenClientURLs {
			endpoints = append(endpoints, v)
		}
	}

	cfg := client.Config{
		Endpoints:               endpoints,
		Transport:               client.DefaultTransport,
		HeaderTimeoutPerRequest: time.Second,
		// SelectionMode:           client.EndpointSelectionPrioritizeLeader,
	}
	ct, err := client.New(cfg)
	if err != nil {
		return err
	}
	kapi := client.NewKeysAPI(ct)

	ts := time.Now()
	resp, err := kapi.Get(context.Background(), string(key), nil)
	if err != nil {
		return err
	}

	fmt.Fprintf(w, "[GetV2] Done! Took %v for %s/%s.\n", time.Since(ts), key, resp.Node.Value)
	return nil
}

func NewKeysAPI(cfg etcd.Config) (etcd.KeysAPI, error) {
	eCli, err := etcd.New(cfg)
	if err != nil {
		return nil, err
	}
	return etcd.NewKeysAPI(eCli), nil
}

func main() {
	myFlagSet.Parse(os.Args[1:])
	logger := log.New(os.Stderr, "wr ", log.LstdFlags)

	client, err := etcd.New(etcd.Config{Endpoints: []string{"http://localhost:2379"}})
	if err != nil {
		logger.Fatalf("Failed setting up %v", err)
	}
	w, err := watcher.New(myFlagSet, etcd.NewKeysAPI(client), "/example/flagz", logger)
	if err != nil {
		logger.Fatalf("Failed setting up %v", err)
	}
	err = w.Initialize()
	if err != nil {
		logger.Fatalf("Failed setting up %v", err)
	}
	w.Start()

	for true {
		logger.Printf("staticint: %v dynint: %v dynstring: %v",
			*staticInt,
			dynInt.Get(),
			dynStr.Get())
		time.Sleep(1500 * time.Millisecond)
	}
}

func NewNode(stop chan bool) *Node {
	n := new(Node)
	n.stop = stop
	n.stopBoard = make(chan bool, 2)
	n.stopThread = make(chan bool, 2)
	n.stopPost = make(chan bool, 2)
	n.stopFile = make(chan bool, 2)
	n.Stats = NewNodeStats()
	n.Config = parseFlags()
	n.Storage = fourchan.NewStorage(n.Config.CassKeyspace, n.Config.CassEndpoints...)
	cfg := etcd.Config{
		Endpoints:               n.Config.EtcdEndpoints,
		Transport:               etcd.DefaultTransport,
		HeaderTimeoutPerRequest: 3 * time.Second,
	}
	c, err := etcd.New(cfg)
	if err != nil {
		log.Fatal("Failed to connected to etcd: ", err)
	}
	n.Keys = etcd.NewKeysAPI(c)
	n.Closed = false
	// TODO these chan sizes are rather arbitrary...
	n.CBoard = make(chan *fourchan.Board, numBoardRoutines)
	n.CThread = make(chan *fourchan.Thread, numThreadRoutines)
	n.CPost = make(chan *fourchan.Post, numPostRoutines)
	n.CFile = make(chan *fourchan.File, numFileRoutines)
	n.Files = make(map[int]string)

	return n
}

func UpdateConfig(c client.Client, cluster, confType, serviceName, confFile string) (string, string) {

	kapi := client.NewKeysAPI(c)

	key := ""
	successMessage := ""
	switch confType {
	case "agent":
		key = "/gru/" + cluster + "/config"
		successMessage = ADD_AGENT
	case "service":
		key = "/gru/" + cluster + "/services/" + serviceName
		successMessage = ADD_SERVICE
	case "policy":
		key = "/gru/" + cluster + "/policy"
		successMessage = ADD_POLICY
	case "analytic":
		analyticsName := serviceName
		key = "/gru/" + cluster + "/analytics/" + analyticsName
		successMessage = ADD_ANALYTICS
	default:
		log.Fatal("Unrecognized configuration type")
	}

	resp, err := kapi.Update(context.Background(), key, confFile)
	if err != nil {
		log.Printf(err.Error())
		return "", err.Error()
	}
	// print common key info
	log.Printf("Set is done. Metadata is %q\n", resp)

	return successMessage, ""
}

// watchBackends monitors the registrator namespace for
// changes affecting services which are monitored.  It
// calls update if there are changes detected.
func watchBackends(ctx context.Context) {
	k := client.NewKeysAPI(etcd)
	watcher := k.Watcher(registratorNamespace, &client.WatcherOptions{
		Recursive: true,
	})

	// Process changes
	for {
		resp, err := watcher.Next(ctx)
		if err != nil {
			fmt.Println("Error watching backends", err.Error())
			continue
		}

		// Ignore read-only events
		if resp.Action == "get" {
			continue
		}

		// Find the service of the modified key
		serviceName := serviceFromRegistratorKey(resp.Node.Key)
		if serviceName == "" {
			continue
		}

		// Update only if this service is one we are watching
		for _, s := range services {
			if serviceName == s.Name {
				Update()
				break
			}
		}
	}
}

func Conn() {
	cfg := client.Config{
		Endpoints:               []string{"http://127.0.0.1:2379/", "http://127.0.0.1:4001"},
		Transport:               client.DefaultTransport,
		HeaderTimeoutPerRequest: time.Second,
	}

	c, err := client.New(cfg)

	if err != nil {
		log.Fatal(err)
	}

	kapi := client.NewKeysAPI(c)

	log.Print("Setting '/foo' key with 'bar' value")
	resp, err := kapi.Set(context.Background(), "/foo", "bar", nil)
	if err != nil {
		log.Fatal(err)
	} else {
		log.Printf("Set is done.Metadata is %q\n", resp)
	}

	log.Print("Getting '/foo' key value")
	resp, err = kapi.Get(context.Background(), "/foo", nil)
	if err != nil {
		log.Fatal(err)
	} else {
		log.Printf("Get is done,Metadata is %q\n", resp)
		log.Printf("%q key has %q value\n", resp.Node.Key, resp.Node.Value)
	}

}

func (s *Store) New(c config.Config) *Store {
	// Open an etcd client
	var endpoints []string
	endpoints = append(endpoints, fmt.Sprintf("http://%v:%v", c.Etcd.Hostname, c.Etcd.Port))

	ec := client.Config{
		Endpoints: endpoints,
		Transport: client.DefaultTransport,
	}

	e, err := client.New(ec)
	if err != nil {
		log.Fatalln("Could not connect to etcd:", err)
	}

	// Create a new KeysAPI for etcd
	k := client.NewKeysAPI(e)

	ns := &Store{
		e:        e,
		c:        c,
		k:        k,
		basePath: fmt.Sprint(c.Etcd.BasePath, "/vips/"),
	}

	return ns
}

func TestBasic(t *testing.T) {

	cfg := client.Config{
		Endpoints:               []string{"http://127.0.0.1:4001"},
		Transport:               client.DefaultTransport,
		HeaderTimeoutPerRequest: time.Second,
	}
	c, err := client.New(cfg)

	if err != nil {
		log.Panicln(err)
	}

	kapi := client.NewKeysAPI(c)

	client := EtcdReigistryClient{
		EtcdRegistryConfig{
			ServiceName:  "test",
			InstanceName: "test1",
			BaseURL:      "127.0.0.1:8080",
		},
		kapi,
	}

	client.Register()
	response, _ := client.ServicesByName("test")
	if len(response) == 0 {
		t.Error("No service registered")
	}
	client.Unregister()
	response, _ = client.ServicesByName("test")
	if len(response) != 0 {
		t.Error("Service not  unregistered")
	}
}

func getFleetRegistryClient(fleetEndpoints []string) (fleetClient.API, error) {
	var dial func(string, string) (net.Conn, error)

	tlsConfig, err := fleetPkg.ReadTLSConfigFiles("", "", "")
	if err != nil {
		return nil, err
	}

	trans := &http.Transport{
		Dial:            dial,
		TLSClientConfig: tlsConfig,
	}

	timeout := 3 * time.Second
	eCfg := etcd.Config{
		Endpoints: fleetEndpoints,
		Transport: trans,
	}
	eClient, err := etcd.New(eCfg)
	if err != nil {
		return nil, err
	}
	kAPI := etcd.NewKeysAPI(eClient)
	reg := registry.NewEtcdRegistry(kAPI, registry.DefaultKeyPrefix, timeout)

	return &fleetClient.RegistryClient{Registry: reg}, nil
}

func TestKeepAlive(t *testing.T) {

	cfg := client.Config{
		Endpoints:               []string{"http://127.0.0.1:2379"},
		Transport:               client.DefaultTransport,
		HeaderTimeoutPerRequest: time.Second,
	}
	c, err := client.New(cfg)

	if err != nil {
		log.Panicln(err)
	}

	kapi := client.NewKeysAPI(c)

	client := EtcdReigistryClient{
		EtcdRegistryConfig{
			ServiceName:  "test",
			InstanceName: "test1",
			BaseURL:      "127.0.0.1:8080",
		},
		kapi,
	}

	client.Register()

	time.Sleep(50 * time.Second)

	response, _ := client.ServicesByName("test")
	log.Println(response)

	// Ahoj
}