// NewAgent returns an initalized Agent.
func NewAgent(prefix string, c *consul.Client, d *docker.Client) *Agent {
	return &Agent{
		KV:     c.KV(),
		Docker: d,
		Prefix: prefix,
	}
}

func getAttributes(keys []string, overwriteAttributes map[string]interface{}) (map[string]interface{}, error) {
	var attributes map[string]interface{}
	var c *api.Client = util.Consul()
	attributes = make(map[string]interface{})

	//	Get attributes from consul KVS
	for _, key := range keys {
		list, _, err := c.KV().List(key, &api.QueryOptions{})
		if err != nil {
			return nil, err
		}

		for _, kv := range list {
			var a map[string]interface{}
			if err := json.Unmarshal(kv.Value, &a); err != nil {
				return nil, err
			}
			if err := mergo.MergeWithOverwrite(&attributes, a); err != nil {
				return nil, errors.New(fmt.Sprintf("Failed to merge attributes(%v)", err))
			}
		}
	}

	//	Overwrite some attributes by specified parameter in task.yml
	if err := mergo.MergeWithOverwrite(&attributes, overwriteAttributes); err != nil {
		return nil, err
	}
	return attributes, nil
}

func NewConsulClient(config *consulAPI.Config) (KVClient, error) {
	var (
		c   *consulAPI.Client
		err error
	)
	if config != nil {
		c, err = consulAPI.NewClient(config)
	} else {
		c, err = consulAPI.NewClient(consulAPI.DefaultConfig())
	}
	if err != nil {
		return nil, err
	}
	maxRetries := 30
	i := 0
	for {
		leader, err := c.Status().Leader()
		if err != nil || leader == "" {
			log.Info("Waiting for consul client to be ready...")
			time.Sleep(2 * time.Second)
			i++
			if i > maxRetries {
				e := fmt.Errorf("Unable to contact consul: %s", err)
				log.Error(e)
				return nil, e
			}
		} else {
			log.Info("Consul client ready")
			break
		}
	}
	return &ConsulClient{c}, nil
}

// setupSemaphore is used to setup a new Semaphore given the API client, key
// prefix, session name, and slot holder limit. If oneshot is true then we will
// set up for a single attempt at acquisition, using the given wait time. The
// retry parameter sets how many 500 errors the lock monitor will tolerate
// before giving up the semaphore.
func (c *LockCommand) setupSemaphore(client *api.Client, limit int, prefix, name string,
	oneshot bool, wait time.Duration, retry int) (*LockUnlock, error) {
	if c.verbose {
		c.Ui.Info(fmt.Sprintf("Setting up semaphore (limit %d) at prefix: %s", limit, prefix))
	}
	opts := api.SemaphoreOptions{
		Prefix:           prefix,
		Limit:            limit,
		SessionName:      name,
		MonitorRetries:   retry,
		MonitorRetryTime: defaultMonitorRetryTime,
	}
	if oneshot {
		opts.SemaphoreTryOnce = true
		opts.SemaphoreWaitTime = wait
	}
	s, err := client.SemaphoreOpts(&opts)
	if err != nil {
		return nil, err
	}
	lu := &LockUnlock{
		lockFn:    s.Acquire,
		unlockFn:  s.Release,
		cleanupFn: s.Destroy,
		inUseErr:  api.ErrSemaphoreInUse,
		rawOpts:   &opts,
	}
	return lu, nil
}

// setupLock is used to setup a new Lock given the API client, the key prefix to
// operate on, and an optional session name. If oneshot is true then we will set
// up for a single attempt at acquisition, using the given wait time. The retry
// parameter sets how many 500 errors the lock monitor will tolerate before
// giving up the lock.
func (c *LockCommand) setupLock(client *api.Client, prefix, name string,
	oneshot bool, wait time.Duration, retry int) (*LockUnlock, error) {
	// Use the DefaultSemaphoreKey extension, this way if a lock and
	// semaphore are both used at the same prefix, we will get a conflict
	// which we can report to the user.
	key := path.Join(prefix, api.DefaultSemaphoreKey)
	if c.verbose {
		c.Ui.Info(fmt.Sprintf("Setting up lock at path: %s", key))
	}
	opts := api.LockOptions{
		Key:              key,
		SessionName:      name,
		MonitorRetries:   retry,
		MonitorRetryTime: defaultMonitorRetryTime,
	}
	if oneshot {
		opts.LockTryOnce = true
		opts.LockWaitTime = wait
	}
	l, err := client.LockOpts(&opts)
	if err != nil {
		return nil, err
	}
	lu := &LockUnlock{
		lockFn:    l.Lock,
		unlockFn:  l.Unlock,
		cleanupFn: l.Destroy,
		inUseErr:  api.ErrLockInUse,
		rawOpts:   &opts,
	}
	return lu, nil
}

func writeToConsul(t *testing.T, prefix, key string, client *consulapi.Client) []byte {
	token := os.Getenv("TOKEN")
	dc := os.Getenv("DC")
	if dc == "" {
		dc = "dc1"
	}

	kv := client.KV()

	writeOptions := &consulapi.WriteOptions{Token: token, Datacenter: dc}

	// Delete all keys in the prefixed KV space
	if _, err := kv.DeleteTree(prefix, writeOptions); err != nil {
		t.Fatalf("err: %v", err)
	}

	// Put a test KV
	encodedValue := make([]byte, base64.StdEncoding.EncodedLen(1024))
	base64.StdEncoding.Encode(encodedValue, createRandomBytes(1024))
	p := &consulapi.KVPair{Key: key, Flags: 42, Value: encodedValue}
	if _, err := kv.Put(p, writeOptions); err != nil {
		t.Fatalf("err: %v", err)
	}

	return encodedValue
}

// NewWaiter creates a new Wait entry with a sensible default isReady function if
// nil is provided in its place.
func NewWaiter(client *api.Client, prefix string, minimumNodes int, isReady func(n *WaitNode) bool) *Wait {
	if isReady == nil {
		isReady = func(n *WaitNode) bool {
			return true
		}
	}

	nodeUpdateCh := make(chan WaitNodeUpdate, 2)
	nodeReadyCh := make(chan WaitNode, 2)
	allReadyCh := make(chan []WaitNode, 2)

	return &Wait{
		Prefix:       prefix,
		MinimumNodes: minimumNodes,
		IsReady:      isReady,

		NodeUpdate: nodeUpdateCh,
		NodeReady:  nodeReadyCh,
		AllReady:   allReadyCh,

		nodeUpdate: nodeUpdateCh,
		nodeReady:  nodeReadyCh,
		allReady:   allReadyCh,

		client: client,
		kv:     client.KV(),
	}
}

func NewInstall(consulClient *consul.Client, marathon *marathon.Marathon, mesos *mesos.Mesos, zkHosts []string) (*Install, error) {
	apiConfig = map[string]interface{}{
		"mantl": map[string]interface{}{
			"zookeeper": map[string]interface{}{
				"hosts": strings.Join(zkHosts, ","),
			},
		},
	}

	if mesos != nil {
		mesosAuthRequired, err := mesos.RequiresAuthentication()
		if err != nil {
			return nil, err
		}

		mesosConfig := map[string]interface{}{
			"principal":              mesos.Principal,
			"secret":                 mesos.Secret,
			"secret-path":            mesos.SecretPath,
			"authentication-enabled": mesosAuthRequired,
		}

		mantlConfig := apiConfig["mantl"].(map[string]interface{})
		mantlConfig["mesos"] = mesosConfig
	}

	zookeeper := zookeeper.NewZookeeper(zkHosts)
	return &Install{consulClient, consulClient.KV(), marathon, mesos, zookeeper}, nil
}

// serviceConfig constructs the config for all good instances of a single service.
func serviceConfig(client *api.Client, name string, passing map[string]bool, tagPrefix string) (config []string) {
	if name == "" || len(passing) == 0 {
		return nil
	}

	q := &api.QueryOptions{RequireConsistent: true}
	svcs, _, err := client.Catalog().Service(name, "", q)
	if err != nil {
		log.Printf("[WARN] consul: Error getting catalog service %s. %v", name, err)
		return nil
	}

	for _, svc := range svcs {
		// check if the instance is in the list of instances
		// which passed the health check
		if _, ok := passing[svc.ServiceID]; !ok {
			continue
		}

		for _, tag := range svc.ServiceTags {
			if host, path, ok := parseURLPrefixTag(tag, tagPrefix); ok {
				name, addr, port := svc.ServiceName, svc.ServiceAddress, svc.ServicePort
				if runtime.GOOS == "darwin" && !strings.Contains(addr, ".") {
					addr += ".local"
				}
				config = append(config, fmt.Sprintf("route add %s %s%s http://%s:%d/ tags %q", name, host, path, addr, port, strings.Join(svc.ServiceTags, ",")))
			}
		}
	}
	return config
}

func removeSession(cl *api.Client, id string) {
	session := cl.Session()
	_, err := session.Destroy(id, nil)
	if err != nil {
		log.Printf("Error while destroying session: %v", err)
	}
}

func writeFileToConsul(t *testing.T, prefix, key string, file string, client *consulapi.Client) []byte {
	token := os.Getenv("TOKEN")
	dc := os.Getenv("DC")
	if dc == "" {
		dc = "dc1"
	}

	kv := client.KV()

	writeOptions := &consulapi.WriteOptions{Token: token, Datacenter: dc}

	// Delete all keys in the prefixed KV space
	if _, err := kv.DeleteTree(prefix, writeOptions); err != nil {
		t.Fatalf("err: %v", err)
	}

	fileBytes, err := ioutil.ReadFile(file)
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	p := &consulapi.KVPair{Key: key, Flags: 42, Value: fileBytes}

	if _, err := kv.Put(p, writeOptions); err != nil {
		t.Fatalf("err: %v", err)
	}
	return fileBytes
}

// getDC is used to get the datacenter of the local agent
func getDC(client *consulapi.Client) (string, error) {
	info, err := client.Agent().Self()
	if err != nil {
		return "", fmt.Errorf("Failed to get datacenter from Consul agent: %v", err)
	}
	dc := info["Config"]["Datacenter"].(string)
	return dc, nil
}

func putKV(client *api.Client, key, value string, index uint64) (bool, error) {
	p := &api.KVPair{Key: key[1:], Value: []byte(value), ModifyIndex: index}
	ok, _, err := client.KV().CAS(p, nil)
	if err != nil {
		return false, err
	}
	return ok, nil
}

/*
   Get full tree under a key and optionally unmarshal.

    Args:
        client : Consul client
        key    : Key to query for.
        output : Unmarshal data to this interface{} if non-nil
*/
func GetKVTree(client *api.Client, key string, output interface{}) (pairs api.KVPairs, err error) {
	kv := client.KV()
	pairs, _, err = kv.List(key, nil)
	if output != nil {
		err = Unmarshal(pairs, output)
	}
	return
}

func (q *QueryCommand) Query(client *api.Client, tag string) []*api.CatalogService {
	services, _, err := client.Catalog().Service(q.service, tag, &api.QueryOptions{AllowStale: true, RequireConsistent: false})
	if err != nil {
		panic(err)
		return nil
	}

	return services
}

// Ping confirms that Consul can be reached and it has a leader. If the return
// is nil, then consul should be ready to accept requests.
//
// If the return is non-nil, this typically indicates that either Consul is
// unreachable (eg the agent is not listening on the target port) or has not
// found a leader (in which case Consul returns a 500 to all endpoints, except
// the status types).
//
// If a cluster is starting for the first time, it may report a leader just
// before beginning raft replication, thus rejecting requests made at that
// exact moment.
func Ping(client *api.Client) error {
	_, qm, err := client.Catalog().Nodes(&api.QueryOptions{RequireConsistent: true})
	if err != nil {
		return consulutil.NewKVError("ping", "/catalog/nodes", err)
	}
	if qm == nil || !qm.KnownLeader {
		return util.Errorf("No known leader")
	}
	return nil
}

// Set sets a key's value inside Consul.
func Set(c *consul.Client, key, value string) bool {
	p := &consul.KVPair{Key: key, Value: []byte(value)}
	kv := c.KV()
	Log(fmt.Sprintf("key='%s' value='%s'", key, value), "info")
	_, err := kv.Put(p, nil)
	if err != nil {
		panic(err)
	}
	return true
}

// consulDel removes a key from the Consul KV store.
func consulDel(c *consul.Client, key string) (bool, error) {
	kv := c.KV()
	key = strings.TrimPrefix(key, "/")
	_, err := kv.Delete(key, nil)
	if err != nil {
		return false, err
	}
	Log(fmt.Sprintf("action='consulDel' key='%s'", key), "info")
	return true, err
}

// getDC is used to get the datacenter of the local agent
func getDC(d *schema.ResourceData, client *consulapi.Client) (string, error) {
	if v, ok := d.GetOk("datacenter"); ok {
		return v.(string), nil
	}
	info, err := client.Agent().Self()
	if err != nil {
		return "", fmt.Errorf("Failed to get datacenter from Consul agent: %v", err)
	}
	return info["Config"]["Datacenter"].(string), nil
}

// serviceConfig constructs the config for all good instances of a single service.
func serviceConfig(client *api.Client, name string, passing map[string]bool, tagPrefix string) (config []string) {
	if name == "" || len(passing) == 0 {
		return nil
	}

	dc, err := datacenter(client)
	if err != nil {
		log.Printf("[WARN] consul: Error getting datacenter. %s", err)
		return nil
	}

	q := &api.QueryOptions{RequireConsistent: true}
	svcs, _, err := client.Catalog().Service(name, "", q)
	if err != nil {
		log.Printf("[WARN] consul: Error getting catalog service %s. %v", name, err)
		return nil
	}

	env := map[string]string{
		"DC": dc,
	}

	for _, svc := range svcs {
		// check if the instance is in the list of instances
		// which passed the health check
		if _, ok := passing[svc.ServiceID]; !ok {
			continue
		}

		for _, tag := range svc.ServiceTags {
			if route, opts, ok := parseURLPrefixTag(tag, tagPrefix, env); ok {
				name, addr, port := svc.ServiceName, svc.ServiceAddress, svc.ServicePort

				// use consul node address if service address is not set
				if addr == "" {
					addr = svc.Address
				}

				// add .local suffix on OSX for simple host names w/o domain
				if runtime.GOOS == "darwin" && !strings.Contains(addr, ".") && !strings.HasSuffix(addr, ".local") {
					addr += ".local"
				}

				addr = net.JoinHostPort(addr, strconv.Itoa(port))

				cfg := fmt.Sprintf("route add %s %s http://%s/ tags %q", name, route, addr, strings.Join(svc.ServiceTags, ","))
				if opts != "" {
					cfg += ` opts "` + opts + `"`
				}
				config = append(config, cfg)
			}
		}
	}
	return config
}