// Writes a key to the /reality tree to signify that the pod specified by the UUID has been
// launched on the given node.
func (c *consulStore) WriteRealityIndex(podKey types.PodUniqueKey, node types.NodeName) error {
	if podKey == "" {
		return util.Errorf("Pod store can only write index for pods with uuid keys")
	}

	realityIndexPath := computeRealityIndexPath(podKey, node)

	// Now, write the secondary index to /intent/<node>/<key>
	index := PodIndex{
		PodKey: podKey,
	}

	indexBytes, err := json.Marshal(index)
	if err != nil {
		return util.Errorf("Could not marshal index as json: %s", err)
	}

	indexPair := &api.KVPair{
		Key:   realityIndexPath,
		Value: indexBytes,
	}
	_, err = c.consulKV.Put(indexPair, nil)
	if err != nil {
		return consulutil.NewKVError("put", realityIndexPath, err)
	}

	return nil
}

// Polls for either farm to get be populated by a daemon set with the same
// node selector as the daemon set in the argument
func waitForMutateSelectorFarms(firstFarm *Farm, secondFarm *Farm, ds ds_fields.DaemonSet) error {
	condition := func() error {
		if anotherDS, ok := firstFarm.children[ds.ID]; ok {
			if ds.ID != anotherDS.ds.ID() ||
				ds.NodeSelector.String() != anotherDS.ds.GetNodeSelector().String() {
				return util.Errorf(
					"Daemon sets do not match, expected '%v', '%v', got '%v', '%v'",
					ds.ID,
					ds.NodeSelector.String(),
					anotherDS.ds.ID(),
					anotherDS.ds.GetNodeSelector().String(),
				)
			}
			return nil
		} else if anotherDS, ok := secondFarm.children[ds.ID]; ok {
			if ds.ID != anotherDS.ds.ID() ||
				ds.NodeSelector.String() != anotherDS.ds.GetNodeSelector().String() {
				return util.Errorf(
					"Daemon sets do not match, expected '%v', '%v', got '%v', '%v'",
					ds.ID,
					ds.NodeSelector.String(),
					anotherDS.ds.ID(),
					anotherDS.ds.GetNodeSelector().String(),
				)
			}
			return nil
		}
		return util.Errorf("Farm does not have daemon set id")
	}
	return waitForCondition(condition)
}

func parseNodeSelectorWithPrompt(
	oldSelector klabels.Selector,
	newSelectorString string,
	applicator labels.Applicator,
) (klabels.Selector, error) {
	newSelector, err := parseNodeSelector(newSelectorString)
	if err != nil {
		return newSelector, err
	}
	if oldSelector.String() == newSelector.String() {
		return newSelector, nil
	}

	newNodeLabels, err := applicator.GetMatches(newSelector, labels.NODE, false)
	if err != nil {
		return newSelector, util.Errorf("Error getting matching labels: %v", err)
	}

	oldNodeLabels, err := applicator.GetMatches(oldSelector, labels.NODE, false)
	if err != nil {
		return newSelector, util.Errorf("Error getting matching labels: %v", err)
	}

	toRemove, toAdd := makeNodeChanges(oldNodeLabels, newNodeLabels)

	fmt.Printf("Changing deployment from '%v' to '%v':\n", oldSelector.String(), newSelectorString)
	fmt.Printf("Removing:%9s hosts %s\n", fmt.Sprintf("-%v", len(toRemove)), toRemove)
	fmt.Printf("Adding:  %9s hosts %s\n", fmt.Sprintf("+%v", len(toAdd)), toAdd)
	fmt.Println("Continue?")
	if !confirm() {
		return newSelector, util.Errorf("User cancelled")
	}

	return newSelector, nil
}

func getArtifactVerifier(preparerConfig *PreparerConfig, logger *logging.Logger) (auth.ArtifactVerifier, error) {
	var verif ManifestVerification
	var err error
	switch t, _ := preparerConfig.ArtifactAuth["type"].(string); t {
	case "", auth.VerifyNone:
		return auth.NopVerifier(), nil
	case auth.VerifyManifest:
		err = castYaml(preparerConfig.ArtifactAuth, &verif)
		if err != nil {
			return nil, util.Errorf("error configuring artifact verification: %v", err)
		}
		return auth.NewBuildManifestVerifier(verif.KeyringPath, uri.DefaultFetcher, logger)
	case auth.VerifyBuild:
		err = castYaml(preparerConfig.ArtifactAuth, &verif)
		if err != nil {
			return nil, util.Errorf("error configuring artifact verification: %v", err)
		}
		return auth.NewBuildVerifier(verif.KeyringPath, uri.DefaultFetcher, logger)
	case auth.VerifyEither:
		err = castYaml(preparerConfig.ArtifactAuth, &verif)
		if err != nil {
			return nil, util.Errorf("error configuring artifact verification: %v", err)
		}
		return auth.NewCompositeVerifier(verif.KeyringPath, uri.DefaultFetcher, logger)
	default:
		return nil, util.Errorf("Unrecognized artifact verification type: %v", t)
	}
}

// Attempts to claim a lock. If the overrideLock is set, any existing lock holder
// will be destroyed and one more attempt will be made to acquire the lock
func (r Replicator) lock(lock kp.Lock, lockPath string, overrideLock bool) error {
	err := lock.Lock(lockPath)

	if _, ok := err.(kp.AlreadyLockedError); ok {
		holder, id, err := r.Store.LockHolder(lockPath)
		if err != nil {
			return util.Errorf("Lock already held for %q, could not determine holder due to error: %s", lockPath, err)
		} else if holder == "" {
			// we failed to acquire this lock, but there is no outstanding
			// holder
			// this indicates that the previous holder had a LockDelay,
			// which prevents other parties from acquiring the lock for a
			// limited time
			return util.Errorf("Lock for %q is blocked due to delay by previous holder", lockPath)
		} else if overrideLock {
			err = r.Store.DestroyLockHolder(id)
			if err != nil {
				return util.Errorf("Unable to destroy the current lock holder (%s) for %q: %s", holder, lockPath, err)
			}

			// try acquiring the lock again, but this time don't destroy holders so we don't try forever
			return r.lock(lock, lockPath, false)

		} else {
			return util.Errorf("Lock for %q already held by lock %q", lockPath, holder)
		}
	}

	return err
}

// Given a consul key path, returns the RC ID and the lock type. Returns an err
// if the key does not resemble an RC lock key
func (s *consulStore) lockTypeFromKey(key string) (fields.ID, LockType, error) {
	keyParts := strings.Split(key, "/")
	// Sanity check key structure e.g. /lock/replication_controllers/abcd-1234
	if len(keyParts) < 3 || len(keyParts) > 4 {
		return "", UnknownLockType, util.Errorf("Key '%s' does not resemble an RC lock", key)
	}

	if keyParts[0] != consulutil.LOCK_TREE {
		return "", UnknownLockType, util.Errorf("Key '%s' does not resemble an RC lock", key)
	}

	if keyParts[1] != rcTree {
		return "", UnknownLockType, util.Errorf("Key '%s' does not resemble an RC lock", key)
	}

	rcID := keyParts[2]
	if len(keyParts) == 3 {
		// There's no lock suffix, so this is an ownership lock
		return fields.ID(rcID), OwnershipLockType, nil
	}

	switch keyParts[3] {
	case mutationSuffix:
		return fields.ID(rcID), MutationLockType, nil
	case updateCreationSuffix:
		return fields.ID(rcID), UpdateCreationLockType, nil
	default:
		return fields.ID(rcID), UnknownLockType, nil
	}
}

// UnmarshalConfig reads the preparer's configuration from its bytes.
func UnmarshalConfig(config []byte) (*PreparerConfig, error) {
	appConfig := AppConfig{}
	err := yaml.Unmarshal(config, &appConfig)
	preparerConfig := appConfig.P2PreparerConfig
	if err != nil {
		return nil, util.Errorf("The config file %s was malformatted - %s", config, err)
	}

	if preparerConfig.NodeName == "" {
		hostname, err := os.Hostname()
		if err != nil {
			return nil, util.Errorf("Couldn't determine hostname: %s", err)
		}

		preparerConfig.NodeName = types.NodeName(hostname)
	}
	if preparerConfig.ConsulAddress == "" {
		preparerConfig.ConsulAddress = DefaultConsulAddress
	}
	if preparerConfig.HooksDirectory == "" {
		preparerConfig.HooksDirectory = hooks.DEFAULT_PATH
	}
	if preparerConfig.PodRoot == "" {
		preparerConfig.PodRoot = pods.DefaultPath
	}
	return &preparerConfig, nil

}

// addPods schedules pods for all unscheduled nodes selected by ds.nodeSelector
func (ds *daemonSet) addPods() error {
	podLocations, err := ds.CurrentPods()
	if err != nil {
		return util.Errorf("Error retrieving pod locations from daemon set: %v", err)
	}
	currentNodes := podLocations.Nodes()

	eligible, err := ds.EligibleNodes()
	if err != nil {
		return util.Errorf("Error retrieving eligible nodes for daemon set: %v", err)
	}
	// TODO: Grab a lock here for the pod_id before adding something to check
	// contention and then disable

	// Get the difference in nodes that we need to schedule on and then sort them
	// for deterministic ordering
	toScheduleSorted := types.NewNodeSet(eligible...).Difference(types.NewNodeSet(currentNodes...)).ListNodes()
	ds.logger.NoFields().Infof("Need to label %d nodes", len(toScheduleSorted))

	for _, node := range toScheduleSorted {
		err := ds.labelPod(node)
		if err != nil {
			return util.Errorf("Error labeling node: %v", err)
		}
	}

	ds.logger.Infof("Need to schedule %v nodes", len(currentNodes))
	if len(currentNodes) > 0 {
		return ds.PublishToReplication()
	}

	return nil
}

// removePods unschedules pods for all scheduled nodes not selected
// by ds.nodeSelector
func (ds *daemonSet) removePods() error {
	podLocations, err := ds.CurrentPods()
	if err != nil {
		return util.Errorf("Error retrieving pod locations from daemon set: %v", err)
	}
	currentNodes := podLocations.Nodes()

	eligible, err := ds.EligibleNodes()
	if err != nil {
		return util.Errorf("Error retrieving eligible nodes for daemon set: %v", err)
	}

	// Get the difference in nodes that we need to unschedule on and then sort them
	// for deterministic ordering
	toUnscheduleSorted := types.NewNodeSet(currentNodes...).Difference(types.NewNodeSet(eligible...)).ListNodes()
	ds.logger.NoFields().Infof("Need to unschedule %d nodes", len(toUnscheduleSorted))

	ds.cancelReplication()

	for _, node := range toUnscheduleSorted {
		err := ds.unschedule(node)
		if err != nil {
			return util.Errorf("Error unscheduling node: %v", err)
		}
	}

	ds.logger.Infof("Need to schedule %v nodes", len(currentNodes))
	if len(currentNodes)-len(toUnscheduleSorted) > 0 {
		return ds.PublishToReplication()
	}

	return nil
}

func (p UserPolicy) AuthorizePod(podUser string, manifest Signed, logger logging.Logger) error {
	// Verify that the signature is valid
	plaintext, signature := manifest.SignatureData()
	if signature == nil {
		return Error{util.Errorf("received unsigned manifest"), nil}
	}
	keyringChan := p.keyringWatcher.GetAsync()
	dpolChan := p.deployWatcher.GetAsync()
	keyring := (<-keyringChan).(openpgp.EntityList)
	dpol := (<-dpolChan).(DeployPol)

	signer, err := checkDetachedSignature(keyring, plaintext, signature)
	if err != nil {
		return err
	}

	// Check if any of the signer's identites is authorized
	lastIdName := "(unknown)"
	for name, id := range signer.Identities {
		if dpol.Authorized(podUser, id.UserId.Email) {
			return nil
		}
		lastIdName = name
	}
	return Error{util.Errorf("user not authorized to deploy app: %s", lastIdName), nil}
}

// Watch for changes to nodes and sends update and delete signals
func (ds *daemonSet) handleNodeChanges(changes *labels.LabeledChanges) error {
	if len(changes.Updated) > 0 {
		ds.logger.NoFields().Infof("Received node change signal")
		err := ds.removePods()
		if err != nil {
			return util.Errorf("Unable to remove pods from intent tree: %v", err)
		}
		err = ds.addPods()
		if err != nil {
			return util.Errorf("Unable to add pods to intent tree: %v", err)
		}
		return nil
	}

	if len(changes.Created) > 0 {
		ds.logger.NoFields().Infof("Received node create signal")
		err := ds.addPods()
		if err != nil {
			return util.Errorf("Unable to add pods to intent tree: %v", err)
		}
	}

	if len(changes.Deleted) > 0 {
		ds.logger.NoFields().Infof("Received node delete signal")
		err := ds.removePods()
		if err != nil {
			return util.Errorf("Unable to remove pods from intent tree: %v", err)
		}
	}

	return nil
}

func LoadKeyring(path string) (openpgp.EntityList, error) {
	if path == "" {
		return nil, util.Errorf("no keyring configured")
	}
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close()

	// Accept both ASCII-armored and binary encodings
	keyring, err := openpgp.ReadArmoredKeyRing(f)
	if err != nil && err.Error() == "openpgp: invalid argument: no armored data found" {
		offset, seekErr := f.Seek(0, os.SEEK_SET)
		if offset != 0 || seekErr != nil {
			return nil, util.Errorf(
				"couldn't seek to beginning, got %d %s",
				offset,
				seekErr,
			)
		}
		keyring, err = openpgp.ReadKeyRing(f)
	}

	return keyring, err
}

func (p FixedKeyringPolicy) AuthorizeApp(manifest Manifest, logger logging.Logger) error {
	plaintext, signature := manifest.SignatureData()
	if signature == nil {
		return Error{util.Errorf("received unsigned manifest (expected signature)"), nil}
	}
	signer, err := checkDetachedSignature(p.Keyring, plaintext, signature)
	if err != nil {
		return err
	}

	signerId := fmt.Sprintf("%X", signer.PrimaryKey.Fingerprint)
	logger.WithField("signer_key", signerId).Debugln("resolved manifest signature")

	// Check authorization for this package to be deployed by this
	// key, if configured.
	if len(p.AuthorizedDeployers[manifest.ID()]) > 0 {
		found := false
		for _, deployerId := range p.AuthorizedDeployers[manifest.ID()] {
			if deployerId == signerId {
				found = true
				break
			}
		}
		if !found {
			return Error{
				util.Errorf("manifest signer not authorized to deploy " + manifest.ID()),
				map[string]interface{}{"signer_key": signerId},
			}
		}
	}

	return nil
}

func (c *consulStore) ReadPod(podKey types.PodUniqueKey) (Pod, error) {
	if podKey == "" {
		return Pod{}, util.Errorf("Pod store can only read pods with uuid keys")
	}

	if pod, ok := c.fetchFromCache(podKey); ok {
		return pod, nil
	}

	podPath := computePodPath(podKey)

	pair, _, err := c.consulKV.Get(podPath, nil)
	if err != nil {
		return Pod{}, consulutil.NewKVError("get", podPath, err)
	}

	if pair == nil {
		return Pod{}, NoPodError(podKey)
	}

	var pod Pod
	err = json.Unmarshal(pair.Value, &pod)
	if err != nil {
		return Pod{}, util.Errorf("Could not unmarshal pod '%s' as json: %s", podKey, err)
	}

	c.addToCache(podKey, pod)

	return pod, nil
}

func (e EnvironmentExtractor) constructFinishFromEnvironment(exitCode int, exitStatus int) (FinishOutput, error) {
	podID := os.Getenv(pods.PodIDEnvVar)
	if podID == "" {
		return FinishOutput{}, util.Errorf("No %s env var set", pods.PodIDEnvVar)
	}

	launchableID := os.Getenv(pods.LaunchableIDEnvVar)
	if launchableID == "" {
		return FinishOutput{}, util.Errorf("No %s env var set", pods.LaunchableIDEnvVar)
	}

	entryPoint := os.Getenv(launch.EntryPointEnvVar)
	if entryPoint == "" {
		return FinishOutput{}, util.Errorf("No %s env var set", launch.EntryPointEnvVar)
	}

	// It's okay if this one is missing, most pods are "legacy" pods that have a blank unique key
	podUniqueKey := os.Getenv(pods.PodUniqueKeyEnvVar)

	return FinishOutput{
		PodID:        types.PodID(podID),
		LaunchableID: launch.LaunchableID(launchableID),
		EntryPoint:   entryPoint,
		PodUniqueKey: types.PodUniqueKey(podUniqueKey),
		ExitCode:     exitCode,
		ExitStatus:   exitStatus,
	}, nil
}

func (s *consulStore) Get(id fields.ID) (fields.DaemonSet, *api.QueryMeta, error) {
	var metadata *api.QueryMeta
	dsPath, err := s.dsPath(id)
	if err != nil {
		return fields.DaemonSet{}, metadata, util.Errorf("Error getting daemon set path: %v", err)
	}

	kvp, metadata, err := s.kv.Get(dsPath, nil)
	if err != nil {
		return fields.DaemonSet{}, metadata, consulutil.NewKVError("get", dsPath, err)
	}
	if metadata == nil {
		// no metadata returned
		return fields.DaemonSet{}, metadata, errors.New("No metadata found")
	}
	if kvp == nil {
		// ID didn't exist
		return fields.DaemonSet{}, metadata, NoDaemonSet
	}

	ds, err := kvpToDS(kvp)
	if err != nil {
		return fields.DaemonSet{}, metadata, util.Errorf("Error translating kvp to daemon set: %v", err)
	}
	return ds, metadata, nil
}

func verifyConsulUp(timeout string) error {
	timeoutDur, err := time.ParseDuration(timeout)
	if err != nil {
		return err
	}
	if timeoutDur == 0 {
		return nil
	}

	config := api.DefaultConfig()
	config.Token = *consulToken
	client, err := api.NewClient(config)
	if err != nil {
		return util.Errorf("Could not construct consul client: '%s'", err)
	}
	consulIsUp := make(chan struct{})
	go func() {
		for {
			time.Sleep(200 * time.Millisecond)
			err := Ping(client)
			if err == nil {
				consulIsUp <- struct{}{}
				return
			}
		}
	}()
	select {
	case <-time.After(timeoutDur):
		return util.Errorf("Consul did not start or was not available after %v", timeoutDur)
	case <-consulIsUp:
		return nil
	}
}

func (s *consulStore) Create(
	manifest manifest.Manifest,
	minHealth int,
	name fields.ClusterName,
	nodeSelector klabels.Selector,
	podID types.PodID,
	timeout time.Duration,
) (fields.DaemonSet, error) {
	if err := checkManifestPodID(podID, manifest); err != nil {
		return fields.DaemonSet{}, util.Errorf("Error verifying manifest pod id: %v", err)
	}

	ds, err := s.innerCreate(manifest, minHealth, name, nodeSelector, podID, timeout)
	// TODO: measure whether retries are is important in practice
	for i := 0; i < s.retries; i++ {
		if _, ok := err.(CASError); ok {
			ds, err = s.innerCreate(manifest, minHealth, name, nodeSelector, podID, timeout)
		} else {
			break
		}
	}
	if err != nil {
		return fields.DaemonSet{}, util.Errorf("Error creating daemon set: %v", err)
	}
	return ds, nil
}

// symlink the runit service directory into the actual directory being monitored
// by runsvdir
// runsvdir will automatically start a service for each new directory (unless a
// down file exists)
func (s *ServiceBuilder) activate(templates map[string]ServiceTemplate) error {
	for serviceName := range templates {
		linkPath := filepath.Join(s.RunitRoot, serviceName)
		stageDir := filepath.Join(s.StagingRoot, serviceName)

		info, err := os.Lstat(linkPath)
		if err == nil {
			// if it exists, make sure it is actually a symlink
			if info.Mode()&os.ModeSymlink == 0 {
				return util.Errorf("%s is not a symlink", linkPath)
			}

			// and that it points to the right place
			target, err := os.Readlink(linkPath)
			if err != nil {
				return err
			}
			if target != stageDir {
				return util.Errorf("%s is a symlink to %s (expected %s)", linkPath, target, stageDir)
			}
		} else if os.IsNotExist(err) {
			if err = os.Symlink(stageDir, linkPath); err != nil {
				return err
			}
		} else if err != nil {
			return err
		}
	}
	return nil
}

func (a registry) authDataFromRegistryResponse(registryResponse RegistryResponse) (auth.VerificationData, error) {
	verificationData := auth.VerificationData{}
	if registryResponse.ManifestLocation != "" {
		manifestURL, err := url.Parse(registryResponse.ManifestLocation)
		if err != nil {
			return verificationData, util.Errorf("Couldn't parse manifest URL from registry response: %s", err)
		}
		verificationData.ManifestLocation = manifestURL
	}

	if registryResponse.ManifestSignatureLocation != "" {
		manifestSignatureURL, err := url.Parse(registryResponse.ManifestSignatureLocation)
		if err != nil {
			return verificationData, util.Errorf("Couldn't parse manifest signature URL from registry response: %s", err)
		}
		verificationData.ManifestSignatureLocation = manifestSignatureURL
	}

	if registryResponse.BuildSignatureLocation != "" {
		buildSignatureURL, err := url.Parse(registryResponse.BuildSignatureLocation)
		if err != nil {
			return verificationData, util.Errorf("Couldn't parse build signature URL from registry response: %s", err)
		}
		verificationData.BuildSignatureLocation = buildSignatureURL
	}

	return verificationData, nil
}