// Set takes the object and commits it to the database.
func (c *Client) Set(obj db.Entity) error {
	if err := obj.Validate(); err != nil {
		return err
	}

	if obj.Hooks().PreSet != nil {
		if err := obj.Hooks().PreSet(c, obj); err != nil {
			return errors.EtcdToErrored(err)
		}
	}

	content, err := jsonio.Write(obj)
	if err != nil {
		return err
	}

	path, err := obj.Path()
	if err != nil {
		return err
	}

	if _, err := c.client.Set(context.Background(), c.qualified(path), string(content), nil); err != nil {
		return errors.EtcdToErrored(err)
	}

	if obj.Hooks().PostSet != nil {
		if err := obj.Hooks().PostSet(c, obj); err != nil {
			return errors.EtcdToErrored(err)
		}
	}

	return nil
}

// Get retrieves the item from etcd's key/value store and then populates obj with its data.
func (c *Client) Get(obj db.Entity) error {
	if obj.Hooks().PreGet != nil {
		if err := obj.Hooks().PreGet(c, obj); err != nil {
			return errors.EtcdToErrored(err)
		}
	}

	path, err := obj.Path()
	if err != nil {
		return err
	}

	resp, err := c.client.Get(context.Background(), c.qualified(path), nil)
	if err != nil {
		return errors.EtcdToErrored(err)
	}

	if err := jsonio.Read(obj, []byte(resp.Node.Value)); err != nil {
		return err
	}

	if err := obj.SetKey(c.trimPath(resp.Node.Key)); err != nil {
		return err
	}

	if obj.Hooks().PostGet != nil {
		if err := obj.Hooks().PostGet(c, obj); err != nil {
			return errors.EtcdToErrored(err)
		}
	}

	return obj.Validate()
}

// GetVolume returns the Volume for a given volume.
func (c *Client) GetVolume(policy, name string) (*Volume, error) {
	// FIXME make this take a single string and not a split one
	resp, err := c.etcdClient.Get(context.Background(), c.volume(policy, name, "create"), nil)
	if err != nil {
		return nil, errors.EtcdToErrored(err)
	}

	ret := &Volume{}

	if err := json.Unmarshal([]byte(resp.Node.Value), ret); err != nil {
		return nil, err
	}

	if err := ret.Validate(); err != nil {
		return nil, err
	}

	runtime, err := c.GetVolumeRuntime(policy, name)
	if err != nil {
		return nil, err
	}

	ret.RuntimeOptions = runtime

	return ret, nil
}

// PublishUseWithTTL pushes the use to etcd, with a TTL that expires the record
// if it has not been updated within that time.
func (c *Client) PublishUseWithTTL(ut UseLocker, ttl time.Duration) error {
	content, err := json.Marshal(ut)
	if err != nil {
		return err
	}

	if ttl < 0 {
		err := errored.Errorf("TTL was less than 0 for locker %#v!!!! This should not happen!", ut)
		logrus.Error(err)
		return err
	}

	logrus.Debugf("Publishing use with TTL %v: %#v", ttl, ut)
	value := string(content)

	// attempt to set the lock. If the lock cannot be set and it is is empty, attempt to set it now.
	_, err = c.etcdClient.Set(context.Background(), c.use(ut.Type(), ut.GetVolume()), string(content), &client.SetOptions{TTL: ttl, PrevValue: value})
	if err != nil {
		if er, ok := errors.EtcdToErrored(err).(*errored.Error); ok && er.Contains(errors.NotExists) {
			_, err := c.etcdClient.Set(context.Background(), c.use(ut.Type(), ut.GetVolume()), string(content), &client.SetOptions{TTL: ttl, PrevExist: client.PrevNoExist})
			if err != nil {
				return errors.PublishMount.Combine(err)
			}
		} else {
			return errors.PublishMount.Combine(err)
		}
	}

	return nil
}

// PublishPolicy publishes policy intent to the configuration store.
func (c *Client) PublishPolicy(name string, cfg *Policy) error {
	cfg.Name = name

	if err := cfg.Validate(); err != nil {
		return err
	}

	value, err := json.Marshal(cfg)
	if err != nil {
		return err
	}

	// NOTE: The creation of the policy revision entry and the actual publishing of the policy
	//       should be wrapped in a transaction so they either both succeed or both fail, but
	//       etcd2 doesn't support transactions (etcd3 does/will).
	//
	//       For now, we create the revision entry first and then publish the policy.  It's
	//       better to have an entry for a policy revision that was never actually published
	//       than to have a policy published which has no revision recorded for it.
	if err := c.CreatePolicyRevision(name, string(value)); err != nil {
		return err
	}

	// create the volume directory for the policy so that files can be written there.
	// for example: /volplugin/policies/policy1 will create
	// /volplugin/volumes/policy1 so that a volume of policy1/test can be created
	// at /volplugin/volumes/policy1/test
	c.etcdClient.Set(context.Background(), c.prefixed(rootVolume, name), "", &client.SetOptions{Dir: true})

	if _, err := c.etcdClient.Set(context.Background(), c.policy(name), string(value), &client.SetOptions{PrevExist: client.PrevIgnore}); err != nil {
		return errors.EtcdToErrored(err)
	}

	return nil
}

// Delete removes the object from the store.
func (c *Client) Delete(obj db.Entity) error {
	return helpers.WrapDelete(c, obj, func(path string) error {
		if _, err := c.client.Delete(context.Background(), c.qualified(path), nil); err != nil {
			return errors.EtcdToErrored(err)
		}

		return nil
	})
}

// PublishUse pushes the use to etcd.
func (c *Client) PublishUse(ut UseLocker) error {
	content, err := json.Marshal(ut)
	if err != nil {
		return err
	}

	_, err = c.etcdClient.Set(context.Background(), c.use(ut.Type(), ut.GetVolume()), string(content), &client.SetOptions{PrevExist: client.PrevNoExist})
	if _, ok := err.(client.Error); ok && err.(client.Error).Code == client.ErrorCodeNodeExist {
		if ut.MayExist() {
			_, err := c.etcdClient.Set(context.Background(), c.use(ut.Type(), ut.GetVolume()), string(content), &client.SetOptions{PrevExist: client.PrevExist, PrevValue: string(content)})
			return errors.EtcdToErrored(err)
		}
		return errors.Exists.Combine(err)
	}

	logrus.Debugf("Publishing use: (error: %v) %#v", err, ut)
	return errors.EtcdToErrored(err)
}

// GetPolicyRevision returns a single revision for a given policy.
func (c *Client) GetPolicyRevision(name, revision string) (string, error) {
	keyspace := c.policyArchiveEntry(name, revision)

	resp, err := c.etcdClient.Get(context.Background(), keyspace, &client.GetOptions{Sort: false, Recursive: false, Quorum: true})
	if err != nil {
		return "", errors.EtcdToErrored(err)
	}

	return resp.Node.Value, nil
}

// GetVolumeRuntime retrieves only the runtime parameters for the volume.
func (c *Client) GetVolumeRuntime(policy, name string) (RuntimeOptions, error) {
	runtime := RuntimeOptions{}

	resp, err := c.etcdClient.Get(context.Background(), c.volume(policy, name, "runtime"), nil)
	if err != nil {
		return runtime, errors.EtcdToErrored(err)
	}

	return runtime, json.Unmarshal([]byte(resp.Node.Value), &runtime)
}

// Get retrieves the item from etcd's key/value store and then populates obj with its data.
func (c *Client) Get(obj db.Entity) error {
	return helpers.WrapGet(c, obj, func(path string) (string, []byte, error) {
		resp, err := c.client.Get(context.Background(), c.qualified(path), nil)
		if err != nil {
			return "", nil, errors.EtcdToErrored(err)
		}

		return resp.Node.Key, []byte(resp.Node.Value), nil
	})
}

// CreatePolicyRevision creates an revision entry in a policy's history.
func (c *Client) CreatePolicyRevision(name string, policy string) error {
	timestamp := fmt.Sprint(time.Now().Unix())
	key := c.policyArchiveEntry(name, timestamp)

	_, err := c.etcdClient.Set(context.Background(), key, policy, nil)
	if err != nil {
		return errors.EtcdToErrored(err)
	}

	return nil
}

// ListVolumes returns a map of volume name -> Volume.
func (c *Client) ListVolumes(policy string) (map[string]*Volume, error) {
	policyPath := c.prefixed(rootVolume, policy)

	resp, err := c.etcdClient.Get(context.Background(), policyPath, &client.GetOptions{Recursive: true, Sort: true})
	if err != nil {
		return nil, errors.EtcdToErrored(err)
	}

	configs := map[string]*Volume{}

	for _, node := range resp.Node.Nodes {
		if len(node.Nodes) > 0 {
			node = node.Nodes[0]
			key := strings.TrimPrefix(node.Key, policyPath)
			if !node.Dir && strings.HasSuffix(node.Key, "/create") {
				key = strings.TrimSuffix(key, "/create")

				config, ok := configs[key[1:]]
				if !ok {
					config = new(Volume)
				}

				if err := json.Unmarshal([]byte(node.Value), config); err != nil {
					return nil, err
				}
				// trim leading slash
				configs[key[1:]] = config
			}

			if !node.Dir && strings.HasSuffix(node.Key, "/runtime") {
				key = strings.TrimSuffix(key, "/create")

				config, ok := configs[key[1:]]
				if !ok {
					config = new(Volume)
				}

				if err := json.Unmarshal([]byte(node.Value), &config.RuntimeOptions); err != nil {
					return nil, err
				}
				// trim leading slash
				configs[key[1:]] = config
			}
		}
	}

	for _, config := range configs {
		if _, err := config.CreateOptions.ActualSize(); err != nil {
			return nil, err
		}
	}

	return configs, nil
}

// ListAllVolumes returns an array with all the named policies and volumes the
// apiserver knows about. Volumes have syntax: policy/volumeName which will be
// reflected in the returned string.
func (c *Client) ListAllVolumes() ([]string, error) {
	resp, err := c.etcdClient.Get(context.Background(), c.prefixed(rootVolume), &client.GetOptions{Recursive: true, Sort: true})
	if err != nil {
		if er, ok := errors.EtcdToErrored(err).(*errored.Error); ok && er.Contains(errors.NotExists) {
			return []string{}, nil
		}

		return nil, errors.EtcdToErrored(err)
	}

	ret := []string{}

	for _, node := range resp.Node.Nodes {
		for _, innerNode := range node.Nodes {
			ret = append(ret, path.Join(path.Base(node.Key), path.Base(innerNode.Key)))
		}
	}

	return ret, nil
}

// GetUse retrieves the UseMount for the given volume name.
func (c *Client) GetUse(ut UseLocker, vc *Volume) error {
	resp, err := c.etcdClient.Get(context.Background(), c.use(ut.Type(), vc.String()), nil)
	if err != nil {
		return errors.EtcdToErrored(err)
	}

	if err := json.Unmarshal([]byte(resp.Node.Value), ut); err != nil {
		return err
	}

	return nil
}

// PublishGlobal publishes the global configuration.
func (tlc *Client) PublishGlobal(g *Global) error {
	gcPath := tlc.prefixed("global-config")

	value, err := json.Marshal(g.Canonical())
	if err != nil {
		return err
	}

	if _, err := tlc.etcdClient.Set(context.Background(), gcPath, string(value), &client.SetOptions{PrevExist: client.PrevIgnore}); err != nil {
		return errors.EtcdToErrored(err)
	}

	return nil
}

// Delete removes the object from the store.
func (c *Client) Delete(obj db.Entity) error {
	if obj.Hooks().PreDelete != nil {
		if err := obj.Hooks().PreDelete(c, obj); err != nil {
			return errors.EtcdToErrored(err)
		}
	}

	path, err := obj.Path()
	if err != nil {
		return err
	}

	if _, err := c.client.Delete(context.Background(), c.qualified(path), nil); err != nil {
		return errors.EtcdToErrored(err)
	}

	if obj.Hooks().PostDelete != nil {
		if err := obj.Hooks().PostDelete(c, obj); err != nil {
			return errors.EtcdToErrored(err)
		}
	}

	return nil
}

// RemoveUse will remove a user from etcd. Does not fail if the user does
// not exist.
func (c *Client) RemoveUse(ut UseLocker, force bool) error {
	content, err := json.Marshal(ut)
	if err != nil {
		return err
	}

	logrus.Debugf("Removing Use Lock: %#v", ut)

	opts := &client.DeleteOptions{PrevValue: string(content)}
	if force {
		opts = nil
	}

	_, err = c.etcdClient.Delete(context.Background(), c.use(ut.Type(), ut.GetVolume()), opts)
	return errors.EtcdToErrored(err)
}

// PublishVolumeRuntime publishes the runtime parameters for each volume.
func (c *Client) PublishVolumeRuntime(vo *Volume, ro RuntimeOptions) error {
	if err := ro.ValidateJSON(); err != nil {
		return errors.ErrJSONValidation.Combine(err)
	}

	content, err := json.Marshal(ro)
	if err != nil {
		return err
	}

	c.etcdClient.Set(context.Background(), c.prefixed(rootVolume, vo.PolicyName, vo.VolumeName), "", &client.SetOptions{Dir: true})
	if _, err := c.etcdClient.Set(context.Background(), c.volume(vo.PolicyName, vo.VolumeName, "runtime"), string(content), nil); err != nil {
		return errors.EtcdToErrored(err)
	}

	return nil
}

// CurrentSchemaVersion returns the version of the last migration which was successfully run.
// If no previous migrations have been run, the schema version is 0.
// If there's any error besides "key doesn't exist", execution is aborted.
func (e *Engine) CurrentSchemaVersion() int64 {
	resp, err := e.etcdClient.Get(context.Background(), path.Join(e.prefix, backend.SchemaVersionKey), nil)
	if err != nil {
		if err := errors.EtcdToErrored(err); err != nil && err == errors.NotExists {
			return 0 // no key = schema version 0
		}

		logrus.Fatalf("Unexpected error when looking up schema version: %v\n", err)
	}

	i, err := strconv.Atoi(resp.Node.Value)
	if err != nil {
		logrus.Fatalf("Got back unexpected schema version data: %v\n", resp.Node.Value)
	}

	return int64(i)
}

// ListPolicies provides an array of strings corresponding to the name of each
// policy.
func (c *Client) ListPolicies() ([]Policy, error) {
	resp, err := c.etcdClient.Get(context.Background(), c.prefixed(rootPolicy), &client.GetOptions{Recursive: true, Sort: true})
	if err != nil {
		return nil, errors.EtcdToErrored(err)
	}

	policies := []Policy{}
	for _, node := range resp.Node.Nodes {
		policy := Policy{}
		if err := json.Unmarshal([]byte(node.Value), &policy); err != nil {
			return nil, err
		}
		policies = append(policies, policy)
	}

	return policies, nil
}