// DEPRECATED: use one of the Create* functions instead
func (s consulStore) Put(u roll_fields.Update) error {
	b, err := json.Marshal(u)
	if err != nil {
		return err
	}

	key, err := RollPath(u.ID())
	if err != nil {
		return err
	}

	success, _, err := s.kv.CAS(&api.KVPair{
		Key:   key,
		Value: b,
		// it must not already exist
		ModifyIndex: 0,
	}, nil)
	if err != nil {
		return consulutil.NewKVError("cas", key, err)
	}
	if !success {
		return fmt.Errorf("update with new RC ID %s already exists", u.NewRC)
	}
	return nil
}

// Attempts to create a rolling update. Checks sessionErrCh for session renewal
// errors just before actually doing the creation to minimize the likelihood of
// race conditions resulting in conflicting RUs
func (s consulStore) attemptRUCreation(u roll_fields.Update, rollLabels klabels.Set, sessionErrCh chan error) (createdRU roll_fields.Update, err error) {
	// If we create an RU, we also want to create its labels. If the second step
	// fails, we want to best-effort remove the RU
	var ruCleanup func()
	defer func() {
		if err != nil && ruCleanup != nil {
			ruCleanup()
		}
	}()

	b, err := json.Marshal(u)
	if err != nil {
		return u, err
	}

	key, err := RollPath(roll_fields.ID(u.NewRC))
	if err != nil {
		return u, err
	}

	// Confirm that our lock session is still valid, and then create the
	// rolling update. If session isn't valid, delete the newRC we just
	// created
	select {
	case err := <-sessionErrCh:
		if err == nil {
			err = util.Errorf("Cannot create ru because session was destroyed")
		}
		return u, err
	default:
		success, _, err := s.kv.CAS(&api.KVPair{
			Key:   key,
			Value: b,
		}, nil)
		if err != nil {
			return u, consulutil.NewKVError("cas", key, err)
		}

		// Shouldn't be possible if our session is still valid, preventing other insertions
		if !success {
			return u, util.Errorf("update with new RC ID %s already exists", u.NewRC)
		}

		ruCleanup = func() {
			err := s.Delete(u.ID())
			if err != nil {
				s.logger.WithError(err).Errorln("Unable to cleanup RU %s after failed labeling attempt", u.ID())
			}
		}
	}

	err = s.labeler.SetLabels(labels.RU, u.ID().String(), rollLabels)
	if err != nil {
		return roll_fields.Update{}, err
	}

	return u, nil
}

func TestCreateRollingUpdateFromExistingRCs(t *testing.T) {
	rollstore := newRollStore(t, nil)

	newRCID := rc_fields.ID("new_rc")
	oldRCID := rc_fields.ID("old_rc")

	update := fields.Update{
		NewRC: newRCID,
		OldRC: oldRCID,
	}

	newRCLabels := klabels.Set(map[string]string{
		"some_key": "some_val",
	})

	u, err := rollstore.CreateRollingUpdateFromExistingRCs(update, newRCLabels, newRCLabels)
	if err != nil {
		t.Fatalf("Unexpected error creating update: %s", err)
	}

	storedUpdate, err := rollstore.Get(update.ID())
	if err != nil {
		t.Fatalf("Unable to retrieve value put in roll store: %s", err)
	}

	if storedUpdate.NewRC != newRCID {
		t.Errorf("Stored update didn't have expected new rc value: wanted '%s' but got '%s'", newRCID, storedUpdate.NewRC)
	}

	if storedUpdate.OldRC != oldRCID {
		t.Errorf("Stored update didn't have expected old rc value: wanted '%s' but got '%s'", oldRCID, storedUpdate.OldRC)
	}

	rcLabels, err := rollstore.labeler.GetLabels(labels.RC, newRCID.String())
	if err != nil {
		t.Fatalf("Unable to fetch labels for newly created new RC: %s", err)
	}

	if rcLabels.Labels["some_key"] != "some_val" {
		t.Errorf("Expected labels to be set on new RC")
	}

	ruLabels, err := rollstore.labeler.GetLabels(labels.RU, u.ID().String())
	if err != nil {
		t.Fatalf("Unable to fetch labels for newly created new RU: %s", err)
	}

	if ruLabels.Labels["some_key"] != "some_val" {
		t.Errorf("Expected labels to be set on new RU")
	}
}

// Validates that the rolling update is capable of being processed. If not, an
// error is returned.
// The following conditions make an RU invalid:
// 1) New RC does not exist
// 2) Old RC does not exist
func (rlf *Farm) validateRoll(update roll_fields.Update, logger logging.Logger) error {
	_, err := rlf.rcs.Get(update.NewRC)
	if err == rcstore.NoReplicationController {
		return fmt.Errorf("RU '%s' is invalid, new RC '%s' did not exist", update.ID(), update.NewRC)
	} else if err != nil {
		// There was a potentially transient consul error, we don't necessarily want to delete the RU
		logger.WithError(err).Errorln("Could not fetch new RC to validate RU, assuming it's valid")
	}

	_, err = rlf.rcs.Get(update.OldRC)
	if err == rcstore.NoReplicationController {
		return fmt.Errorf("RU '%s' is invalid, old RC '%s' did not exist", update.ID(), update.OldRC)
	} else if err != nil {
		// There was a potentially transient consul error, we don't necessarily want to delete the RU
		logger.WithError(err).Errorln("Could not fetch old RC in order to validate RU, assuming it's valid")
	}

	return nil
}

// Test that if a conflicting update exists, a new one will not be admitted
func TestCreateExistingRCsMutualExclusion(t *testing.T) {
	newRCID := rc_fields.ID("new_rc")
	oldRCID := rc_fields.ID("old_rc")

	conflictingEntry := fields.Update{
		OldRC: newRCID,
		NewRC: rc_fields.ID("some_other_rc"),
	}

	rollstore := newRollStore(t, []fields.Update{conflictingEntry})

	update := fields.Update{
		NewRC: newRCID,
		OldRC: oldRCID,
	}

	_, err := rollstore.CreateRollingUpdateFromExistingRCs(update, nil, nil)
	if err == nil {
		t.Fatal("Expected update creation to fail due to conflict")
	}

	if conflictingErr, ok := err.(*ConflictingRUError); !ok {
		t.Error("Returned error didn't have ConflictingRUError type")
	} else {
		if conflictingErr.ConflictingID != conflictingEntry.ID() {
			t.Errorf("Expected error to have conflicting ID of '%s', was '%s'", conflictingEntry.ID(), conflictingErr.ConflictingID)
		}

		if conflictingErr.ConflictingRCID != conflictingEntry.OldRC {
			t.Errorf("Expected error to have conflicting rc ID of '%s', was '%s'", conflictingEntry.OldRC, conflictingErr.ConflictingRCID)
		}
	}

	ru, _ := rollstore.Get(fields.ID(update.NewRC))
	if ru.NewRC != "" || ru.OldRC != "" {
		t.Fatal("New ru shouldn't have been created but it was")
	}
}